You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ka...@apache.org on 2026/09/10 12:38:16 UTC
(spark) branch branch-4.x updated: [SPARK-59315][SQL][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark)
This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 989e1bd3ae3c [SPARK-59315][SQL][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark)
989e1bd3ae3c is described below
commit 989e1bd3ae3c488eca5629aefb1c81c5baea5788
Author: Jungtaek Lim <ka...@gmail.com>
AuthorDate: Thu Sep 10 21:37:49 2026 +0900
[SPARK-59315][SQL][SS] Fix the correctness issue on new resolution logic for dropDuplicates(WithinWatermark)
### 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 during Spark 4.3.0. 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)
Closes #58592 from HeartSaVioR/fix-deduplicate-metadata.
Authored-by: Jungtaek Lim <ka...@gmail.com>
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
(cherry picked from commit 541347c498598d196a382fc98a294f87e98f6c23)
Signed-off-by: Jungtaek Lim <ka...@gmail.com>
---
.../sql/catalyst/analysis/ResolveDeduplicate.scala | 23 ++-
.../analysis/ResolveDeduplicateSuite.scala | 97 +++++++++-
.../streaming/runtime/MicroBatchExecution.scala | 14 +-
.../.metadata.crc | Bin 0 -> 12 bytes
.../commits/.0.crc | Bin 0 -> 12 bytes
.../commits/0 | 2 +
.../metadata | 1 +
.../offsets/.0.crc | Bin 0 -> 20 bytes
.../offsets/0 | 3 +
.../sources/0/.0.crc | Bin 0 -> 12 bytes
.../sources/0/0 | 2 +
.../state/0/0/.1.delta.crc | Bin 0 -> 12 bytes
.../state/0/0/1.delta | Bin 0 -> 46 bytes
.../state/0/0/_metadata/.schema.crc | Bin 0 -> 12 bytes
.../state/0/0/_metadata/schema | Bin 0 -> 401 bytes
.../state/0/1/.1.delta.crc | Bin 0 -> 12 bytes
.../state/0/1/1.delta | Bin 0 -> 46 bytes
.../state/0/2/.1.delta.crc | Bin 0 -> 12 bytes
.../state/0/2/1.delta | Bin 0 -> 105 bytes
.../state/0/3/.1.delta.crc | Bin 0 -> 12 bytes
.../state/0/3/1.delta | Bin 0 -> 46 bytes
.../state/0/4/.1.delta.crc | Bin 0 -> 12 bytes
.../state/0/4/1.delta | Bin 0 -> 46 bytes
.../state/0/_metadata/.metadata.crc | Bin 0 -> 12 bytes
.../state/0/_metadata/metadata | 2 +
.../StreamingDeduplicationWithMetadataSuite.scala | 203 +++++++++++++++++++++
26 files changed, 337 insertions(+), 10 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicate.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicate.scala
index 8d6b94b728e4..15ba881092a7 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicate.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicate.scala
@@ -19,8 +19,8 @@ package org.apache.spark.sql.catalyst.analysis
import scala.collection.mutable
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.plans.logical.{Deduplicate, DeduplicateAllColumnsAsKey, DeduplicateKeyColumns, DeduplicateSpec, DeduplicateWithinWatermark, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, MetadataAttribute}
+import org.apache.spark.sql.catalyst.plans.logical.{Deduplicate, DeduplicateAllColumnsAsKey, DeduplicateKeyColumns, DeduplicateSpec, DeduplicateWithinWatermark, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.UNRESOLVED_DEDUPLICATE
import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -89,6 +89,25 @@ 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)))
+ val outputAtMetadataBoundary = child.output.filter { attribute =>
+ !MetadataAttribute.isValid(attribute.metadata) || originalMetadataKeys.contains(attribute)
+ }
+ computeKeys(Project(outputAtMetadataBoundary, child), spec, orderDeterministically, resolver)
+ }
+
/**
* Reproduces Spark Classic's legacy dedup-key resolution (SPARK-31990), which collapsed the
* requested column names through a `Set`. The resulting (Set iteration) order is preserved for
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicateSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicateSuite.scala
index e75bd06af4b4..89de6463b178 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicateSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicateSuite.scala
@@ -19,9 +19,10 @@ package org.apache.spark.sql.catalyst.analysis
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.dsl.expressions._
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.plans.logical.{Deduplicate, DeduplicateAllColumnsAsKey, DeduplicateKeyColumns, DeduplicateWithinWatermark, LocalRelation, LogicalPlan}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, MetadataAttribute}
+import org.apache.spark.sql.catalyst.plans.logical.{Deduplicate, DeduplicateAllColumnsAsKey, DeduplicateKeyColumns, DeduplicateSpec, DeduplicateWithinWatermark, LeafNode, LocalRelation, LogicalPlan, Project}
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StringType
/**
* Unit tests for [[ResolveDeduplicate]], the analyzer rule that resolves [[UnresolvedDeduplicate]]
@@ -30,6 +31,10 @@ import org.apache.spark.sql.internal.SQLConf
*/
class ResolveDeduplicateSuite extends AnalysisTest {
+ private case class RelationWithMetadata(
+ override val output: Seq[Attribute],
+ override val metadataOutput: Seq[Attribute]) extends LeafNode
+
private val a = $"a".int
private val b = $"b".int
private val c = $"c".int
@@ -130,6 +135,94 @@ class ResolveDeduplicateSuite extends AnalysisTest {
assert(resolved.asInstanceOf[DeduplicateWithinWatermark].keys.map(_.name) === Seq("a"))
}
+ test("recomputation excludes metadata added after the deduplication boundary") {
+ val metadataAddedAfterDeduplication = MetadataAttribute("_metadata", StringType)
+ val metadataRelation = RelationWithMetadata(rel.output, Seq(metadataAddedAfterDeduplication))
+ // Model AddMetadataColumns promoting a downstream metadata reference after keys were resolved.
+ val childAfterMetadataPropagation = Project(
+ metadataRelation.output :+ metadataAddedAfterDeduplication, metadataRelation)
+ val originalKeys = metadataRelation.output
+ val spec = DeduplicateSpec(DeduplicateAllColumnsAsKey, viaSparkClassic = true)
+
+ Seq(false, true).foreach { orderDeterministically =>
+ val expectedKeys = ResolveDeduplicate.computeKeys(
+ metadataRelation, spec, orderDeterministically, SQLConf.get.resolver)
+ val recomputedKeys = ResolveDeduplicate.recomputeKeysPreservingMetadataBoundary(
+ originalKeys, childAfterMetadataPropagation, spec, orderDeterministically,
+ SQLConf.get.resolver)
+
+ withClue(s"orderDeterministically=$orderDeterministically: ") {
+ assert(recomputedKeys === expectedKeys)
+ }
+ }
+ }
+
+ test("recomputation retains metadata visible at the deduplication boundary") {
+ val metadataSelectedBeforeDeduplication = MetadataAttribute("_metadata", StringType)
+ val metadataRelation = RelationWithMetadata(
+ rel.output, Seq(metadataSelectedBeforeDeduplication))
+ // Model an explicit metadata projection before deduplication keys were resolved.
+ val childWithSelectedMetadata = Project(
+ metadataRelation.output :+ metadataSelectedBeforeDeduplication, metadataRelation)
+ val originalKeys = childWithSelectedMetadata.output
+ val spec = DeduplicateSpec(DeduplicateAllColumnsAsKey, viaSparkClassic = true)
+
+ Seq(false, true).foreach { orderDeterministically =>
+ val expectedKeys = ResolveDeduplicate.computeKeys(
+ childWithSelectedMetadata, spec, orderDeterministically, SQLConf.get.resolver)
+ val recomputedKeys = ResolveDeduplicate.recomputeKeysPreservingMetadataBoundary(
+ originalKeys, childWithSelectedMetadata, spec, orderDeterministically,
+ SQLConf.get.resolver)
+
+ withClue(s"orderDeterministically=$orderDeterministically: ") {
+ assert(recomputedKeys === expectedKeys)
+ }
+ }
+ }
+
+ test("metadata boundary preservation does not change legacy recomputation without metadata") {
+ val specs = Seq(
+ DeduplicateSpec(DeduplicateAllColumnsAsKey, viaSparkClassic = true),
+ DeduplicateSpec(DeduplicateAllColumnsAsKey, viaSparkClassic = false),
+ DeduplicateSpec(DeduplicateKeyColumns(Seq("c", "a", "c", "b")),
+ viaSparkClassic = true),
+ DeduplicateSpec(DeduplicateKeyColumns(Seq("c", "a", "c", "b")),
+ viaSparkClassic = false))
+
+ specs.foreach { spec =>
+ val legacyKeys = ResolveDeduplicate.computeKeys(
+ rel, spec, orderDeterministically = false, SQLConf.get.resolver)
+ val recomputedKeys = ResolveDeduplicate.recomputeKeysPreservingMetadataBoundary(
+ legacyKeys, rel, spec, orderDeterministically = false, SQLConf.get.resolver)
+
+ withClue(s"spec=$spec: ") {
+ assert(recomputedKeys === legacyKeys)
+ }
+ }
+ }
+
+ test("metadata propagation does not change legacy key order") {
+ val ordinaryRelation = LocalRelation(
+ Symbol("column_1").int,
+ Symbol("column_2").int,
+ Symbol("column_3").int,
+ Symbol("column_4").int)
+ val metadataAddedAfterDeduplication = MetadataAttribute("_metadata", StringType)
+ val metadataRelation = RelationWithMetadata(
+ ordinaryRelation.output, Seq(metadataAddedAfterDeduplication))
+ val childAfterMetadataPropagation = Project(
+ metadataRelation.output :+ metadataAddedAfterDeduplication, metadataRelation)
+ val spec = DeduplicateSpec(DeduplicateAllColumnsAsKey, viaSparkClassic = true)
+ val legacyKeys = ResolveDeduplicate.computeKeys(
+ metadataRelation, spec, orderDeterministically = false, SQLConf.get.resolver)
+
+ val recomputedKeys = ResolveDeduplicate.recomputeKeysPreservingMetadataBoundary(
+ legacyKeys, childAfterMetadataPropagation, spec, orderDeterministically = false,
+ SQLConf.get.resolver)
+
+ assert(recomputedKeys === legacyKeys)
+ }
+
test("SPARK-57489: duplicate-named columns produce multiple keys (filter, not find)") {
val a2 = $"a".int
val dupRel = LocalRelation(a, a2)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
index e7500e6d634d..8c70efcb8615 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala
@@ -247,13 +247,15 @@ class MicroBatchExecution(
}
}.getOrElse(streamConf.getConf(SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER))
val dedupResolver = sparkSessionForStream.sessionState.analyzer.resolver
+ // Recompute streaming subplans for checkpoint compatibility and static subplans so batch
+ // deduplication keeps the same semantics when embedded in a streaming query.
val planWithDedupKeys = analyzedPlan.transformUp {
- case d @ Deduplicate(_, child, Some(spec)) =>
- d.copy(keys =
- ResolveDeduplicate.computeKeys(child, spec, orderDeterministically, dedupResolver))
- case d @ DeduplicateWithinWatermark(_, child, Some(spec)) =>
- d.copy(keys =
- ResolveDeduplicate.computeKeys(child, spec, orderDeterministically, dedupResolver))
+ case d @ Deduplicate(keys, child, Some(spec)) =>
+ d.copy(keys = ResolveDeduplicate.recomputeKeysPreservingMetadataBoundary(
+ keys, child, spec, orderDeterministically, dedupResolver))
+ case d @ DeduplicateWithinWatermark(keys, child, Some(spec)) =>
+ d.copy(keys = ResolveDeduplicate.recomputeKeysPreservingMetadataBoundary(
+ keys, child, spec, orderDeterministically, dedupResolver))
}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/.metadata.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/.metadata.crc
new file mode 100644
index 000000000000..948e724b8c9b
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/.metadata.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/commits/.0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/commits/.0.crc
new file mode 100644
index 000000000000..1aee7033161e
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/commits/.0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/commits/0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/commits/0
new file mode 100644
index 000000000000..9c1e3021c3ea
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/commits/0
@@ -0,0 +1,2 @@
+v1
+{"nextBatchWatermarkMs":0}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/metadata b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/metadata
new file mode 100644
index 000000000000..1895d989bcd8
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/metadata
@@ -0,0 +1 @@
+{"id":"ac2d2ec9-d12e-4951-a216-4c582c7f71da"}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/offsets/.0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/offsets/.0.crc
new file mode 100644
index 000000000000..48368ebbff42
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/offsets/.0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/offsets/0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/offsets/0
new file mode 100644
index 000000000000..ed3c94cff58b
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/offsets/0
@@ -0,0 +1,3 @@
+v1
+{"batchWatermarkMs":0,"batchTimestampMs":1788920698107,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider","spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion":"2","spark.sql.streaming.stateStore.rocksdb.formatVersion":"5","spark.sql.streaming.queryEvolution.enableSourceEvolution":"false","spark.sql.streaming.stateStore.encodingFormat":"unsaferow","spark.sql.streaming.statefulOperator.useStrictDistr [...]
+{"logOffset":0}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/sources/0/.0.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/sources/0/.0.crc
new file mode 100644
index 000000000000..e9faddc2de67
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/sources/0/.0.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/sources/0/0 b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/sources/0/0
new file mode 100644
index 000000000000..96f2b232259b
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/sources/0/0
@@ -0,0 +1,2 @@
+v1
+{"path":"file:///private/tmp/spark-59315-golden-source/first","timestamp":1788920696120,"batchId":0}
\ No newline at end of file
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/.1.delta.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/.1.delta.crc
new file mode 100644
index 000000000000..cf1d68e2acee
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/.1.delta.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/1.delta b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/1.delta
new file mode 100644
index 000000000000..635297805184
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/1.delta differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/_metadata/.schema.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/_metadata/.schema.crc
new file mode 100644
index 000000000000..5f3f51bed64f
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/_metadata/.schema.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/_metadata/schema b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/_metadata/schema
new file mode 100644
index 000000000000..89a4574cf57b
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/0/_metadata/schema differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/1/.1.delta.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/1/.1.delta.crc
new file mode 100644
index 000000000000..cf1d68e2acee
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/1/.1.delta.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/1/1.delta b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/1/1.delta
new file mode 100644
index 000000000000..635297805184
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/1/1.delta differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/2/.1.delta.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/2/.1.delta.crc
new file mode 100644
index 000000000000..e626a1c45b0e
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/2/.1.delta.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/2/1.delta b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/2/1.delta
new file mode 100644
index 000000000000..78a846a6645e
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/2/1.delta differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/3/.1.delta.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/3/.1.delta.crc
new file mode 100644
index 000000000000..cf1d68e2acee
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/3/.1.delta.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/3/1.delta b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/3/1.delta
new file mode 100644
index 000000000000..635297805184
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/3/1.delta differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/4/.1.delta.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/4/.1.delta.crc
new file mode 100644
index 000000000000..cf1d68e2acee
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/4/.1.delta.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/4/1.delta b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/4/1.delta
new file mode 100644
index 000000000000..635297805184
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/4/1.delta differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/_metadata/.metadata.crc b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/_metadata/.metadata.crc
new file mode 100644
index 000000000000..b29ccc7fb0bc
Binary files /dev/null and b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/_metadata/.metadata.crc differ
diff --git a/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/_metadata/metadata b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/_metadata/metadata
new file mode 100644
index 000000000000..39ce28c9b4aa
--- /dev/null
+++ b/sql/core/src/test/resources/structured-streaming/checkpoint-version-4.2.0-deduplication-metadata-boundary/state/0/_metadata/metadata
@@ -0,0 +1,2 @@
+v1
+{"operatorInfo":{"operatorId":0,"operatorName":"dedupe"},"stateStoreInfo":[{"storeName":"default","numColsPrefixKey":0,"numPartitions":5}]}
\ No newline at end of file
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationWithMetadataSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationWithMetadataSuite.scala
new file mode 100644
index 000000000000..467b16418387
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationWithMetadataSuite.scala
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.streaming
+
+import java.io.File
+import java.nio.charset.StandardCharsets.UTF_8
+
+import org.apache.hadoop.fs.Path
+
+import org.apache.spark.sql.catalyst.util.stringToFile
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions.{concat, lit, timestamp_seconds}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.util.Utils
+
+class StreamingDeduplicationWithMetadataSuite extends StreamTest {
+
+ import testImplicits._
+
+ private val confKey = SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER.key
+
+ private def testWithKeyOrders(name: String)(f: => Unit): Unit = {
+ Seq(false, true).foreach { orderDeterministically =>
+ test(s"$name (orderDeterministically = $orderDeterministically)") {
+ withSQLConf(confKey -> orderDeterministically.toString)(f)
+ }
+ }
+ }
+
+ testWithKeyOrders("metadata added downstream does not become a streaming deduplication key") {
+ withTempDir { src =>
+ stringToFile(new File(src, "first"), "same")
+ stringToFile(new File(src, "second"), "same")
+ val input = spark.readStream.format("text").load(src.getCanonicalPath)
+ val metadataAfterDedup = input
+ .dropDuplicates()
+ .select($"value", $"_metadata.file_path".isNotNull.as("hasMetadata"))
+
+ testStream(metadataAfterDedup)(
+ StartStream(),
+ ProcessAllAvailable(),
+ CheckAnswer(("same", true)))
+ }
+ }
+
+ testWithKeyOrders("metadata visible before streaming deduplication remains a key") {
+ withTempDir { src =>
+ stringToFile(new File(src, "first"), "same")
+ stringToFile(new File(src, "second"), "same")
+ val input = spark.readStream.format("text").load(src.getCanonicalPath)
+ val metadataBeforeDedup = input
+ .select($"value", $"_metadata")
+ .dropDuplicates()
+ .select($"value", $"_metadata.file_path".isNotNull.as("hasMetadata"))
+
+ testStream(metadataBeforeDedup)(
+ StartStream(),
+ ProcessAllAvailable(),
+ CheckAnswer(("same", true), ("same", true)))
+ }
+ }
+
+ testWithKeyOrders(
+ "metadata added downstream does not become a within-watermark deduplication key") {
+ withTempDir { src =>
+ stringToFile(new File(src, "first"), "same")
+ stringToFile(new File(src, "second"), "same")
+ val input = spark.readStream.format("text").load(src.getCanonicalPath)
+ .withColumn("eventTime", timestamp_seconds(lit(1)))
+ .withWatermark("eventTime", "10 seconds")
+ val metadataAfterDedup = input
+ .dropDuplicatesWithinWatermark()
+ .select($"value", $"_metadata.file_path".isNotNull.as("hasMetadata"))
+
+ testStream(metadataAfterDedup)(
+ StartStream(),
+ ProcessAllAvailable(),
+ CheckAnswer(("same", true)))
+ }
+ }
+
+ testWithKeyOrders("metadata visible before within-watermark deduplication remains a key") {
+ withTempDir { src =>
+ stringToFile(new File(src, "first"), "same")
+ stringToFile(new File(src, "second"), "same")
+ val input = spark.readStream.format("text").load(src.getCanonicalPath)
+ .withColumn("eventTime", timestamp_seconds(lit(1)))
+ .withWatermark("eventTime", "10 seconds")
+ val metadataBeforeDedup = input
+ .select($"value", $"eventTime", $"_metadata")
+ .dropDuplicatesWithinWatermark()
+ .select($"value", $"_metadata.file_path".isNotNull.as("hasMetadata"))
+
+ testStream(metadataBeforeDedup)(
+ StartStream(),
+ ProcessAllAvailable(),
+ CheckAnswer(("same", true), ("same", true)))
+ }
+ }
+
+ test("metadata propagation preserves key order from a Spark 4.2 checkpoint") {
+ withTempDir { src =>
+ withTempDir { checkpoint =>
+ val resource = this.getClass.getResource(
+ "/structured-streaming/" +
+ "checkpoint-version-4.2.0-deduplication-metadata-boundary/").toURI
+ Utils.copyDirectory(new File(resource), checkpoint)
+ val firstFile = stringToFile(new File(src, "first"), "same")
+ stringToFile(new File(src, "second"), "same")
+
+ // File source logs contain absolute paths. Point the copied batch-0 entry at the equivalent
+ // file in this test's temporary source directory while leaving the 4.2 state untouched.
+ val sourceLog = new Path(checkpoint.getCanonicalPath + "/sources/0/0")
+ val fileSystem = sourceLog.getFileSystem(spark.sessionState.newHadoopConf())
+ val output = fileSystem.create(sourceLog, true)
+ try {
+ output.write(
+ s"""v1
+ |{"path":"${firstFile.toURI}","timestamp":0,"batchId":0}""".stripMargin
+ .getBytes(UTF_8))
+ } finally {
+ output.close()
+ }
+
+ // Spark 4.2 wrote the checkpoint after processing another file containing "same" with
+ // four ordinary columns as keys. Its offset log has no deterministic-key-order setting,
+ // so restart must use the legacy key order even though the session default is true.
+ val result = spark.readStream.format("text").load(src.getCanonicalPath)
+ .select(
+ concat($"value", lit("_1")).as("column_1"),
+ concat($"value", lit("_2")).as("column_2"),
+ concat($"value", lit("_3")).as("column_3"),
+ concat($"value", lit("_4")).as("column_4"))
+ .dropDuplicates()
+ .select($"*", $"_metadata.file_path".as("filePath"))
+
+ testStream(result)(
+ StartStream(checkpointLocation = checkpoint.getCanonicalPath),
+ ProcessAllAvailable(),
+ CheckLastBatch(),
+ StopStream)
+ }
+ }
+ }
+
+ testWithKeyOrders("metadata added downstream does not become a batch deduplication key") {
+ withTempDir { src =>
+ stringToFile(new File(src, "first"), "same")
+ stringToFile(new File(src, "second"), "same")
+ val input = spark.read.format("text").load(src.getCanonicalPath)
+ val batchMetadataAfterDedup = input
+ .dropDuplicates()
+ .select($"value".as("batchValue"), $"_metadata.file_path".isNotNull.as("hasMetadata"))
+ val stream = MemoryStream[String]
+ val streamingInput = stream.toDF().toDF("streamValue")
+ val result = streamingInput.join(
+ batchMetadataAfterDedup,
+ $"streamValue" === $"batchValue")
+ .select($"batchValue", $"hasMetadata")
+
+ testStream(result)(
+ AddData(stream, "same"),
+ CheckAnswer(("same", true)))
+ }
+ }
+
+ testWithKeyOrders("metadata visible before batch deduplication remains a key") {
+ withTempDir { src =>
+ stringToFile(new File(src, "first"), "same")
+ stringToFile(new File(src, "second"), "same")
+ val input = spark.read.format("text").load(src.getCanonicalPath)
+ val batchMetadataBeforeDedup = input
+ .select($"value", $"_metadata")
+ .dropDuplicates()
+ .select($"value".as("batchValue"), $"_metadata.file_path".isNotNull.as("hasMetadata"))
+ val stream = MemoryStream[String]
+ val streamingInput = stream.toDF().toDF("streamValue")
+ val result = streamingInput.join(
+ batchMetadataBeforeDedup,
+ $"streamValue" === $"batchValue")
+ .select($"batchValue", $"hasMetadata")
+
+ testStream(result)(
+ AddData(stream, "same"),
+ CheckAnswer(("same", true), ("same", true)))
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org