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: [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]
