You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "HeartSaVioR (via GitHub)" <gi...@apache.org> on 2026/09/08 06:41:15 UTC
[PR] [SPARK-59315][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark) [spark]
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59315][SQL][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark) [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #58592:
URL: https://github.com/apache/spark/pull/58592#issuecomment-5618803361
Thanks! Merging to master/4.x/4.3.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59315][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark) [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #58592:
URL: https://github.com/apache/spark/pull/58592#issuecomment-5594993279
@huanliwang-db @sunchao
Thanks for the comments - I guess both of you are pointing the same issue; I've addressed it via https://github.com/apache/spark/pull/58592/commits/a7c4130fc75d88ac673f43ab33fdaade10afd72d. Please take another look, thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59315][SQL][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark) [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #58592:
URL: https://github.com/apache/spark/pull/58592#issuecomment-5618804086
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/541347c498598d196a382fc98a294f87e98f6c23
- merged into branch-4.x https://github.com/apache/spark/commit/989e1bd3ae3c488eca5629aefb1c81c5baea5788
- merged into branch-4.3 https://github.com/apache/spark/commit/17cdb0b3630a3745f3bad73add679d4ce14def3a
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59315][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark) [spark]
Posted by "huanliwang-db (via GitHub)" <gi...@apache.org>.
huanliwang-db commented on code in PR #58592:
URL: https://github.com/apache/spark/pull/58592#discussion_r3961824958
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicate.scala:
##########
@@ -89,6 +89,24 @@ object ResolveDeduplicate extends Rule[LogicalPlan] {
}
}
+ /**
+ * Recomputes batch and streaming deduplication keys while preserving which metadata columns were
+ * visible at the original deduplication boundary. Analyzer rules may add metadata columns to the
+ * child later to satisfy downstream references, but those columns must not silently become keys.
+ */
+ def recomputeKeysPreservingMetadataBoundary(
Review Comment:
is it possible because of introducing the `_metadata`, the ordering of the original keys are changed
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59315][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark) [spark]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #58592:
URL: https://github.com/apache/spark/pull/58592#discussion_r3963641545
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicate.scala:
##########
@@ -89,6 +89,24 @@ object ResolveDeduplicate extends Rule[LogicalPlan] {
}
}
+ /**
+ * Recomputes batch and streaming deduplication keys while preserving which metadata columns were
+ * visible at the original deduplication boundary. Analyzer rules may add metadata columns to the
+ * child later to satisfy downstream references, but those columns must not silently become keys.
+ */
+ def recomputeKeysPreservingMetadataBoundary(
+ originalKeys: Seq[Attribute],
+ child: LogicalPlan,
+ spec: DeduplicateSpec,
+ orderDeterministically: Boolean,
+ resolver: Resolver): Seq[Attribute] = {
+ val originalMetadataKeys = AttributeSet(
+ originalKeys.filter(key => MetadataAttribute.isValid(key.metadata)))
+ computeKeys(child, spec, orderDeterministically, resolver).filter { key =>
+ !MetadataAttribute.isValid(key.metadata) || originalMetadataKeys.contains(key)
Review Comment:
[P1] Filter metadata before computing legacy key order
The metadata filter runs after legacy `names.toSet.toSeq` ordering. With columns `d,c,b,a`, appending `_metadata` and then filtering changes the surviving order to `a,b,c,d`. Recovery from an older checkpoint can therefore silently emit duplicates or discard distinct rows when key types match. Filter eligible child attributes before computing legacy ordering, and add a four-column checkpoint-recovery regression. Confirmed with isolated Scala 2.13.18 execution and source tracing; no full Spark test ran.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59315][SQL][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark) [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #58592: [SPARK-59315][SQL][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark)
URL: https://github.com/apache/spark/pull/58592
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org