viirya commented on code in PR #37368:
URL: https://github.com/apache/spark/pull/37368#discussion_r935082833
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala:
##########
@@ -445,6 +445,44 @@ class DataStreamTableAPISuite extends StreamTest with
BeforeAndAfter {
}
}
+ test("SPARK-39940: refresh table when streaming query writes to the catalog
table via DSv1") {
+ withTable("tbl1", "tbl2") {
+ withTempDir { dir =>
+ val baseTbls = new File(dir, "tables")
+ val tbl1File = new File(baseTbls, "tbl1")
+ val tbl2File = new File(baseTbls, "tbl2")
+ val checkpointLocation = new File(dir, "checkpoint")
+
+ val format = "parquet"
+ Seq((1, 2)).toDF("i", "d")
+ .write.format(format).option("path",
tbl1File.getCanonicalPath).saveAsTable("tbl1")
+
+ val query = spark.readStream.format(format).table("tbl1")
+ .writeStream.format(format)
+ .option("checkpointLocation", checkpointLocation.getCanonicalPath)
+ .option("path", tbl2File.getCanonicalPath)
+ .toTable("tbl2")
+
+ try {
+ query.processAllAvailable()
+ checkAnswer(spark.table("tbl2").sort($"i"), Seq(Row(1, 2)))
+
+ Seq((3, 4)).toDF("i", "d")
+ .write.format(format).option("path", tbl1File.getCanonicalPath)
+ .mode(SaveMode.Append).saveAsTable("tbl1")
+
+ query.processAllAvailable()
+ checkAnswer(spark.table("tbl2").sort($"i"), Seq(Row(1, 2), Row(3,
4)))
+
+ assert(query.exception.isEmpty, s"No exception should happen in
streaming query: " +
Review Comment:
```suggestion
assert(query.exception.isEmpty, "No exception should happen in
streaming query: " +
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
