This is an automated email from the ASF dual-hosted git repository.
szehon-ho 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 6eb0fb8e7f83 [SPARK-60094][SDP] Gate AutoCDC ignore-null updates
6eb0fb8e7f83 is described below
commit 6eb0fb8e7f83d585ae5720822607ef0582ee60a7
Author: AnishMahto <[email protected]>
AuthorDate: Thu Oct 8 18:02:54 2026 -0700
[SPARK-60094][SDP] Gate AutoCDC ignore-null updates
### What changes were proposed in this pull request?
Until AutoCDC ignore-null is fully built E2E for both SCD1 and SCD2, we
should explicitly gate its usage.
In ignore-null's current partial implementation state, if it runs then we
may produce partial results that get persisted against existing AutoCDC
target/aux tables.
Both the current (partially complete) implementation and API support for
ignore-null is merged into Spark master, and is planned to go out as of Spark
4.4. Once the implementation is complete, this gating can be removed.
**Gating is intentionally done as a `GraphValidation` to avoid reverting
ignore-null Python/SQL API and tests, which are currently unreleased and only
in Spark master.**
The consequence is if a Spark release was made today, the ignore-null API
would be released but usage invalidated at execution time.
### Why are the changes needed?
Gate incomplete AutoCDC ignore-null implementation.
### Does this PR introduce _any_ user-facing change?
No, currently the user facing ignore-null APIs are unreleased.
### How was this patch tested?
Unit tests.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Opus 5.5
Closes #59309 from AnishMahto/SPARK-60094-autocdc-gate-ignore-null-updates.
Authored-by: AnishMahto <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit 26445bf61aa36a7001e0be4afebc76d691a24661)
Signed-off-by: Szehon Ho <[email protected]>
---
.../connect/pipelines/PythonPipelineSuite.scala | 12 +++----
.../spark/sql/pipelines/graph/DataflowGraph.scala | 1 +
.../sql/pipelines/graph/GraphValidations.scala | 13 +++++++
.../graph/ConnectInvalidPipelineSuite.scala | 40 +++++++++++++++++++++-
4 files changed, 58 insertions(+), 8 deletions(-)
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
index c85820f0079a..d4edee481f5f 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
@@ -972,15 +972,13 @@ class PythonPipelineSuite
}
/**
- * Builds the graph and runs it through resolution and validation, returning
the resolved
- * [[AutoCdcMergeFlow]]. Unlike [[buildAutoCdcFlow]], this exercises the
graph analysis that
- * rejects invalid change args (e.g. an ignore-null column that is also a
key), so a test that
- * asserts on the resolved flow is guaranteed to describe a usable
configuration.
+ * Builds and resolves the graph, returning the resolved
[[AutoCdcMergeFlow]]. Unlike
+ * [[buildAutoCdcFlow]], this exercises the flow analysis that rejects
invalid change args (e.g.
+ * an ignore-null column that is also a key), so a test that asserts on the
resolved flow is
+ * guaranteed to describe a configuration the flow accepts.
*/
private def resolveAutoCdcMergeFlow(pipelineSource: String):
AutoCdcMergeFlow = {
- val graph = buildGraph(pipelineSource)
- .resolve(sessionCaseSensitive)
- .validate(sessionCaseSensitive)
+ val graph = buildGraph(pipelineSource).resolve(sessionCaseSensitive)
graph.resolvedFlow(graphIdentifier("target")) match {
case f: AutoCdcMergeFlow => f
case other => fail(s"Expected an AutoCdcMergeFlow, got: $other")
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala
index cd1dbf63ccdf..4ff45e330125 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala
@@ -249,6 +249,7 @@ case class DataflowGraph(
validateEveryDatasetHasFlow()
validateTablesAreResettable()
validateFlowStreamingness()
+ validateNoAutoCdcIgnoreNullSelection()
}.failed
/**
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala
index 5b4eaba4e709..d017a81d5b1d 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala
@@ -83,6 +83,19 @@ trait GraphValidations extends Logging {
case _ => false
}
+ /** Rejects any AutoCDC flow with an ignore-null selection, since neither
SCD type supports it. */
+ protected[graph] def validateNoAutoCdcIgnoreNullSelection(): Unit = {
+ resolvedFlows
+ .collectFirst {
+ case f: AutoCdcMergeFlow if f.changeArgs.ignoreNullSelection.isDefined
=> f
+ }
+ .foreach { flow =>
+ throw new UnsupportedOperationException(
+ s"AutoCDC flow ${flow.displayName} specifies ignore-null updates,
which are not yet " +
+ "supported.")
+ }
+ }
+
/**
* Validate that each resolved flow is correctly either a streaming flow or
non-streaming flow,
* depending on the flow type (ex. once flow vs non-once flow) and the
dataset type the flow
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
index 172d8354932b..5fd1190ca6c3 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.pipelines.graph
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
-import org.apache.spark.sql.pipelines.autocdc.{ChangeArgs, ScdType,
UnqualifiedColumnName}
+import org.apache.spark.sql.pipelines.autocdc.{ChangeArgs, ColumnSelection,
ScdType, UnqualifiedColumnName}
import org.apache.spark.sql.pipelines.utils.{PipelineTest,
TestGraphRegistrationContext}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{IntegerType, StructType}
@@ -772,6 +772,44 @@ class ConnectInvalidPipelineSuite extends PipelineTest
with SharedSparkSession {
)
}
+ gridTest("AutoCDC flow with ignore-null updates is rejected as unsupported")(
+ Seq(ScdType.Type1, ScdType.Type2)
+ ) { scdType =>
+ val session = spark
+ import session.implicits._
+
+ val graph = new TestGraphRegistrationContext(spark) {
+ val cdcEvents = MemoryStream[Int].toDF().select($"value" as "id",
$"value" as "seq")
+ registerTable("target")
+ registerFlow(
+ AutoCdcFlow(
+ identifier = fullyQualifiedIdentifier("auto_cdc_flow"),
+ destinationIdentifier = fullyQualifiedIdentifier("target"),
+ func = dfFlowFunc(cdcEvents),
+ queryContext = QueryContext(
+ currentCatalog =
Some(TestGraphRegistrationContext.DEFAULT_CATALOG),
+ currentDatabase =
Some(TestGraphRegistrationContext.DEFAULT_DATABASE)
+ ),
+ origin = QueryOrigin.empty,
+ changeArgs = ChangeArgs(
+ keys = Seq(UnqualifiedColumnName("id")),
+ sequencing = $"seq",
+ storedAsScdType = scdType,
+ ignoreNullSelection =
Some(ColumnSelection.ExcludeColumns(Seq.empty))
+ )
+ )
+ )
+ }.resolveToDataflowGraph()
+
+ val ex = intercept[UnsupportedOperationException] {
+ validateGraph(graph)
+ }
+ val flowName = fullyQualifiedIdentifier("auto_cdc_flow").unquotedString
+ assert(
+ ex.getMessage ==
+ s"AutoCDC flow $flowName specifies ignore-null updates, which are not
yet supported.")
+ }
+
test("DUPLICATE_GRAPH_ELEMENT: duplicate graph element identifiers") {
checkError(
exception = intercept[AnalysisException] {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
