cloud-fan commented on code in PR #58562:
URL: https://github.com/apache/spark/pull/58562#discussion_r3966369239
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryManagerSuite.scala:
##########
@@ -60,6 +60,43 @@ class StreamingQueryManagerSuite extends StreamTest {
}
}
+ test("streaming EXCEPT compatibility is restored before
unsupported-operation checks") {
+ withTempDir { checkpointDir =>
+ val input = MemoryStream[Int]
+ val result = input.toDS().except(Seq(100).toDS())
+ val checkpointLocation = checkpointDir.getCanonicalPath
+
+ def startQuery(): StreamingQuery = result.writeStream
+ .outputMode("update")
+ .foreachBatch { (batch: Dataset[Int], _: Long) =>
+ batch.collect()
+ ()
+ }
+ .option("checkpointLocation", checkpointLocation)
+ .start()
+
+ withSQLConf(SQLConf.ALLOW_EXCEPT_ON_STREAMING_DATAFRAME.key -> "true") {
+ val query = startQuery()
+ try {
+ input.addData(1)
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+ }
+
+ withSQLConf(SQLConf.ALLOW_EXCEPT_ON_STREAMING_DATAFRAME.key -> "false") {
Review Comment:
**Non-blocking (P2):** The integration coverage currently exercises only
persisted `true` overriding a current-session `false`; the separate metadata
test does not prove that this manager restores `false`. Please add the reverse
restart case: create a checkpoint whose metadata stores `false`, restart with
the session flag `true` and a streaming-left EXCEPT, and assert that
unsupported-operation analysis still rejects it. That gives a failure signal if
restoration is hard-coded to `true` or otherwise loses the persisted negative
value.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationsSuite.scala:
##########
@@ -774,13 +774,33 @@ class UnsupportedOperationsSuite extends SparkFunSuite
with SQLHelper {
streamBatchSupported = false,
batchStreamSupported = false)
- // Except: *-stream not supported
+ // Except: streaming input on either side is not supported
Review Comment:
**Nit (P3):** This is true only under the default configuration: the
compatibility-enabled assertions immediately below accept a streaming-left,
batch-right EXCEPT. Please qualify the comment with `by default` or mention the
legacy exception explicitly so the stated support matrix matches the test block.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3702,6 +3702,17 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val ALLOW_EXCEPT_ON_STREAMING_DATAFRAME =
+ buildConf("spark.sql.streaming.allowExceptOnStreamingDataFrame")
+ .internal()
+ .doc("When true, allow EXCEPT operations with a streaming DataFrame on
the left side. " +
Review Comment:
**Nit (P3):** Could we make this grammatical and name the actual
compatibility domain? The right-streaming branch still rejects unconditionally,
so `true` permits only a streaming-left, batch-right EXCEPT. For example: `When
true, allows EXCEPT operations with a streaming DataFrame on the left and a
batch DataFrame on the right.`
--
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]
