HeartSaVioR opened a new pull request, #58562:
URL: https://github.com/apache/spark/pull/58562
### What changes were proposed in this pull request?
This PR rejects streaming `EXCEPT` queries when the left input is streaming.
It also records the compatibility setting in checkpoint metadata and
restores it when an existing query is restarted, so checkpoints created before
this validation was introduced remain recoverable.
### Why are the changes needed?
Streaming `EXCEPT` with a streaming left input is not supported correctly.
Allowing such a query to
start bypasses the normal checks for the aggregation introduced by the
optimizer rewrite.
For example, `EXCEPT DISTINCT` is rewritten to a left anti join followed by
a distinct aggregation:
```scala
val input = MemoryStream[Int]
val result = input.toDS().except(Seq(2).toDS())
val query = result.writeStream
.outputMode("update")
.foreachBatch { (batch: Dataset[Int], id: Long) =>
println(s"batch $id: ${batch.collect().toSeq}")
}
.start()
input.addData(1)
query.processAllAvailable() // prints: batch 0: Seq(1)
input.addData(1)
query.processAllAvailable() // prints: batch 1: Seq(1)
```
The second batch should not produce another `1`: adding a duplicate
left-side row does not change
the result of `EXCEPT DISTINCT`. However, Update mode reports the grouping
key touched by the
rewritten streaming aggregation, producing duplicate output without any
indication that it replaces
an earlier row. With Append mode, the same query without a watermark is
accepted by the
unsupported-operation checker but later fails because the rewritten
aggregation cannot produce
Append output.
`EXCEPT ALL` is likewise rewritten using union and aggregation. New queries
should therefore fail
analysis, while queries recovering from existing checkpoints need to retain
the compatibility
behavior that was active when their checkpoint was created.
### Does this PR introduce _any_ user-facing change?
Yes. A new streaming query whose left side of `EXCEPT` is streaming now
fails analysis instead of
being allowed to start. Existing queries with a persistent checkpoint can
still recover using the
compatibility setting stored with the checkpoint.
### How was this patch tested?
Added and ran tests in:
- `UnsupportedOperationsSuite` (225 tests passed)
- `OffsetSeqLogSuite` (22 tests passed)
- `StreamingQueryManagerSuite` restart compatibility test
- `SparkConfigBindingPolicySuite` (3 tests passed)
Also ran `git diff --check` and checks for non-ASCII characters and overlong
changed source lines.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
--
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]
