gengliangwang commented on code in PR #56122:
URL: https://github.com/apache/spark/pull/56122#discussion_r3306861423
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowPlanner.scala:
##########
@@ -79,6 +80,18 @@ class FlowPlanner(
s"streaming flow ${sf.identifier}
(${flow.destinationIdentifier})"
)
}
+ case acmf: AutoCdcMergeFlow if acmf.changeArgs.storedAsScdType ==
ScdType.Type1 =>
+ val flowMetadata = FlowSystemMetadata(updateContext, acmf, graph)
+ new Scd1MergeStreamingWrite(
+ identifier = acmf.identifier,
+ flow = acmf,
+ graph = graph,
+ updateContext = updateContext,
+ checkpointPath = flowMetadata.latestCheckpointLocation,
+ trigger = triggerFor(acmf),
+ destination = output.asInstanceOf[Table],
Review Comment:
The sibling `StreamingFlow` case (lines 54–82) handles unsupported outputs
with `output match { case o: Table => …; case s: Sink => …; case _ => throw
UnsupportedOperationException(…) }`. The new `AutoCdcMergeFlow` case bypasses
that and will surface a bare `ClassCastException` if a future change makes an
AutoCDC flow's output anything other than `Table`. Mirroring the StreamingFlow
case keeps error messages aligned. (The pre-existing `CompleteFlow` case at
line 49 has the same flaw and is the precedent being copied — worth fixing both
in the same place.)
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -303,6 +303,25 @@ object DatasetManager extends Logging {
context.spark.sql(s"TRUNCATE TABLE ${table.identifier.quotedString}")
}
+ if (isFullRefresh) {
+ // On full refresh, drop the AutoCDC auxiliary state associated with
this table (if any) so
+ // that stale delete-tracking data and table properties are not carried
forward into the new
+ // table generation. We unconditionally issue the DROP for every
fully-refreshed target; for
+ // non-AutoCDC tables this is a no-op because
[[AutoCdcAuxiliaryTable.identifier]] derives
+ // its name from [[AutoCdcReservedNames.prefix]], which is reserved
across AutoCDC and
+ // therefore cannot collide with a user-managed table.
+
+ // Intentionally DROP and not TRUNCATE for two reasons; First, the
auxiliary table may
+ // contain table properties that represent stateful information (ex. SCD
key count) that
+ // should not be carried forward on a full refresh. Second, the
auxiliary table is an
+ // internal table and not part of the dataflow graph. That means it does
not go through
+ // schema evolution like other tables and hence on a full refresh, we
should explicitly
+ // drop the existing auxiliary table schema so it can be recomputed.
+
+ val auxiliaryTableId = AutoCdcAuxiliaryTable.identifier(table.identifier)
+ context.spark.sql(s"DROP TABLE IF EXISTS
${auxiliaryTableId.quotedString}")
Review Comment:
**Layer-mixing / asymmetry.** `DatasetManager` is otherwise
flow-type-agnostic, but this block reaches into AutoCDC's identifier scheme to
drop a feature-specific internal table. Meanwhile, aux-table *creation* lives
in `Scd1MergeStreamingWrite.startStream`. The asymmetry will compound when
another internal-state table (SCD2, future formats) lands — each will need a
matching drop sprinkled here.
Worth considering a per-flow `onFullRefresh` hook (or a registry of "managed
internal tables per target") so the create/drop pair stays in the feature
module. Acceptable in this PR if the team explicitly accepts this as the single
exception for now, but please leave that decision visible in the comment or PR
description.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
+ /** The spark session the AutoCDC flow is going to be planned in. */
+ protected def spark: SparkSession
+
+ /** The destination (target) table entity the AutoCDC flow will be writing
to. */
+ protected def destination: Table
+
+ /** The AutoCDC flow's [[ChangeArgs]] (keys, sequencing, columnSelection,
...). */
+ protected def changeArgs: ChangeArgs
+
+ /** Full schema of the auxiliary table for this SCD type. */
+ protected def auxiliaryTableSchema: StructType
+
+ // Eagerly validate at construction time that the destination supports
row-level ops.
+ requireDestinationSupportsRowLevelOps()
+
+ /**
+ * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
+ * and return its [[TableIdentifier]].
+ */
+ protected def createAuxiliaryTableIfNotExists(spark: SparkSession):
TableIdentifier = {
+ val auxIdent = AutoCdcAuxiliaryTable.identifier(destination.identifier)
+ // The auxiliary table inherits the target's format so MERGE semantics
line up. When the
+ // target's format is unspecified (None), omit the USING clause and fall
back to the
+ // session's default source provider.
+ val usingClause = destination.format.map(fmt => s"USING
$fmt").getOrElse("")
+ spark.sql(
+ s"""CREATE TABLE IF NOT EXISTS
+ |${auxIdent.quotedString}
+ |(${auxiliaryTableSchema.toDDL}) $usingClause""".stripMargin
+ )
+ auxIdent
+ }
+
+ /**
+ * Validate that the target table's underlying connector implements
+ * [[SupportsRowLevelOperations]], which is the V2 connector contract for
MERGE/UPDATE/DELETE
+ * with rewrite - all operations that the AutoCDC transformation executes.
+ */
+ private def requireDestinationSupportsRowLevelOps(): Unit = {
+ val (catalog, v2Identifier) = resolveTableCatalog(spark,
destination.identifier)
+ val destinationTable = catalog.loadTable(v2Identifier)
+
+ if (!destinationTable.isInstanceOf[SupportsRowLevelOperations]) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE",
+ messageParameters = Map(
+ "tableName" -> destination.identifier.quotedString,
+ "format" -> destination.format.getOrElse("<session default>")
+ )
+ )
+ }
+ }
+
+ private def resolveTableCatalog(
+ spark: SparkSession,
+ ident: TableIdentifier): (TableCatalog, Identifier) = {
+ val catalogManager = spark.sessionState.catalogManager
+ val catalog = ident.catalog
+ .map(catalogManager.catalog)
+ .getOrElse(catalogManager.currentCatalog)
+ .asInstanceOf[TableCatalog]
+ val namespace = ident.database.getOrElse(
+ throw SparkException.internalError(
+ s"Cannot resolve table identifier ${ident.quotedString}: namespace is
unspecified."
+ )
+ )
+ (catalog, Identifier.of(Array(namespace), ident.table))
+ }
+}
+
+/**
+ * A [[StreamingFlowExecution]] that applies a CDC event stream to a target
[[Table]] via
+ * SCD Type 1 MERGE semantics.
+ */
+class Scd1MergeStreamingWrite(
+ val identifier: TableIdentifier,
+ val flow: AutoCdcMergeFlow,
+ val graph: DataflowGraph,
+ val updateContext: PipelineUpdateContext,
+ val checkpointPath: String,
+ val trigger: Trigger,
+ val destination: Table,
+ val sqlConf: Map[String, String]
+) extends StreamingFlowExecution with AutoCdcMergeWriteMixin {
+
+ override def getOrigin: QueryOrigin = flow.origin
+
+ override protected def changeArgs: ChangeArgs = flow.changeArgs
+
+ /**
+ * Resolved Spark [[DataType]] of the sequencing expression.
+ */
+ private def sequencingType: DataType =
+ flow.df.select(flow.changeArgs.sequencing).schema.head.dataType
Review Comment:
**Duplication.** `AutoCdcMergeFlow.sequencingType` (`Flow.scala:275-276`)
already computes exactly this —
`df.select(changeArgs.sequencing).schema.head.dataType` — and caches it as a
`private val`. Re-deriving it here means two Catalyst analyses for the same
expression, and because this is a `def`, every access inside the `foreachBatch`
closure re-runs the analysis on each microbatch start (the closure is rebuilt
per batch).
Suggest widening the flow's `sequencingType` visibility (e.g.
`private[graph]`) and reusing it. At minimum, make this a `lazy val`.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
+ /** The spark session the AutoCDC flow is going to be planned in. */
+ protected def spark: SparkSession
+
+ /** The destination (target) table entity the AutoCDC flow will be writing
to. */
+ protected def destination: Table
+
+ /** The AutoCDC flow's [[ChangeArgs]] (keys, sequencing, columnSelection,
...). */
+ protected def changeArgs: ChangeArgs
+
+ /** Full schema of the auxiliary table for this SCD type. */
+ protected def auxiliaryTableSchema: StructType
+
+ // Eagerly validate at construction time that the destination supports
row-level ops.
+ requireDestinationSupportsRowLevelOps()
+
+ /**
+ * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
+ * and return its [[TableIdentifier]].
+ */
+ protected def createAuxiliaryTableIfNotExists(spark: SparkSession):
TableIdentifier = {
+ val auxIdent = AutoCdcAuxiliaryTable.identifier(destination.identifier)
+ // The auxiliary table inherits the target's format so MERGE semantics
line up. When the
+ // target's format is unspecified (None), omit the USING clause and fall
back to the
+ // session's default source provider.
+ val usingClause = destination.format.map(fmt => s"USING
$fmt").getOrElse("")
+ spark.sql(
+ s"""CREATE TABLE IF NOT EXISTS
+ |${auxIdent.quotedString}
+ |(${auxiliaryTableSchema.toDDL}) $usingClause""".stripMargin
+ )
+ auxIdent
+ }
+
+ /**
+ * Validate that the target table's underlying connector implements
+ * [[SupportsRowLevelOperations]], which is the V2 connector contract for
MERGE/UPDATE/DELETE
+ * with rewrite - all operations that the AutoCDC transformation executes.
+ */
+ private def requireDestinationSupportsRowLevelOps(): Unit = {
+ val (catalog, v2Identifier) = resolveTableCatalog(spark,
destination.identifier)
+ val destinationTable = catalog.loadTable(v2Identifier)
+
+ if (!destinationTable.isInstanceOf[SupportsRowLevelOperations]) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE",
+ messageParameters = Map(
+ "tableName" -> destination.identifier.quotedString,
+ "format" -> destination.format.getOrElse("<session default>")
+ )
+ )
+ }
+ }
+
+ private def resolveTableCatalog(
+ spark: SparkSession,
+ ident: TableIdentifier): (TableCatalog, Identifier) = {
+ val catalogManager = spark.sessionState.catalogManager
+ val catalog = ident.catalog
+ .map(catalogManager.catalog)
+ .getOrElse(catalogManager.currentCatalog)
+ .asInstanceOf[TableCatalog]
Review Comment:
`asInstanceOf[TableCatalog]` produces a cryptic `ClassCastException` if the
user's current catalog only implements e.g. `FunctionCatalog` or
`SupportsNamespaces`. Prefer:
```scala
val catalog = currentCatalog match {
case t: TableCatalog => t
case other => throw new AnalysisException(
errorClass = "...",
messageParameters = Map("catalogName" -> other.name)
)
}
```
so the user gets a typed error mentioning the actual catalog.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala:
##########
@@ -391,7 +401,12 @@ case class Scd1BatchProcessor(
// New key: only insert upserts; deletes for absent keys are no-ops for
the target table
// merge, and instead would have been inserted as tombstones into the
auxiliary table.
.whenNotMatched(microbatchDeleteVersionField.isNull)
- .insertAll()
+ // When inserting a brand new row for a new key, construct column
mappings from microbatch.
+ // The microbatch's columns may be a strict subset of the target's
columns -- e.g. the user
+ // narrowed `column_list` between runs, or the source DF dropped a
column. The target's
+ // columns can never be a strict subset of the microbatch's however,
because SDP's schema
Review Comment:
```suggestion
// columns can never be a strict subset of the microbatch's, however,
because SDP's schema
```
nit: missing commas around the parenthetical "however".
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcGraphExecutionTestMixin.scala:
##########
@@ -0,0 +1,213 @@
+/*
+ * 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.pipelines.graph
+
+import org.scalatest.{BeforeAndAfterEach, Suite}
+
+import org.apache.spark.SparkThrowable
+import org.apache.spark.sql.{Column, Row}
+import
org.apache.spark.sql.connector.catalog.SharedTablesInMemoryRowLevelOperationTableCatalog
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.pipelines.autocdc.{
+ ChangeArgs,
+ ColumnSelection,
+ Scd1BatchProcessor,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.common.RunState
+import org.apache.spark.sql.pipelines.logging.RunProgress
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Shared helpers for AutoCDC end-to-end graph-execution test suites.
+ */
+trait AutoCdcGraphExecutionTestMixin extends BeforeAndAfterEach {
+ self: Suite with ExecutionTest with SharedSparkSession =>
+
+ /** v2 catalog name registered for AutoCDC E2E tests. Tests qualify tables
as `cat.ns1.t`. */
+ protected val catalog: String = "cat"
+
+ /** Namespace under [[catalog]] used by AutoCDC E2E tests. */
+ protected val namespace: String = "ns1"
+
+ override protected def beforeEach(): Unit = {
+ super.beforeEach()
+ spark.conf.set(
+ s"spark.sql.catalog.$catalog",
+ classOf[SharedTablesInMemoryRowLevelOperationTableCatalog].getName
+ )
+ // Disable per-flow retries so failure-path tests (e.g. INCOMPATIBLE_DATA)
surface the
+ // AnalysisException after the first attempt instead of going through the
default 2 retries,
+ // which would otherwise emit duplicate FAILED events and inflate test
runtime without
+ // changing the asserted outcome.
+ spark.conf.set(SQLConf.PIPELINES_MAX_FLOW_RETRY_ATTEMPTS.key, "0")
+ spark.sql(s"CREATE NAMESPACE IF NOT EXISTS $catalog.$namespace")
+ }
+
+ override protected def afterEach(): Unit = {
+ SharedTablesInMemoryRowLevelOperationTableCatalog.reset()
+ spark.sessionState.catalogManager.reset()
+ spark.sessionState.conf.unsetConf(s"spark.sql.catalog.$catalog")
+
spark.sessionState.conf.unsetConf(SQLConf.PIPELINES_MAX_FLOW_RETRY_ATTEMPTS.key)
+ super.afterEach()
+ }
+
+ /**
+ * Run a pipeline to completion. If any flow emitted a [[RunProgress]] event
with state
+ * [[RunState.FAILED]], collect every error from the event buffer and throw
a single
+ * exception listing them, so that test failures surface meaningful stack
traces instead of
+ * generic "test exited normally but flow failed" errors.
+ */
+ protected def runPipeline(ctx: TestGraphRegistrationContext): Unit = {
+ val updateCtx = TestPipelineUpdateContext(spark, ctx.toDataflowGraph,
storageRoot)
+ updateCtx.pipelineExecution.runPipeline()
+ updateCtx.pipelineExecution.awaitCompletion()
+
+ if (updateCtx.eventBuffer.getEvents.exists(_.details ==
RunProgress(RunState.FAILED))) {
+ val errors = updateCtx.eventBuffer.getEvents.flatMap(_.error)
+ val ex = new RuntimeException(
+ s"Pipeline run failed with ${errors.size} error(s):\n" +
+ errors.map { e =>
+ val stackSnippet = e.getStackTrace
+ .map(f => s" at $f")
+ .mkString("\n")
+ s" ${e.getClass.getSimpleName}: ${e.getMessage}\n$stackSnippet"
+ }.mkString("\n")
+ )
+ errors.foreach(ex.addSuppressed)
+ throw ex
+ }
+ }
+
+ /**
+ * Walk every [[Throwable]] reachable from `failure` via
[[Throwable#getSuppressed]] and
+ * [[Throwable#getCause]] for the first [[SparkThrowable]] whose
Review Comment:
```suggestion
* [[Throwable#getCause]], searching for the first [[SparkThrowable]] whose
```
nit: the sentence is missing a verb between `getCause` and `for the first` —
reads as a broken clause.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ ChangeArgs,
+ ColumnSelection,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Smoke tests for AutoCDC SCD type 1 flows running within a single pipeline:
one
+ * [[DataflowGraph]] / [[TestPipelineUpdateContext]] executes one or more
AutoCDC flows,
+ * and the target table contents are asserted at the end. Multi-pipeline
scenarios (where
+ * multiple pipelines write to the same target) live in
[[AutoCdcScd1MultiPipelineSuite]].
+ */
+class AutoCdcScd1SinglePipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ test("an upsert event lands a new row in an empty target table") {
+ val session = spark
+ import session.implicits._
+
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 1L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+
+ runPipeline(ctx)
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 1L, cdcMeta(None, Some(1L))))
+ )
+ }
+
+ test("consecutive upsert, delete, and re-upsert events for the same key in
one run " +
+ "converge to the latest event") {
+ val session = spark
+ import session.implicits._
+
+ // Target schema deliberately omits `is_delete`: the source carries it as
a control
+ // column, drives the deleteCondition, and is excluded from the target
projection.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, Long, Boolean)]
+ stream.addData(
+ (1, "alice", 1L, false), // initial upsert
+ (1, "alice", 2L, true), // delete
+ (1, "alice2", 3L, false) // reinsert
+ )
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version",
"is_delete")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ deleteCondition = Some(functions.col("is_delete") === true),
+ columnSelection = Some(ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("is_delete"))
+ ))
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // After all three events at seqs 1, 2, 3: row "alice2" wins as the
highest-sequenced
+ // upsert; the delete at seq=2 is bounded by the seq=3 upsert.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice2", 3L, cdcMeta(None, Some(3L))))
+ )
+ }
+
+ test("two AutoCDC flows targeting separate tables in one pipeline produce
independent " +
+ "results") {
+ val session = spark
+ import session.implicits._
+
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_a " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+ )
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_b " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+ )
+
+ val streamA = MemoryStream[(Int, Long)]
+ val streamB = MemoryStream[(Int, Long)]
+ streamA.addData((1, 1L), (2, 1L))
+ streamB.addData((10, 1L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("t_a", catalog = Some(catalog), database = Some(namespace))
+ registerTable("t_b", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "flow_a",
+ target = "t_a",
+ query = dfFlowFunc(streamA.toDF().toDF("id", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ registerFlow(autoCdcFlow(
+ name = "flow_b",
+ target = "t_b",
+ query = dfFlowFunc(streamB.toDF().toDF("id", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+ runPipeline(ctx)
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.t_a"),
+ Seq(Row(1, 1L, cdcMeta(None, Some(1L))), Row(2, 1L, cdcMeta(None,
Some(1L))))
+ )
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.t_b"),
+ Seq(Row(10, 1L, cdcMeta(None, Some(1L))))
+ )
+ assert(spark.catalog.tableExists(auxTableNameFor("t_a")))
+ assert(spark.catalog.tableExists(auxTableNameFor("t_b")))
+ }
+
+ test("an AutoCDC flow targeting a table whose format does not support
row-level " +
+ "operations fails with AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE") {
+ val session = spark
+ import session.implicits._
+
+ // Intentionally use a non-merge compatible catalog, whose default table
format is parquet.
Review Comment:
```suggestion
// Intentionally use a non-merge-compatible catalog, whose default table
format is parquet.
```
nit: missing hyphen in the compound modifier.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SinglePipelineSuite.scala:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ ChangeArgs,
+ ColumnSelection,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Smoke tests for AutoCDC SCD type 1 flows running within a single pipeline:
one
+ * [[DataflowGraph]] / [[TestPipelineUpdateContext]] executes one or more
AutoCDC flows,
+ * and the target table contents are asserted at the end. Multi-pipeline
scenarios (where
+ * multiple pipelines write to the same target) live in
[[AutoCdcScd1MultiPipelineSuite]].
+ */
+class AutoCdcScd1SinglePipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ test("an upsert event lands a new row in an empty target table") {
+ val session = spark
+ import session.implicits._
+
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 1L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")
+ ))
+ }
+
+ runPipeline(ctx)
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 1L, cdcMeta(None, Some(1L))))
+ )
+ }
+
+ test("consecutive upsert, delete, and re-upsert events for the same key in
one run " +
+ "converge to the latest event") {
+ val session = spark
+ import session.implicits._
+
+ // Target schema deliberately omits `is_delete`: the source carries it as
a control
+ // column, drives the deleteCondition, and is excluded from the target
projection.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, String, Long, Boolean)]
+ stream.addData(
+ (1, "alice", 1L, false), // initial upsert
+ (1, "alice", 2L, true), // delete
+ (1, "alice2", 3L, false) // reinsert
+ )
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version",
"is_delete")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ deleteCondition = Some(functions.col("is_delete") === true),
+ columnSelection = Some(ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("is_delete"))
+ ))
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // After all three events at seqs 1, 2, 3: row "alice2" wins as the
highest-sequenced
+ // upsert; the delete at seq=2 is bounded by the seq=3 upsert.
Review Comment:
```suggestion
// upsert; the delete at seq=2 is superseded by the seq=3 upsert.
```
nit: "bounded" doesn't describe the semantics here — the delete is
overridden / superseded by the later upsert.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
+ /** The spark session the AutoCDC flow is going to be planned in. */
+ protected def spark: SparkSession
+
+ /** The destination (target) table entity the AutoCDC flow will be writing
to. */
+ protected def destination: Table
+
+ /** The AutoCDC flow's [[ChangeArgs]] (keys, sequencing, columnSelection,
...). */
+ protected def changeArgs: ChangeArgs
+
+ /** Full schema of the auxiliary table for this SCD type. */
+ protected def auxiliaryTableSchema: StructType
+
+ // Eagerly validate at construction time that the destination supports
row-level ops.
+ requireDestinationSupportsRowLevelOps()
+
+ /**
+ * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
+ * and return its [[TableIdentifier]].
+ */
+ protected def createAuxiliaryTableIfNotExists(spark: SparkSession):
TableIdentifier = {
+ val auxIdent = AutoCdcAuxiliaryTable.identifier(destination.identifier)
+ // The auxiliary table inherits the target's format so MERGE semantics
line up. When the
+ // target's format is unspecified (None), omit the USING clause and fall
back to the
+ // session's default source provider.
+ val usingClause = destination.format.map(fmt => s"USING
$fmt").getOrElse("")
+ spark.sql(
+ s"""CREATE TABLE IF NOT EXISTS
+ |${auxIdent.quotedString}
+ |(${auxiliaryTableSchema.toDDL}) $usingClause""".stripMargin
+ )
+ auxIdent
+ }
+
+ /**
+ * Validate that the target table's underlying connector implements
+ * [[SupportsRowLevelOperations]], which is the V2 connector contract for
MERGE/UPDATE/DELETE
+ * with rewrite - all operations that the AutoCDC transformation executes.
+ */
+ private def requireDestinationSupportsRowLevelOps(): Unit = {
+ val (catalog, v2Identifier) = resolveTableCatalog(spark,
destination.identifier)
+ val destinationTable = catalog.loadTable(v2Identifier)
+
+ if (!destinationTable.isInstanceOf[SupportsRowLevelOperations]) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE",
+ messageParameters = Map(
+ "tableName" -> destination.identifier.quotedString,
+ "format" -> destination.format.getOrElse("<session default>")
+ )
+ )
+ }
+ }
+
+ private def resolveTableCatalog(
+ spark: SparkSession,
+ ident: TableIdentifier): (TableCatalog, Identifier) = {
+ val catalogManager = spark.sessionState.catalogManager
+ val catalog = ident.catalog
+ .map(catalogManager.catalog)
+ .getOrElse(catalogManager.currentCatalog)
+ .asInstanceOf[TableCatalog]
+ val namespace = ident.database.getOrElse(
+ throw SparkException.internalError(
+ s"Cannot resolve table identifier ${ident.quotedString}: namespace is
unspecified."
+ )
+ )
+ (catalog, Identifier.of(Array(namespace), ident.table))
+ }
+}
+
+/**
+ * A [[StreamingFlowExecution]] that applies a CDC event stream to a target
[[Table]] via
+ * SCD Type 1 MERGE semantics.
+ */
+class Scd1MergeStreamingWrite(
+ val identifier: TableIdentifier,
+ val flow: AutoCdcMergeFlow,
+ val graph: DataflowGraph,
+ val updateContext: PipelineUpdateContext,
+ val checkpointPath: String,
+ val trigger: Trigger,
+ val destination: Table,
+ val sqlConf: Map[String, String]
+) extends StreamingFlowExecution with AutoCdcMergeWriteMixin {
+
+ override def getOrigin: QueryOrigin = flow.origin
+
+ override protected def changeArgs: ChangeArgs = flow.changeArgs
+
+ /**
+ * Resolved Spark [[DataType]] of the sequencing expression.
+ */
+ private def sequencingType: DataType =
+ flow.df.select(flow.changeArgs.sequencing).schema.head.dataType
+
+ override def startStream(): StreamingQuery = {
+ val sourceChangeDataFeed = graph.reanalyzeFlow(flow).df
+ val auxiliaryTableIdentifier = createAuxiliaryTableIfNotExists(spark =
updateContext.spark)
Review Comment:
Per your inline self-comment
(https://github.com/apache/spark/pull/56122#discussion_r3305582944): the
rationale for creating the aux table here rather than in `DatasetManager` is
exactly the kind of "why" a future reader needs. Please land it as a code
comment near `createAuxiliaryTableIfNotExists` or the call site here before
merging.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -303,6 +303,25 @@ object DatasetManager extends Logging {
context.spark.sql(s"TRUNCATE TABLE ${table.identifier.quotedString}")
}
+ if (isFullRefresh) {
+ // On full refresh, drop the AutoCDC auxiliary state associated with
this table (if any) so
+ // that stale delete-tracking data and table properties are not carried
forward into the new
+ // table generation. We unconditionally issue the DROP for every
fully-refreshed target; for
+ // non-AutoCDC tables this is a no-op because
[[AutoCdcAuxiliaryTable.identifier]] derives
+ // its name from [[AutoCdcReservedNames.prefix]], which is reserved
across AutoCDC and
Review Comment:
```suggestion
// its name from [[AutoCdcReservedNames.prefix]], which is reserved by
AutoCDC and
```
nit: "reserved by AutoCDC" reads more naturally than "reserved across
AutoCDC".
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -303,6 +303,25 @@ object DatasetManager extends Logging {
context.spark.sql(s"TRUNCATE TABLE ${table.identifier.quotedString}")
}
+ if (isFullRefresh) {
+ // On full refresh, drop the AutoCDC auxiliary state associated with
this table (if any) so
+ // that stale delete-tracking data and table properties are not carried
forward into the new
+ // table generation. We unconditionally issue the DROP for every
fully-refreshed target; for
+ // non-AutoCDC tables this is a no-op because
[[AutoCdcAuxiliaryTable.identifier]] derives
+ // its name from [[AutoCdcReservedNames.prefix]], which is reserved
across AutoCDC and
+ // therefore cannot collide with a user-managed table.
+
+ // Intentionally DROP and not TRUNCATE for two reasons; First, the
auxiliary table may
Review Comment:
```suggestion
// Intentionally DROP and not TRUNCATE for two reasons: first, the
auxiliary table may
```
nit: "for two reasons; First, ..." is grammatically awkward (semicolon +
capitalized continuation). Colon + lowercase "first" matches the "Second," two
lines down.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala:
##########
@@ -0,0 +1,296 @@
+/*
+ * 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.pipelines.graph
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end tests that exercise interactions between separate AutoCDC
pipelines (i.e.
+ * distinct [[DataflowGraph]] / [[TestPipelineUpdateContext]] invocations)
sharing the same
+ * v2 catalog. These complement the single-pipeline AutoCDC suites by
validating the
+ * boundary semantics between independently-deployed pipelines.
+ *
+ * Each test constructs two graphs and runs them sequentially. In real
deployments these
+ * could be two different pipeline definitions writing into the same
metastore; the tests
+ * here verify that AutoCDC's per-target catalog state (target table,
auxiliary table,
+ * schema invariants) behaves correctly across these pipeline boundaries.
+ */
+class AutoCdcScd1MultiPipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ test("two AutoCDC pipelines targeting separate tables maintain independent
target and " +
+ "auxiliary tables") {
+ val session = spark
+ import session.implicits._
+
+ // Two distinct target tables created up-front.
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_a " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.t_b " +
+ s"(id INT NOT NULL, name STRING, version BIGINT NOT NULL,
$cdcMetadataDdl)"
+ )
+
+ // Pipeline #1 only knows about `t_a`. Its auxiliary table
cat.ns1.__auxiliary_..._t_a
Review Comment:
```suggestion
// Pipeline #1 only knows about `t_a`. Its auxiliary table
`cat.ns1.__spark_autocdc_aux_state_t_a`
```
Stale aux-table name: this looks like pre-rename leftover.
`AutoCdcAuxiliaryTable.identifier` derives
`__spark_autocdc_aux_state_${target.table}` (i.e.
`cat.ns1.__spark_autocdc_aux_state_t_a`), not `__auxiliary_..._t_a`. Future
readers should be able to grep this and find the actual aux table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
