HeartSaVioR opened a new pull request, #58592:
URL: https://github.com/apache/spark/pull/58592
### What changes were proposed in this pull request?
This PR preserves the metadata-column boundary when deduplication keys are
recomputed at streaming query startup.
Analyzer rules may add a hidden metadata column to a source to resolve a
downstream reference after `dropDuplicates()` or
`dropDuplicatesWithinWatermark()`. The recomputation now excludes such metadata
columns unless the same attribute was already present in the originally
resolved keys.
The change applies to both streaming subplans and static subplans embedded
in a streaming query. It also retains the existing deterministic and legacy
key-order behavior.
### Why are the changes needed?
SPARK-57489 made streaming query startup recompute deduplication keys from
the analyzed child's output. If a query referenced `_metadata` only after
deduplication, later metadata propagation added `_metadata` to that output and
recomputation silently made it a deduplication key.
For example, two files containing the same value should produce one row here:
```scala
spark.readStream.format("text").load(path)
.dropDuplicates()
.select($"value", $"_metadata.file_path")
```
Instead, current master/branch-4.x (including v4.3.0-rc1) produces one row
per source file because `_metadata` incorrectly participates in the state key.
Metadata explicitly selected before deduplication must continue to participate
in the key.
### Does this PR introduce _any_ user-facing change?
Yes. It fixes a correctness regression introduced after Spark 4.2. A
metadata column referenced only downstream of deduplication no longer changes
deduplication semantics. Metadata selected before deduplication remains a key.
Hopefully the offending resolution logic is yet to be released to official
Apache Spark version, so fixing the behavior isn't a behavior/breaking change.
### How was this patch tested?
Added Catalyst tests for metadata-boundary key recomputation and end-to-end
file-stream tests covering:
- `dropDuplicates()`
- `dropDuplicatesWithinWatermark()`
- batch deduplication embedded in a streaming query
- metadata referenced after and selected before deduplication
- deterministic and legacy key-order modes
The following suites pass:
```text
build/sbt 'catalyst/testOnly
org.apache.spark.sql.catalyst.analysis.ResolveDeduplicateSuite'
build/sbt 'sql/testOnly
org.apache.spark.sql.streaming.StreamingDeduplicationWithMetadataSuite'
```
### 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]
