This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 86deed7748a9 [SPARK-57489][SS] Deterministic dropDuplicates key
resolution via shared analyzer rule
86deed7748a9 is described below
commit 86deed7748a96cf6a35b4079b9eb08beb8e9ff1c
Author: Jungtaek Lim <[email protected]>
AuthorDate: Wed Jun 17 10:37:19 2026 +0900
[SPARK-57489][SS] Deterministic dropDuplicates key resolution via shared
analyzer rule
### What changes were proposed in this pull request?
`dropDuplicates` / `dropDuplicatesWithinWatermark` historically resolve
their key columns eagerly: in the DataFrame implementation for Spark Classic
(`Dataset.groupColsFromDropDuplicates`, via `colNames.toSet.toSeq`) and in the
Spark Connect planner (`transformDeduplicate`). The Spark Classic path's key
order therefore depends on Scala's `Set` iteration order, which is undocumented
and changed between Scala 2.12 and 2.13, and which also differs from Spark
Connect. Because streaming de [...]
This keeps binding by position, but makes the resolution deterministic and
shared between engines:
- Add an unresolved logical node `UnresolvedDeduplicate` and an analyzer
rule `ResolveDeduplicate` that both Spark Classic (`Dataset.dropDuplicates*`)
and Spark Connect (`transformDeduplicate`) defer to. The rule resolves the
requested columns with an explicit, order-preserving dedup (intentionally NOT
`Seq.distinct`/`Set`, whose ordering is not contractually guaranteed), so the
key order is deterministic and identical across Scala versions and engines.
- Add `spark.sql.dropDuplicates.deterministicKeyOrder.enabled` (default
`true`), persisted via the offset log (`OffsetSeqMetadata.relevantSQLConfs`,
with legacy default `false`), so the deterministic order applies only to newly
started streaming queries; existing checkpoints keep their original key order.
- The resolved `Deduplicate` / `DeduplicateWithinWatermark` node retains
the original recipe (a `DeduplicateSpec`: a `DeduplicateKeySpec` — explicit key
columns or all columns — plus `viaSparkClassic`). For streaming queries,
`MicroBatchExecution.logicalPlan` recomputes the key columns at query start
using the order pinned in the offset log (read via
`OffsetSeqMetadata.readValueOpt`, mirroring how
`ENABLE_STREAMING_SOURCE_EVOLUTION` is read). When the pinned value is the
legacy order, [...]
### Why are the changes needed?
Streaming deduplication binds state-store keys by position. A key order
that depends on Scala's `Set` iteration order is non-deterministic across Scala
versions and differs between Spark Classic and Spark Connect, which can break
streaming query restarts and cross-engine checkpoint compatibility.
### Does this PR introduce _any_ user-facing change?
Yes.
- For newly started streaming `dropDuplicates` /
`dropDuplicatesWithinWatermark` queries, the key columns are resolved in a
deterministic, first-occurrence order instead of a Scala-`Set`-dependent order.
The deduplication results (rows) are unchanged; only the internal key ordering
used for the state store changes.
- Existing checkpointed queries are unaffected: the effective order is
pinned per query via the offset log.
- Spark Classic and Spark Connect now resolve `dropDuplicates` keys
identically.
- An unresolvable `dropDuplicates` column now fails at query analysis
(consistent with `Project`/`Filter`) rather than at the API call in Spark
Classic. For Spark Connect, it now surfaces the analyzer's
`UNRESOLVED_COLUMN_AMONG_FIELD_NAMES` error instead of the Connect-specific
`CONNECT_INVALID_PLAN.UNRESOLVED_COLUMN_AMONG_FIELD_NAMES` (removed as dead
code), so both engines report the same error condition.
### How was this patch tested?
New tests:
- `ResolveDeduplicateSuite` — unit tests for the analyzer rule
(deterministic first-occurrence order, engine parity, Classic/Connect legacy
fallback, all-columns, within-watermark, duplicate-named columns, and
unknown-column failure).
- `StreamingDeduplicationConnectInteropSuite` — a checkpoint written by one
engine restarts under the other.
- `StreamingDeduplicationFallbackSuite` — an existing checkpoint without
the conf falls back to the legacy key order; a newly started query pins the
deterministic order and keeps it on restart.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #56546 from HeartSaVioR/dropduplicates-deterministic-key-order.
Authored-by: Jungtaek Lim <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 5 -
.../spark/sql/catalyst/analysis/Analyzer.scala | 1 +
.../sql/catalyst/analysis/ResolveDeduplicate.scala | 126 ++++++++++++++++
.../spark/sql/catalyst/analysis/unresolved.scala | 33 ++++-
.../spark/sql/catalyst/optimizer/Optimizer.scala | 18 +--
.../plans/logical/basicLogicalOperators.scala | 38 ++++-
.../spark/sql/catalyst/trees/TreePatterns.scala | 1 +
.../org/apache/spark/sql/internal/SQLConf.scala | 15 ++
.../analysis/ResolveDeduplicateSuite.scala | 164 +++++++++++++++++++++
.../sql/connect/planner/InvalidInputErrors.scala | 5 -
.../sql/connect/planner/SparkConnectPlanner.scala | 31 ++--
...StreamingDeduplicationConnectInteropSuite.scala | 142 ++++++++++++++++++
.../connect/planner/SparkConnectProtoSuite.scala | 14 +-
.../org/apache/spark/sql/classic/Dataset.scala | 55 +++----
.../spark/sql/execution/SparkStrategies.scala | 6 +-
.../streaming/checkpointing/OffsetSeq.scala | 6 +-
.../streaming/runtime/MicroBatchExecution.scala | 27 +++-
.../StreamingDeduplicationFallbackSuite.scala | 143 ++++++++++++++++++
18 files changed, 750 insertions(+), 80 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index ad6e0dfc4c23..54153d09788e 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -1453,11 +1453,6 @@
"Unknown Analyze Method <other>!"
]
},
- "UNRESOLVED_COLUMN_AMONG_FIELD_NAMES" : {
- "message" : [
- "Cannot resolve column name \"<colName>\" among (<fieldNames>)."
- ]
- },
"UNRESOLVED_NAMED_LAMBDA_VARIABLE_REQUIRES_NAME_PART" : {
"message" : [
"UnresolvedNamedLambdaVariable requires at least one name part!"
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 041621b64c13..7a5048ed0a41 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -564,6 +564,7 @@ class Analyzer(
ResolveBinaryArithmetic ::
new ResolveIdentifierClause(earlyBatches) ::
ResolveUnion ::
+ ResolveDeduplicate ::
ResolveZip ::
FlattenSequentialStreamingUnion ::
ValidateSequentialStreamingUnion ::
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
new file mode 100644
index 000000000000..8d6b94b728e4
--- /dev/null
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicate.scala
@@ -0,0 +1,126 @@
+/*
+ * 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.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.rules.Rule
+import org.apache.spark.sql.catalyst.trees.TreePattern.UNRESOLVED_DEDUPLICATE
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Resolves [[UnresolvedDeduplicate]] (built by `Dataset.dropDuplicates*` in
Spark Classic and by
+ * the Deduplicate relation in the Spark Connect planner) into a
[[Deduplicate]] /
+ * [[DeduplicateWithinWatermark]] with resolved key attributes, shared by both
engines.
+ *
+ * The key attributes are computed with
[[SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER]] read
+ * from the current session: true (the default) produces a stable,
first-occurrence ordering; false
+ * reproduces each engine's legacy resolution. The resolved node also carries
the original recipe
+ * (a `DeduplicateSpec` holding the key spec and `viaSparkClassic`) so
streaming queries can
+ * recompute the keys at query start with the value pinned in the offset log
(see
+ * `MicroBatchExecution.logicalPlan` and [[computeKeys]]). See SPARK-57489.
+ */
+object ResolveDeduplicate extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsUpWithPruning(
+ _.containsPattern(UNRESOLVED_DEDUPLICATE)) {
+ case d: UnresolvedDeduplicate if d.child.resolved =>
+ val orderDeterministically =
conf.getConf(SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER)
+ val spec = DeduplicateSpec(d.keySpec, d.viaSparkClassic)
+ val keys = computeKeys(d.child, spec, orderDeterministically,
conf.resolver)
+ if (d.withinWatermark) {
+ DeduplicateWithinWatermark(keys, d.child, Some(spec))
+ } else {
+ Deduplicate(keys, d.child, Some(spec))
+ }
+ }
+
+ /**
+ * Computes the deduplication key attributes from the requested recipe
against `child`'s output.
+ * Shared by the analyzer rule (with the session config) and by the
streaming bootstrap (with the
+ * value pinned in the offset log), so the ordering rules stay in one place.
+ *
+ * @param orderDeterministically when true, a stable first-occurrence order;
when false, the
+ * legacy (engine-specific) order selected by `spec.viaSparkClassic`.
+ */
+ def computeKeys(
+ child: LogicalPlan,
+ spec: DeduplicateSpec,
+ orderDeterministically: Boolean,
+ resolver: Resolver): Seq[Attribute] = {
+ spec.keySpec match {
+ case DeduplicateAllColumnsAsKey =>
+ // All child columns are keys. The deterministic order and legacy
Spark Connect both key on
+ // the child output directly (in output order, no name resolution);
only legacy Spark
+ // Classic reorders the names through a Set (see
legacyClassicColumnNames).
+ if (!orderDeterministically && spec.viaSparkClassic) {
+ resolveColumnNames(child,
legacyClassicColumnNames(child.output.map(_.name)), resolver)
+ } else {
+ child.output
+ }
+ case DeduplicateKeyColumns(colNames) =>
+ val orderedNames = if (orderDeterministically) {
+ dedupKeepingOrder(colNames)
+ } else if (spec.viaSparkClassic) {
+ legacyClassicColumnNames(colNames)
+ } else {
+ // Legacy Spark Connect resolution: no dedup, caller-provided input
order.
+ colNames
+ }
+ resolveColumnNames(child, orderedNames, 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
+ * existing queries whose checkpoints predate the deterministic order.
Shared by the all-columns
+ * and subset forms so both stay faithful to Spark Classic.
+ */
+ private def legacyClassicColumnNames(names: Seq[String]): Seq[String] =
names.toSet.toSeq
+
+ private def resolveColumnNames(
+ child: LogicalPlan, names: Seq[String], resolver: Resolver):
Seq[Attribute] = {
+ names.flatMap { colName =>
+ // It is possible there is more than one column with the same name, so
we call filter
+ // instead of find.
+ val cols = child.output.filter(col => resolver(col.name, colName))
+ if (cols.isEmpty) {
+ throw
QueryCompilationErrors.cannotResolveColumnNameAmongAttributesError(
+ colName, child.output.map(_.name).mkString(", "))
+ }
+ cols
+ }
+ }
+
+ /**
+ * Deduplicate `names` preserving first-occurrence order. The stable
ordering is an invariant for
+ * stateful operators, so we rely only on the documented contract of `Seq`
traversal and explicit
+ * append, not on `Seq.distinct`/`Set` (whose ordering is not contractually
guaranteed).
+ * See SPARK-57489.
+ */
+ private def dedupKeepingOrder(names: Seq[String]): Seq[String] = {
+ val seen = mutable.HashSet.empty[String]
+ val out = mutable.ArrayBuffer.empty[String]
+ names.foreach(name => if (seen.add(name)) out += name)
+ out.toSeq
+ }
+}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
index a5b467d0f081..fc51dd8c72d4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
@@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.{FunctionIdentifier,
InternalRow, TableIden
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
ExprCode}
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
-import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan,
SupportsSubquery, UnaryNode}
+import org.apache.spark.sql.catalyst.plans.logical.{DeduplicateKeySpec,
LeafNode, LogicalPlan, SupportsSubquery, UnaryNode}
import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId
@@ -1168,6 +1168,37 @@ case class UnresolvedQualify(condition: Expression,
child: LogicalPlan) extends
final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_QUALIFY)
}
+/**
+ * An unresolved logical plan for `dropDuplicates` /
`dropDuplicatesWithinWatermark`. It holds the
+ * user-requested deduplication columns (by name, or "all columns") and is
resolved by the
+ * `ResolveDeduplicate` analyzer rule into a [[Deduplicate]] /
[[DeduplicateWithinWatermark]].
+ *
+ * Resolving the keys in the analyzer (rather than eagerly in the DataFrame
API or the Spark Connect
+ * planner) lets both engines share one resolution with a stable key order. A
stable order matters
+ * for streaming deduplication, whose state store binds keys by position: a
different key order
+ * across restarts would break state-store key-schema compatibility. See
SPARK-57489.
+ *
+ * @param keySpec which columns form the deduplication key (an explicit set,
or all columns).
+ * @param withinWatermark when true, resolves to `DeduplicateWithinWatermark`.
+ * @param viaSparkClassic whether the deduplication was requested via Spark
Classic
+ * (`Dataset.dropDuplicates*`, true) or Spark Connect
(`transformDeduplicate`, false). This only
+ * matters on the legacy-fallback path, consulted ONLY when the
deterministic key order is
+ * disabled (an existing query restored from a checkpoint that predates this
change): Spark
+ * Classic reproduces its legacy resolution (`toSet`-based dedup, `Set`
order) while Spark
+ * Connect reproduces its own (no dedup, input order). Ignored on the
deterministic path.
+ */
+case class UnresolvedDeduplicate(
+ keySpec: DeduplicateKeySpec,
+ withinWatermark: Boolean,
+ viaSparkClassic: Boolean,
+ child: LogicalPlan) extends UnaryNode {
+ override lazy val resolved: Boolean = false
+ override def output: Seq[Attribute] = child.output
+ override protected def withNewChildInternal(newChild: LogicalPlan):
UnresolvedDeduplicate =
+ copy(child = newChild)
+ final override val nodePatterns: Seq[TreePattern] =
Seq(UNRESOLVED_DEDUPLICATE)
+}
+
/**
* A place holder expression used in random functions, will be replaced after
analyze.
*/
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index 42914e98aa6d..87500f0ca514 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -894,9 +894,9 @@ object RemoveNoopUnion extends Rule[LogicalPlan] {
_.containsAllPatterns(DISTINCT_LIKE, UNION)) {
case d @ Distinct(u: Union) =>
d.withNewChildren(Seq(simplifyUnion(u)))
- case d @ Deduplicate(_, u: Union) =>
+ case d @ Deduplicate(_, u: Union, _) =>
d.withNewChildren(Seq(simplifyUnion(u)))
- case d @ DeduplicateWithinWatermark(_, u: Union) =>
+ case d @ DeduplicateWithinWatermark(_, u: Union, _) =>
d.withNewChildren(Seq(simplifyUnion(u)))
}
}
@@ -1858,12 +1858,12 @@ object CombineUnions extends Rule[LogicalPlan] {
case SequentialOrSimpleUnion(u) => flattenUnion(u, false)
case Distinct(SequentialOrSimpleUnion(u)) => Distinct(flattenUnion(u,
true))
// Only handle distinct-like 'Deduplicate', where the keys == output
- case Deduplicate(keys: Seq[Attribute], SequentialOrSimpleUnion(u))
+ case d @ Deduplicate(keys: Seq[Attribute], SequentialOrSimpleUnion(u), _)
if AttributeSet(keys) == u.outputSet =>
- Deduplicate(keys, flattenUnion(u, true))
- case DeduplicateWithinWatermark(keys: Seq[Attribute],
SequentialOrSimpleUnion(u))
+ d.copy(child = flattenUnion(u, true))
+ case d @ DeduplicateWithinWatermark(keys: Seq[Attribute],
SequentialOrSimpleUnion(u), _)
if AttributeSet(keys) == u.outputSet =>
- DeduplicateWithinWatermark(keys, flattenUnion(u, true))
+ d.copy(child = flattenUnion(u, true))
}
private def flattenUnion(union: UnionBase, flattenDistinct: Boolean):
UnionBase = {
@@ -1894,7 +1894,7 @@ object CombineUnions extends Rule[LogicalPlan] {
case Distinct(SequentialOrSimpleUnion(u)) if flattenDistinct &&
canMerge(u) =>
stack.pushAll(u.children.reverse)
// Only handle distinct-like 'Deduplicate', where the keys == output
- case Deduplicate(keys: Seq[Attribute], SequentialOrSimpleUnion(u))
+ case Deduplicate(keys: Seq[Attribute], SequentialOrSimpleUnion(u), _)
if flattenDistinct && canMerge(u) && AttributeSet(keys) ==
u.outputSet =>
stack.pushAll(u.children.reverse)
case SequentialOrSimpleUnion(u) if canMerge(u) =>
@@ -1907,7 +1907,7 @@ object CombineUnions extends Rule[LogicalPlan] {
canPushProjectionThroughUnion(project) =>
stack.pushAll(pushProjectionThroughUnion(projectList, u).reverse)
case project @ Project(
- projectList, Deduplicate(keys: Seq[Attribute],
SequentialOrSimpleUnion(u)))
+ projectList, Deduplicate(keys: Seq[Attribute],
SequentialOrSimpleUnion(u), _))
if projectList.forall(_.deterministic) && flattenDistinct &&
canMerge(u) &&
AttributeSet(keys) == u.outputSet &&
canPushProjectionThroughUnion(project) =>
stack.pushAll(pushProjectionThroughUnion(projectList, u).reverse)
@@ -2726,7 +2726,7 @@ object ReplaceDistinctWithAggregate extends
Rule[LogicalPlan] {
*/
object ReplaceDeduplicateWithAggregate extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transformUpWithNewOutput {
- case d @ Deduplicate(keys, child) if !child.isStreaming =>
+ case d @ Deduplicate(keys, child, _) if !child.isStreaming =>
val keyExprIds = keys.map(_.exprId)
val generatedAliasesMap = new mutable.HashMap[Attribute, Alias]();
val aggCols = child.output.map { attr =>
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
index df9a6d87ecfe..0246d0b204ff 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -2312,10 +2312,35 @@ case class OneRowRelation() extends LeafNode {
}
}
+/**
+ * How the deduplication keys are specified: either an explicit set of column
names
+ * ([[DeduplicateKeyColumns]]) or all of the child's columns
([[DeduplicateAllColumnsAsKey]]).
+ */
+sealed trait DeduplicateKeySpec
+case class DeduplicateKeyColumns(colNames: Seq[String]) extends
DeduplicateKeySpec
+case object DeduplicateAllColumnsAsKey extends DeduplicateKeySpec
+
+/**
+ * The original recipe behind a [[Deduplicate]] /
[[DeduplicateWithinWatermark]] node, set by the
+ * `ResolveDeduplicate` analyzer rule and retained so a streaming query can
recompute its key
+ * attributes at query start in the ordering pinned in the offset log (see
+ * `ResolveDeduplicate.computeKeys`). A `None` spec on a node means it was not
built from
+ * `dropDuplicates*` (e.g. an internally/test-constructed node) and its keys
must NOT be recomputed.
+ *
+ * @param keySpec which columns form the deduplication key.
+ * @param viaSparkClassic whether this was built via Spark Classic
(`Dataset.dropDuplicates*`, true)
+ * or Spark Connect (`transformDeduplicate`, false). Only consulted when
recomputing the keys in
+ * the legacy order, where the two engines historically differed. See
SPARK-57489.
+ */
+case class DeduplicateSpec(
+ keySpec: DeduplicateKeySpec,
+ viaSparkClassic: Boolean)
+
/** A logical plan for `dropDuplicates`. */
case class Deduplicate(
keys: Seq[Attribute],
- child: LogicalPlan) extends UnaryNode {
+ child: LogicalPlan,
+ dedupSpec: Option[DeduplicateSpec] = None) extends UnaryNode {
override def maxRows: Option[Long] = child.maxRows
override def output: Seq[Attribute] = {
val base = child.output
@@ -2325,9 +2350,15 @@ case class Deduplicate(
override protected def withNewChildInternal(newChild: LogicalPlan):
Deduplicate =
copy(child = newChild)
override def isStateful: Boolean = child.isStreaming
+ // `dedupSpec` is internal metadata used only to recompute keys on streaming
restart; keep it out
+ // of the tree string (EXPLAIN output and golden plans).
+ override protected def stringArgs: Iterator[Any] = Iterator(keys, child)
}
-case class DeduplicateWithinWatermark(keys: Seq[Attribute], child:
LogicalPlan) extends UnaryNode {
+case class DeduplicateWithinWatermark(
+ keys: Seq[Attribute],
+ child: LogicalPlan,
+ dedupSpec: Option[DeduplicateSpec] = None) extends UnaryNode {
// Ensure that references include event time columns so they are not pruned
away.
override def references: AttributeSet = AttributeSet(keys) ++
AttributeSet(child.output.filter(_.metadata.contains(EventTimeWatermark.delayKey)))
@@ -2340,6 +2371,9 @@ case class DeduplicateWithinWatermark(keys:
Seq[Attribute], child: LogicalPlan)
override protected def withNewChildInternal(newChild: LogicalPlan):
DeduplicateWithinWatermark =
copy(child = newChild)
override def isStateful: Boolean = child.isStreaming
+ // `dedupSpec` is internal metadata used only to recompute keys on streaming
restart; keep it out
+ // of the tree string (EXPLAIN output and golden plans).
+ override protected def stringArgs: Iterator[Any] = Iterator(keys, child)
}
/**
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
index f676750a054a..096a8e3c2487 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
@@ -197,6 +197,7 @@ object TreePattern extends Enumeration {
val NAMED_STREAMING_RELATION: Value = Value
val PLAN_WITH_UNRESOLVED_IDENTIFIER: Value = Value
val UNRESOLVED_BIN_BY: Value = Value
+ val UNRESOLVED_DEDUPLICATE: Value = Value
val UNRESOLVED_EVENT_TIME_WATERMARK: Value = Value
val UNRESOLVED_HAVING: Value = Value
val UNRESOLVED_HINT: Value = Value
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index e017a9076b1d..0a94c5a5980a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -3485,6 +3485,21 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER =
+ buildConf("spark.sql.dropDuplicates.deterministicKeyOrder.enabled")
+ .internal()
+ .doc("When true, dropDuplicates and dropDuplicatesWithinWatermark
resolve their key " +
+ "columns with a stable, first-occurrence ordering shared by Spark
Classic and Spark " +
+ "Connect. The order does not affect batch results, but stateful
streaming " +
+ "deduplication binds state-store keys by position, so for streaming
the effective " +
+ "value is pinned per query in the offset log at batch 0: newly started
queries use " +
+ "the stable order, while queries restored from a checkpoint that
predates this config " +
+ "keep their original key order. Please refer to SPARK-57489 for
details.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(true)
+
val STATEFUL_OPERATOR_ALWAYS_NULLABLE_OUTPUT =
buildConf("spark.sql.streaming.statefulOperator.alwaysNullableOutput.enabled")
.internal()
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
new file mode 100644
index 000000000000..e75bd06af4b4
--- /dev/null
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveDeduplicateSuite.scala
@@ -0,0 +1,164 @@
+/*
+ * 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.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.internal.SQLConf
+
+/**
+ * Unit tests for [[ResolveDeduplicate]], the analyzer rule that resolves
[[UnresolvedDeduplicate]]
+ * into a [[Deduplicate]] / [[DeduplicateWithinWatermark]]. The rule is
applied directly so the
+ * deterministic-key-order conf can be toggled per case. See SPARK-57489.
+ */
+class ResolveDeduplicateSuite extends AnalysisTest {
+
+ private val a = $"a".int
+ private val b = $"b".int
+ private val c = $"c".int
+ private val rel = LocalRelation(c, a, b)
+
+ private val confKey = SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER.key
+
+ private def resolveKeys(
+ columnNames: Seq[String],
+ allColumnsAsKeys: Boolean = false,
+ withinWatermark: Boolean = false,
+ viaSparkClassic: Boolean = true,
+ child: LogicalPlan = rel): Seq[Attribute] = {
+ val keySpec =
+ if (allColumnsAsKeys) DeduplicateAllColumnsAsKey else
DeduplicateKeyColumns(columnNames)
+ ResolveDeduplicate(UnresolvedDeduplicate(
+ keySpec, withinWatermark, viaSparkClassic, child)) match {
+ case d: Deduplicate => d.keys
+ case d: DeduplicateWithinWatermark => d.keys
+ case other => fail(s"Expected a
(Deduplicate|DeduplicateWithinWatermark), got: $other")
+ }
+ }
+
+ test("SPARK-57489: deterministic resolution keeps first-occurrence input
order and dedups") {
+ withSQLConf(confKey -> "true") {
+ // The input order is preserved and duplicate names are collapsed at
first occurrence. Under
+ // the old `toSet.toSeq` this order would not be guaranteed; asserting
it exactly guards
+ // against anyone reintroducing Set/distinct-based ordering.
+ assert(resolveKeys(Seq("c", "a", "c", "b")).map(_.name) === Seq("c",
"a", "b"))
+ assert(resolveKeys(Seq("b", "b", "a")).map(_.name) === Seq("b", "a"))
+ }
+ }
+
+ test("SPARK-57489: deterministic resolution identical across engines
(interop)") {
+ // The deterministic path ignores the engine flag, so Classic and Connect
resolve the SAME
+ // attributes (same exprIds, same order). This is what lets a checkpoint
written by one engine
+ // restart under the other.
+ withSQLConf(confKey -> "true") {
+ val classicSubset = resolveKeys(Seq("c", "a", "c", "b"), viaSparkClassic
= true)
+ val connectSubset = resolveKeys(Seq("c", "a", "c", "b"), viaSparkClassic
= false)
+ assert(classicSubset.map(_.exprId) === connectSubset.map(_.exprId))
+ assert(classicSubset.map(_.name) === Seq("c", "a", "b"))
+
+ val classicAll = resolveKeys(Nil, allColumnsAsKeys = true,
viaSparkClassic = true)
+ val connectAll = resolveKeys(Nil, allColumnsAsKeys = true,
viaSparkClassic = false)
+ assert(classicAll.map(_.exprId) === connectAll.map(_.exprId))
+ assert(classicAll === rel.output)
+ }
+ }
+
+ test("SPARK-57489: legacy fallback - Classic dedups the subset (Set order)")
{
+ withSQLConf(confKey -> "false") {
+ val keys = resolveKeys(Seq("c", "a", "c", "b"), viaSparkClassic = true)
+ // Classic legacy used `toSet.toSeq`: duplicate names are collapsed
(order is Set-defined and
+ // intentionally not asserted here).
+ assert(keys.length === 3)
+ assert(keys.map(_.name).toSet === Set("a", "b", "c"))
+ }
+ }
+
+ test("SPARK-57489: legacy fallback - Connect keeps duplicates in input
order") {
+ withSQLConf(confKey -> "false") {
+ val keys = resolveKeys(Seq("c", "a", "c", "b"), viaSparkClassic = false)
+ // Connect legacy did not dedup the requested names.
+ assert(keys.map(_.name) === Seq("c", "a", "c", "b"))
+ }
+ }
+
+ test("SPARK-57489: allColumnsAsKeys - deterministic and Connect legacy use
child output order") {
+ // Deterministic (regardless of engine flag) keys on all child columns in
output order.
+ withSQLConf(confKey -> "true") {
+ assert(
+ resolveKeys(Nil, allColumnsAsKeys = true, viaSparkClassic = true) ===
rel.output)
+ assert(
+ resolveKeys(Nil, allColumnsAsKeys = true, viaSparkClassic = false) ===
rel.output)
+ }
+ // Legacy Spark Connect also keys on all child columns in output order.
+ withSQLConf(confKey -> "false") {
+ assert(
+ resolveKeys(Nil, allColumnsAsKeys = true, viaSparkClassic = false) ===
rel.output)
+ }
+ }
+
+ test("SPARK-57489: allColumnsAsKeys - Classic legacy fallback uses Set order
over all columns") {
+ withSQLConf(confKey -> "false") {
+ // Legacy Classic `dropDuplicates()` resolved `columns.toSet.toSeq`:
every column is present
+ // (order is Set-defined and intentionally not asserted here).
+ val keys = resolveKeys(Nil, allColumnsAsKeys = true, viaSparkClassic =
true)
+ assert(keys.map(_.exprId).toSet === rel.output.map(_.exprId).toSet)
+ }
+ }
+
+ test("SPARK-57489: withinWatermark resolves to DeduplicateWithinWatermark") {
+ val resolved = ResolveDeduplicate(
+ UnresolvedDeduplicate(DeduplicateKeyColumns(Seq("a")), withinWatermark =
true,
+ viaSparkClassic = true, rel))
+ assert(resolved.isInstanceOf[DeduplicateWithinWatermark])
+ assert(resolved.asInstanceOf[DeduplicateWithinWatermark].keys.map(_.name)
=== Seq("a"))
+ }
+
+ test("SPARK-57489: duplicate-named columns produce multiple keys (filter,
not find)") {
+ val a2 = $"a".int
+ val dupRel = LocalRelation(a, a2)
+ // A single requested name "a" must resolve to BOTH same-named columns
(filter, not find) in
+ // every mode. Compare by exprId so the test fails if we returned the same
attribute twice
+ // instead of the two distinct same-named columns.
+ Seq(
+ ("true", true), // deterministic (engine flag ignored)
+ ("true", false),
+ ("false", true), // legacy Spark Classic
+ ("false", false) // legacy Spark Connect
+ ).foreach { case (deterministic, legacy) =>
+ withSQLConf(confKey -> deterministic) {
+ assert(
+ resolveKeys(Seq("a"), viaSparkClassic = legacy, child =
dupRel).map(_.exprId)
+ === Seq(a.exprId, a2.exprId),
+ s"deterministic=$deterministic, viaSparkClassic=$legacy")
+ }
+ }
+ }
+
+ test("SPARK-57489: unknown column name fails analysis") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ ResolveDeduplicate(UnresolvedDeduplicate(
+ DeduplicateKeyColumns(Seq("does_not_exist")), withinWatermark =
false,
+ viaSparkClassic = true, rel))
+ },
+ condition = "UNRESOLVED_COLUMN_AMONG_FIELD_NAMES",
+ parameters = Map("colName" -> "does_not_exist", "fieldNames" -> "c, a,
b"))
+ }
+}
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/InvalidInputErrors.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/InvalidInputErrors.scala
index cdba6c825332..604ad3860697 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/InvalidInputErrors.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/InvalidInputErrors.scala
@@ -55,11 +55,6 @@ object InvalidInputErrors {
def deduplicateRequiresColumnsOrAll(): InvalidPlanInput =
InvalidPlanInput("CONNECT_INVALID_PLAN.DEDUPLICATE_REQUIRES_COLUMNS_OR_ALL",
Map.empty)
- def invalidDeduplicateColumn(colName: String, fieldNames: String):
InvalidPlanInput =
- InvalidPlanInput(
- "CONNECT_INVALID_PLAN.UNRESOLVED_COLUMN_AMONG_FIELD_NAMES",
- Map("colName" -> colName, "fieldNames" -> fieldNames))
-
def functionEvalTypeNotSupported(evalType: Int): InvalidPlanInput =
InvalidPlanInput(
"CONNECT_INVALID_PLAN.FUNCTION_EVAL_TYPE_NOT_SUPPORTED",
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
index b067efa0579a..94e7938b7c72 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
@@ -43,7 +43,7 @@ import org.apache.spark.internal.LogKeys.{DATAFRAME_ID,
SESSION_ID}
import org.apache.spark.resource.{ExecutorResourceRequest, ResourceProfile,
TaskResourceProfile, TaskResourceRequest}
import org.apache.spark.sql.{AnalysisException, Column, Encoders,
ForeachWriter, Row}
import org.apache.spark.sql.catalyst.{expressions, AliasIdentifier,
FunctionIdentifier, InternalRow, QueryPlanningTracker}
-import org.apache.spark.sql.catalyst.analysis.{ChangelogContextUtils,
FunctionRegistry, GlobalTempView, LocalTempView, MultiAlias, RelationChanges,
UnresolvedAlias, UnresolvedAttribute, UnresolvedDataFrameStar,
UnresolvedDeserializer, UnresolvedExtractValue, UnresolvedFunction,
UnresolvedOrdinal, UnresolvedPlanId, UnresolvedRegex, UnresolvedRelation,
UnresolvedStar, UnresolvedStarWithColumns, UnresolvedStarWithColumnsRenames,
UnresolvedSubqueryColumnAliases, UnresolvedTableValuedFunction [...]
+import org.apache.spark.sql.catalyst.analysis.{ChangelogContextUtils,
FunctionRegistry, GlobalTempView, LocalTempView, MultiAlias, RelationChanges,
UnresolvedAlias, UnresolvedAttribute, UnresolvedDataFrameStar,
UnresolvedDeduplicate, UnresolvedDeserializer, UnresolvedExtractValue,
UnresolvedFunction, UnresolvedOrdinal, UnresolvedPlanId, UnresolvedRegex,
UnresolvedRelation, UnresolvedStar, UnresolvedStarWithColumns,
UnresolvedStarWithColumnsRenames, UnresolvedSubqueryColumnAliases, Unreso [...]
import org.apache.spark.sql.catalyst.encoders.{encoderFor, AgnosticEncoder,
ExpressionEncoder, RowEncoder}
import
org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{ProductEncoder,
RowEncoder => AgnosticRowEncoder, StringEncoder, UnboundRowEncoder}
import org.apache.spark.sql.catalyst.expressions._
@@ -51,7 +51,7 @@ import
org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.parser.{NamedParameterContext,
ParameterContext, ParseException, ParserUtils, PositionalParameterContext}
import org.apache.spark.sql.catalyst.plans.{Cross, FullOuter, Inner, JoinType,
LeftAnti, LeftOuter, LeftSemi, RightOuter, UsingJoin}
import org.apache.spark.sql.catalyst.plans.logical
-import org.apache.spark.sql.catalyst.plans.logical.{AppendColumns, Assignment,
CoGroup, CollectMetrics, CommandResult, Deduplicate,
DeduplicateWithinWatermark, DeleteAction, DeserializeToObject, Except,
FlatMapGroupsWithState, InsertAction, InsertStarAction, Intersect, JoinWith,
LocalRelation, LogicalGroupState, LogicalPlan, MapGroups, MapPartitions,
MergeAction, Project, Sample, SerializeFromObject, Sort, SubqueryAlias,
TimeModes, TransformWithState, TypedFilter, Union, Unpivot, Unresol [...]
+import org.apache.spark.sql.catalyst.plans.logical.{AppendColumns, Assignment,
CoGroup, CollectMetrics, CommandResult, DeduplicateAllColumnsAsKey,
DeduplicateKeyColumns, DeleteAction, DeserializeToObject, Except,
FlatMapGroupsWithState, InsertAction, InsertStarAction, Intersect, JoinWith,
LocalRelation, LogicalGroupState, LogicalPlan, MapGroups, MapPartitions,
MergeAction, Project, Sample, SerializeFromObject, Sort, SubqueryAlias,
TimeModes, TransformWithState, TypedFilter, Union, Unpivo [...]
import org.apache.spark.sql.catalyst.streaming.InternalOutputModes
import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin, TreePattern}
import org.apache.spark.sql.catalyst.types.DataTypeUtils
@@ -1455,27 +1455,16 @@ class SparkConnectPlanner(
if (!rel.getAllColumnsAsKeys && rel.getColumnNamesCount == 0) {
throw InvalidInputErrors.deduplicateRequiresColumnsOrAll()
}
- val queryExecution = new QueryExecution(session,
transformRelation(rel.getInput))
- val resolver = session.sessionState.analyzer.resolver
- val allColumns = queryExecution.analyzed.output
- if (rel.getAllColumnsAsKeys) {
- if (rel.getWithinWatermark) DeduplicateWithinWatermark(allColumns,
queryExecution.analyzed)
- else Deduplicate(allColumns, queryExecution.analyzed)
+ val keySpec = if (rel.getAllColumnsAsKeys) {
+ DeduplicateAllColumnsAsKey
} else {
- val toGroupColumnNames = rel.getColumnNamesList.asScala.toSeq
- val groupCols = toGroupColumnNames.flatMap { (colName: String) =>
- // It is possibly there are more than one columns with the same name,
- // so we call filter instead of find.
- val cols = allColumns.filter(col => resolver(col.name, colName))
- if (cols.isEmpty) {
- val fieldNames = allColumns.map(_.name).mkString(", ")
- throw InvalidInputErrors.invalidDeduplicateColumn(colName,
fieldNames)
- }
- cols
- }
- if (rel.getWithinWatermark) DeduplicateWithinWatermark(groupCols,
queryExecution.analyzed)
- else Deduplicate(groupCols, queryExecution.analyzed)
+ DeduplicateKeyColumns(rel.getColumnNamesList.asScala.toSeq)
}
+ UnresolvedDeduplicate(
+ keySpec = keySpec,
+ withinWatermark = rel.getWithinWatermark,
+ viaSparkClassic = false,
+ child = transformRelation(rel.getInput))
}
private def transformDataType(t: proto.DataType): DataType = {
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/StreamingDeduplicationConnectInteropSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/StreamingDeduplicationConnectInteropSuite.scala
new file mode 100644
index 000000000000..fba8e3dadbc1
--- /dev/null
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/StreamingDeduplicationConnectInteropSuite.scala
@@ -0,0 +1,142 @@
+/*
+ * 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.connect
+
+import java.io.File
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.UUID
+
+import org.apache.spark.sql.streaming.Trigger
+import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
+
+/**
+ * Interop tests for dropDuplicates key resolution across Spark Classic and
Spark Connect. Both
+ * engines now resolve the dedup keys through the same analyzer rule with a
deterministic
+ * (positional) order, so a checkpoint written by one engine must be
restartable under the other:
+ * the state-store keys (bound by position) line up and previously-seen rows
are still recognized
+ * as duplicates.
+ *
+ * The in-process [[SparkConnectServerTest]] harness exposes a Classic session
(`spark`) and a
+ * Connect session (`withSession`) sharing one SparkContext / filesystem, so
both can drive the
+ * same checkpoint directory. Input is plain CSV files (no commit markers)
read by a file source
+ * with Trigger.AvailableNow for deterministic batching; the sink is a
recoverable parquet file
+ * sink. See SPARK-57489.
+ */
+class StreamingDeduplicationConnectInteropSuite extends SparkConnectServerTest
{
+
+ private val rowSchema = StructType(
+ Seq(
+ StructField("a", IntegerType),
+ StructField("b", IntegerType),
+ StructField("c", IntegerType),
+ StructField("d", IntegerType),
+ StructField("e", IntegerType)))
+
+ /**
+ * Writes one CSV file (one batch worth of input) into `inputDir` without
any commit markers.
+ */
+ private def writeInputCsv(inputDir: File, rows: Seq[(Int, Int, Int, Int,
Int)]): Unit = {
+ inputDir.mkdirs()
+ val content =
+ rows.map { case (a, b, c, d, e) => s"$a,$b,$c,$d,$e" }.mkString("",
"\n", "\n")
+ Files.write(
+ new File(inputDir, s"${UUID.randomUUID()}.csv").toPath,
+ content.getBytes(StandardCharsets.UTF_8))
+ }
+
+ /**
+ * Starts a dropDuplicates streaming query on `session` reading `inputDir`,
processes all
+ * currently-available files against `checkpoint`, appending deduplicated
rows to the parquet
+ * sink at `outputDir`.
+ */
+ // Fully qualified on purpose: this file lives in package
org.apache.spark.sql.connect, which also
+ // defines a `SparkSession`; an import of the common type would be shadowed
by it. The
+ // common base accepts both the Classic `spark` and the Connect
`withSession` session.
+ private def runDedup(
+ session: org.apache.spark.sql.SparkSession,
+ inputDir: File,
+ checkpoint: File,
+ outputDir: File): Unit = {
+ val query = session.readStream
+ .schema(rowSchema)
+ .csv(inputDir.getCanonicalPath)
+ .dropDuplicates()
+ .writeStream
+ .format("parquet")
+ .option("path", outputDir.getCanonicalPath)
+ .option("checkpointLocation", checkpoint.getCanonicalPath)
+ .trigger(Trigger.AvailableNow())
+ .start()
+ try {
+ query.processAllAvailable()
+ } finally {
+ query.stop()
+ }
+ }
+
+ /** Reads every row the sink has emitted so far (across runs), sorted for
stable comparison. */
+ private def readOutput(outputDir: File): Seq[Seq[Int]] =
+ spark.read
+ .schema(rowSchema)
+ .parquet(outputDir.getCanonicalPath)
+ .collect()
+ .map(r => (0 until 5).map(r.getInt).toList)
+ .toSeq
+ .sortBy(_.mkString(","))
+
+ test("SPARK-57489: dropDuplicates checkpoint written by Classic restarts
under Connect") {
+ withTempDir { dir =>
+ val inputDir = new File(dir, "input")
+ val outputDir = new File(dir, "output")
+ val checkpoint = new File(dir, "checkpoint")
+
+ // Run 1 on Classic seeds the dedup state with one row.
+ writeInputCsv(inputDir, Seq((1, 2, 3, 4, 5)))
+ runDedup(spark, inputDir, checkpoint, outputDir)
+
+ // Run 2 on Connect restarts the same checkpoint with a duplicate + a
new row. The duplicate
+ // must be recognized against the Classic-created state and dropped;
only the new row is
+ // emitted. A cross-engine key-order mismatch would re-emit the
duplicate.
+ writeInputCsv(inputDir, Seq((1, 2, 3, 4, 5), (6, 7, 8, 9, 10)))
+ withSession { session =>
+ runDedup(session, inputDir, checkpoint, outputDir)
+ }
+
+ assert(readOutput(outputDir) === Seq(Seq(1, 2, 3, 4, 5), Seq(6, 7, 8, 9,
10)))
+ }
+ }
+
+ test("SPARK-57489: dropDuplicates checkpoint written by Connect restarts
under Classic") {
+ withTempDir { dir =>
+ val inputDir = new File(dir, "input")
+ val outputDir = new File(dir, "output")
+ val checkpoint = new File(dir, "checkpoint")
+
+ writeInputCsv(inputDir, Seq((1, 2, 3, 4, 5)))
+ withSession { session =>
+ runDedup(session, inputDir, checkpoint, outputDir)
+ }
+
+ writeInputCsv(inputDir, Seq((1, 2, 3, 4, 5), (6, 7, 8, 9, 10)))
+ runDedup(spark, inputDir, checkpoint, outputDir)
+
+ assert(readOutput(outputDir) === Seq(Seq(1, 2, 3, 4, 5), Seq(6, 7, 8, 9,
10)))
+ }
+ }
+}
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
index 7b734f93e595..aa0345f089fc 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
@@ -30,7 +30,7 @@ import org.apache.spark.sql.{AnalysisException, Column,
Observation, Row, SaveMo
import org.apache.spark.sql.catalyst.analysis
import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
GenericInternalRow, UnsafeProjection}
import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, LeftAnti,
LeftOuter, LeftSemi, PlanTest, RightOuter}
-import org.apache.spark.sql.catalyst.plans.logical.{CollectMetrics, Distinct,
LocalRelation, LogicalPlan}
+import org.apache.spark.sql.catalyst.plans.logical.{CollectMetrics,
Deduplicate, DeduplicateWithinWatermark, Distinct, LocalRelation, LogicalPlan}
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.classic.DataFrame
@@ -1149,10 +1149,18 @@ class SparkConnectProtoSuite extends PlanTest with
SparkConnectPlanTest {
// Compares proto plan with LogicalPlan.
private def comparePlans(connectPlan: proto.Relation, sparkPlan:
LogicalPlan): Unit = {
- def normalizeDataframeId(plan: LogicalPlan): LogicalPlan = plan transform {
+ def normalizeForComparison(plan: LogicalPlan): LogicalPlan = plan
transform {
case cm: CollectMetrics => cm.copy(dataframeId = 0)
+ // Spark Connect and Spark Classic record different
DeduplicateSpec.viaSparkClassic; it only
+ // drives the legacy key-order fallback for streaming restarts, not the
resolved plan, so
+ // ignore it when comparing the two engines.
+ case d: Deduplicate => d.copy(dedupSpec = None)
+ case d: DeduplicateWithinWatermark => d.copy(dedupSpec = None)
}
val connectAnalyzed = analyzePlan(transform(connectPlan))
- comparePlans(normalizeDataframeId(connectAnalyzed),
normalizeDataframeId(sparkPlan), false)
+ comparePlans(
+ normalizeForComparison(connectAnalyzed),
+ normalizeForComparison(sparkPlan),
+ false)
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala
index 3b8712fe0f31..93472d8304b4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala
@@ -1192,8 +1192,9 @@ class Dataset[T] private[sql](
case Distinct(u: Union) =>
Distinct(flattenUnion(u, isUnionDistinct = true))
// Only handle distinct-like 'Deduplicate', where the keys == output
- case Deduplicate(keys: Seq[Attribute], u: Union) if AttributeSet(keys)
== u.outputSet =>
- Deduplicate(keys, flattenUnion(u, isUnionDistinct = true))
+ case d @ Deduplicate(keys: Seq[Attribute], u: Union, _)
+ if AttributeSet(keys) == u.outputSet =>
+ d.copy(child = flattenUnion(u, isUnionDistinct = true))
case u: Union =>
flattenUnion(u, isUnionDistinct = false)
}
@@ -1209,7 +1210,7 @@ class Dataset[T] private[sql](
changed = true
children
// Only handle distinct-like 'Deduplicate', where the keys == output
- case Deduplicate(keys: Seq[Attribute], child @ Union(children, byName,
allowMissingCol))
+ case Deduplicate(keys: Seq[Attribute], child @ Union(children, byName,
allowMissingCol), _)
if AttributeSet(keys) == child.outputSet && isUnionDistinct &&
byName == u.byName &&
allowMissingCol == u.allowMissingCol =>
changed = true
@@ -1417,41 +1418,41 @@ class Dataset[T] private[sql](
}
/** @inheritdoc */
- def dropDuplicates(): Dataset[T] = dropDuplicates(this.columns)
+ def dropDuplicates(): Dataset[T] = withSameTypedPlan {
+ UnresolvedDeduplicate(
+ keySpec = DeduplicateAllColumnsAsKey,
+ withinWatermark = false,
+ viaSparkClassic = true,
+ child = logicalPlan)
+ }
/** @inheritdoc */
def dropDuplicates(colNames: Seq[String]): Dataset[T] = withSameTypedPlan {
- val groupCols = groupColsFromDropDuplicates(colNames)
- Deduplicate(groupCols, logicalPlan)
+ UnresolvedDeduplicate(
+ keySpec = DeduplicateKeyColumns(colNames),
+ withinWatermark = false,
+ viaSparkClassic = true,
+ child = logicalPlan)
}
/** @inheritdoc */
- def dropDuplicatesWithinWatermark(): Dataset[T] = {
- dropDuplicatesWithinWatermark(this.columns)
+ def dropDuplicatesWithinWatermark(): Dataset[T] = withSameTypedPlan {
+ // UnsupportedOperationChecker will fail the query if this is called with
batch Dataset.
+ UnresolvedDeduplicate(
+ keySpec = DeduplicateAllColumnsAsKey,
+ withinWatermark = true,
+ viaSparkClassic = true,
+ child = logicalPlan)
}
/** @inheritdoc */
def dropDuplicatesWithinWatermark(colNames: Seq[String]): Dataset[T] =
withSameTypedPlan {
- val groupCols = groupColsFromDropDuplicates(colNames)
// UnsupportedOperationChecker will fail the query if this is called with
batch Dataset.
- DeduplicateWithinWatermark(groupCols, logicalPlan)
- }
-
- private def groupColsFromDropDuplicates(colNames: Seq[String]):
Seq[Attribute] = {
- val resolver = sparkSession.sessionState.analyzer.resolver
- val allColumns = queryExecution.analyzed.output
- // SPARK-31990: We must keep `toSet.toSeq` here because of the backward
compatibility issue
- // (the Streaming's state store depends on the `groupCols` order).
- colNames.toSet.toSeq.flatMap { (colName: String) =>
- // It is possibly there are more than one columns with the same name,
- // so we call filter instead of find.
- val cols = allColumns.filter(col => resolver(col.name, colName))
- if (cols.isEmpty) {
- throw
QueryCompilationErrors.cannotResolveColumnNameAmongAttributesError(
- colName, schema.fieldNames.mkString(", "))
- }
- cols
- }
+ UnresolvedDeduplicate(
+ keySpec = DeduplicateKeyColumns(colNames),
+ withinWatermark = true,
+ viaSparkClassic = true,
+ child = logicalPlan)
}
/** @inheritdoc */
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index 6444b1ddbdc9..6e761fbe07b2 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -616,10 +616,10 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
*/
object StreamingDeduplicationStrategy extends Strategy {
override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
- case Deduplicate(keys, child) if child.isStreaming =>
+ case Deduplicate(keys, child, _) if child.isStreaming =>
StreamingDeduplicateExec(keys, planLater(child)) :: Nil
- case DeduplicateWithinWatermark(keys, child) if child.isStreaming =>
+ case DeduplicateWithinWatermark(keys, child, _) if child.isStreaming =>
StreamingDeduplicateWithinWatermarkExec(keys, planLater(child)) :: Nil
case _ => Nil
@@ -1038,7 +1038,7 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
case logical.ResolvedHint(child, hints) =>
throw SparkException.internalError(
"ResolvedHint operator should have been replaced by join hint in the
optimizer")
- case Deduplicate(_, child) if !child.isStreaming =>
+ case Deduplicate(_, child, _) if !child.isStreaming =>
throw SparkException.internalError(
"Deduplicate operator for non streaming data source should have been
replaced " +
"by aggregate in the optimizer")
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/OffsetSeq.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/OffsetSeq.scala
index 9ba99ac2c036..c55fda596ff9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/OffsetSeq.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/OffsetSeq.scala
@@ -205,7 +205,8 @@ object OffsetSeqMetadata extends Logging {
PRUNE_FILTERS_CAN_PRUNE_STREAMING_SUBPLAN,
STREAMING_STATE_STORE_ENCODING_FORMAT,
STATE_STORE_ROW_CHECKSUM_ENABLED, PROTOBUF_EXTENSIONS_SUPPORT_ENABLED,
ENABLE_STREAMING_SOURCE_EVOLUTION,
- STATEFUL_OPERATOR_ALWAYS_NULLABLE_OUTPUT
+ STATEFUL_OPERATOR_ALWAYS_NULLABLE_OUTPUT,
+ DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER
)
/**
@@ -256,7 +257,8 @@ object OffsetSeqMetadata extends Logging {
STATE_STORE_ROCKSDB_MERGE_OPERATOR_VERSION.key -> "1",
PROTOBUF_EXTENSIONS_SUPPORT_ENABLED.key -> "false",
ENABLE_STREAMING_SOURCE_EVOLUTION.key -> "false",
- STATEFUL_OPERATOR_ALWAYS_NULLABLE_OUTPUT.key -> "false"
+ STATEFUL_OPERATOR_ALWAYS_NULLABLE_OUTPUT.key -> "false",
+ DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER.key -> "false"
)
def readValue[T](metadataLog: OffsetSeqMetadataBase, confKey:
ConfigEntry[T]): String = {
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 8fc6c718c5a0..17d852f3ed92 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
@@ -29,7 +29,7 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.{SparkException, SparkIllegalArgumentException,
SparkIllegalStateException}
import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.LogKeys._
-import org.apache.spark.sql.catalyst.analysis.V2TableReference
+import org.apache.spark.sql.catalyst.analysis.{ResolveDeduplicate,
V2TableReference}
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
CurrentBatchTimestamp, CurrentDate, CurrentTimestamp,
FileSourceMetadataAttribute, LocalTimestamp}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Deduplicate,
DeduplicateWithinWatermark, Distinct, FlatMapGroupsInPandasWithState,
FlatMapGroupsWithState, GlobalLimit, Join, LeafNode, LocalRelation,
LogicalPlan, Project, StreamSourceAwareLogicalPlan, TransformWithState,
TransformWithStateInPySpark}
@@ -222,8 +222,31 @@ class MicroBatchExecution(
}
}.getOrElse(sparkSessionForStream.sessionState.conf.enableStreamingSourceEvolution)
+ // SPARK-57489: dropDuplicates / dropDuplicatesWithinWatermark resolve
their keys with the
+ // deterministic order by default, but streaming deduplication binds
state-store keys by
+ // position. A query restored from a checkpoint that predates this change
must keep its original
+ // (legacy) key order, so recompute the keys from the recipe carried on
the resolved node using
+ // the order pinned in the offset log (mirroring how `enforceNamed` is
read above). For a new
+ // query the offset log has no entry yet, so the session value is used and
pinned at batch 0.
+ val streamConf = sparkSessionForStream.sessionState.conf
+ val orderDeterministically = initialLatestOffsetSeq.flatMap { case (_,
offsetSeq) =>
+ offsetSeq.metadataOpt.flatMap { metadata =>
+ OffsetSeqMetadata.readValueOpt(metadata,
SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER)
+ .map(_.toBoolean)
+ }
+
}.getOrElse(streamConf.getConf(SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER))
+ val dedupResolver = sparkSessionForStream.sessionState.analyzer.resolver
+ 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))
+ }
+
import
org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
- val _logicalPlan = analyzedPlan.transform {
+ val _logicalPlan = planWithDedupKeys.transform {
case streamingRelation @ StreamingRelation(
dataSourceV1, sourceName, output, sourceIdentifyingName) =>
toExecutionRelationMap.getOrElseUpdate(streamingRelation, {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationFallbackSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationFallbackSuite.scala
new file mode 100644
index 000000000000..f44b8c5787b1
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingDeduplicationFallbackSuite.scala
@@ -0,0 +1,143 @@
+/*
+ * 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 org.apache.hadoop.fs.Path
+
+import org.apache.spark.sql.catalyst.streaming.InternalOutputModes._
+import org.apache.spark.sql.execution.streaming.checkpointing.{OffsetMap,
OffsetSeq, OffsetSeqLog, OffsetSeqMetadata, OffsetSeqMetadataV2}
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * End-to-end tests for the dropDuplicates key-order fallback. A streaming
query started before
+ * [[SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER]] existed has no such
entry in its offset log;
+ * on restart it must fall back to the legacy key order so the persisted
state-store keys (bound by
+ * position) still align. The session default stays true (deterministic), so
these tests prove the
+ * fallback is driven by the offset log, not the session. See SPARK-57489.
+ */
+class StreamingDeduplicationFallbackSuite extends StateStoreMetricsTest {
+ import testImplicits._
+
+ private val confKey = SQLConf.DROP_DUPLICATES_DETERMINISTIC_KEY_ORDER.key
+
+ // A projection AFTER dropDuplicates is the case that matters: by the time
the streaming query is
+ // started the dedup keys are already resolved (with the recipe attached to
the Deduplicate node),
+ // so the offset-log fallback must recompute them from the recipe rather
than rely on an
+ // unresolved node surviving until query start.
+ private def dedupAllColumns(input: MemoryStream[(Int, Int, Int, Int, Int)]) =
+ input.toDS().toDF("a", "b", "c", "d", "e").dropDuplicates().select("a",
"b", "c", "d", "e")
+
+ /** Rewrites batch `batchId`'s offset-log metadata, applying `f` to its
persisted conf map. */
+ private def rewriteOffsetLogConf(
+ checkpoint: File,
+ batchId: Long)(f: Map[String, String] => Map[String, String]): Unit = {
+ val offsetsPath = checkpoint.getCanonicalPath + "/offsets"
+ val offsetLog = new OffsetSeqLog(spark, offsetsPath)
+ val offsetSeq = offsetLog.get(batchId).getOrElse {
+ fail(s"Missing offset log entry for batch $batchId")
+ }
+ val updatedMetadata = offsetSeq.metadataOpt.map {
+ case m: OffsetSeqMetadata => m.copy(conf = f(m.conf))
+ case m: OffsetSeqMetadataV2 => m.copy(conf = f(m.conf))
+ }
+ val updatedOffsetSeq = offsetSeq match {
+ case o: OffsetSeq =>
+ o.copy(metadataOpt =
updatedMetadata.map(_.asInstanceOf[OffsetSeqMetadata]))
+ case o: OffsetMap =>
+ // OffsetMap's metadata is required (not optional), so fall back to
the original when the
+ // rewrite produced none.
+ o.copy(metadata =
+
updatedMetadata.map(_.asInstanceOf[OffsetSeqMetadataV2]).getOrElse(o.metadata))
+ }
+ // Offset log entries are immutable once written; delete before re-adding.
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val batchFile = new Path(offsetsPath, batchId.toString)
+ batchFile.getFileSystem(hadoopConf).delete(batchFile, false)
+ assert(
+ offsetLog.add(batchId, updatedOffsetSeq),
+ s"Failed to rewrite offset log entry for batch $batchId")
+ }
+
+ private def readPinnedConf(checkpoint: File, batchId: Long): Option[String]
= {
+ val offsetLog = new OffsetSeqLog(spark, checkpoint.getCanonicalPath +
"/offsets")
+
offsetLog.get(batchId).flatMap(_.metadataOpt.map(_.conf)).flatMap(_.get(confKey))
+ }
+
+ test("SPARK-57489: existing checkpoint without the conf falls back to the
legacy key order") {
+ withTempDir { checkpoint =>
+ // The same MemoryStream is reused across both runs so source offsets
advance across the
+ // restart (a fresh stream would reset to an already-committed offset
and process nothing).
+ val input = MemoryStream[(Int, Int, Int, Int, Int)]
+
+ // Run 1 with the deterministic order disabled writes the state-store
keys in the legacy
+ // (origin-engine) order; confirm the offset log captured the legacy
value.
+ withSQLConf(confKey -> "false") {
+ testStream(dedupAllColumns(input), Append)(
+ StartStream(checkpointLocation = checkpoint.getCanonicalPath),
+ AddData(input, (1, 2, 3, 4, 5)),
+ CheckLastBatch((1, 2, 3, 4, 5)),
+ StopStream)
+ }
+ assert(readPinnedConf(checkpoint, batchId = 0).contains("false"))
+
+ // Simulate a checkpoint created before this change: drop the conf from
the offset log.
+ rewriteOffsetLogConf(checkpoint, batchId = 0)(_ - confKey)
+
+ // Restart with the deterministic session default (true). If the absent
offset-log entry were
+ // NOT pinned to the legacy order, the new run would encode keys in
schema order while the
+ // restored state holds them in Set order; the previously-seen row would
no longer be
+ // recognized as a duplicate and would be (incorrectly) emitted, failing
CheckLastBatch.
+ testStream(dedupAllColumns(input), Append)(
+ StartStream(checkpointLocation = checkpoint.getCanonicalPath),
+ AddData(input, (1, 2, 3, 4, 5), (6, 7, 8, 9, 10)),
+ CheckLastBatch((6, 7, 8, 9, 10)),
+ StopStream)
+ }
+ }
+
+ test("SPARK-57489: newly started query pins the deterministic order and
keeps it on restart") {
+ withTempDir { checkpoint =>
+ val input = MemoryStream[(Int, Int, Int, Int, Int)]
+
+ // Session default is deterministic (true); a newly started query must
pin "true" at batch 0.
+ testStream(dedupAllColumns(input), Append)(
+ StartStream(checkpointLocation = checkpoint.getCanonicalPath),
+ AddData(input, (1, 2, 3, 4, 5)),
+ CheckLastBatch((1, 2, 3, 4, 5)),
+ StopStream)
+ assert(readPinnedConf(checkpoint, batchId = 0).contains("true"))
+
+ // Restart with the session default flipped to the legacy order. The
"true" pinned in the
+ // offset log must override the session value, so the dedup state is
still honored under the
+ // deterministic order. (If the session value were used instead, the
keys would be encoded in
+ // the legacy order while the restored state holds them in the
deterministic order, so the
+ // previously-seen row would no longer be recognized as a duplicate and
would be
+ // (incorrectly) emitted, failing CheckLastBatch.)
+ withSQLConf(confKey -> "false") {
+ testStream(dedupAllColumns(input), Append)(
+ StartStream(checkpointLocation = checkpoint.getCanonicalPath),
+ AddData(input, (1, 2, 3, 4, 5), (6, 7, 8, 9, 10)),
+ CheckLastBatch((6, 7, 8, 9, 10)),
+ StopStream)
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
