This is an automated email from the ASF dual-hosted git repository.
szehon-ho 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 fb962543082e [SPARK-57625][SDP] Manage Auxiliary Table Lifecycle from
`DatasetManager` 2/2
fb962543082e is described below
commit fb962543082e3638efdc118ea650ba30e60617a8
Author: AnishMahto <[email protected]>
AuthorDate: Fri Jul 10 15:55:49 2026 -0700
[SPARK-57625][SDP] Manage Auxiliary Table Lifecycle from `DatasetManager`
2/2
### What changes were proposed in this pull request?
Today, the auxiliary table is managed (created, full-refreshed, etc.)
during flow execution. For SCD1 this is functionally correct but is poor
timing. The auxiliary table should be managed side-by-side with the target
table that it is a companion to, and table-level validations for the AutoCDC
auxiliary table (ex. key/scd type drifts) should happen well before flow
execution.
For SCD2 which is coming soon, the existing control flow is incompatible.
SCD2 auxiliary tables will contain data columns (SCD1 auxiliary tables only
contain keys columns + CDC metadata column), and therefore will actually need
to undergo schema evolution the same way that target tables do.
The proposal here is to refactor the auxiliary table management such that
DatasetManager recognizes the general concept of an auxiliary table, and manage
full-refresh/schema evolution/catalog table validations along side the target
table it companions.
We should no longer be materializing or validating tables during flow
execution.
**New internal mechanisms**
- Auxiliary tables is now a generic concept in dataflow graph infra,
intentionally decoupled from just AutoCDC.
- Auxiliary tables do not belong directly as nodes in the dataflow graph,
as they are internal tables that should not be depended on/read from. Instead,
we maintain an in-memory map of table<->auxiliary table [spec] in the dataflow
graph
- Auxiliary table spec is a declarative data class that flows can
construct, to tell the dataset manager that an auxiliary table should be
materialized with some desired metadata/state
- The spec is intentionally interpreted during dataset materialization
time. Some properties of the spec are important to validate prior to schema
evolution but only for non-full refresh runs, which is best captured at dataset
materialization.
### Why are the changes needed?
Refactor lifecycle management to `DatasetManager` for better control flow
for SCD1, and necessary for schema evolution of auxiliary table for SCD2 (which
will be introduced in a future PR).
### Does this PR introduce _any_ user-facing change?
Validation now happens earlier and during dataset materialization, rather
than during flow execution. The validations themselves remain unchanged, and
one new validation for SCD-type change was introduced in preparation for SCD2
flows.
No API changes.
### How was this patch tested?
### Was this patch authored or co-authored using generative AI tooling?
Closes #56686 from
AnishMahto/SPARK-57625-manage-aux-table-lifecycle-from-datasetmanager.
Lead-authored-by: AnishMahto <[email protected]>
Co-authored-by: andreas-neumann_data <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 15 +-
.../pipelines/graph/AutoCdcAuxiliaryTable.scala | 335 +++++++++++++++++++
.../sql/pipelines/graph/AuxiliaryTableSpec.scala | 80 +++++
.../spark/sql/pipelines/graph/DataflowGraph.scala | 31 ++
.../spark/sql/pipelines/graph/DatasetManager.scala | 198 +++++++++--
.../spark/sql/pipelines/graph/FlowExecution.scala | 364 +--------------------
.../graph/AutoCdcAuxiliaryTableSuite.scala | 65 +++-
.../AutoCdcScd1AuxiliaryTableDurabilitySuite.scala | 25 ++
.../pipelines/graph/AutoCdcScd1KeyDriftSuite.scala | 40 +--
.../graph/AutoCdcScd1MultiPipelineSuite.scala | 5 +-
.../graph/AutoCdcScd1SchemaEvolutionSuite.scala | 51 +++
11 files changed, 783 insertions(+), 426 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 66b96c9b8657..f04c5a4d5053 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -211,27 +211,32 @@
},
"AUTOCDC_INVALID_STATE" : {
"message" : [
- "AutoCDC flow <flowName> detected an invalid state:"
+ "Detected an invalid AutoCDC state for target table <tableName>:"
],
"subClass" : {
"AUXILIARY_TABLE_KEY_COLUMN_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing key column
<keyColumnName> that is recorded in its <propertyName> table property. The
auxiliary table schema may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing key column <keyColumnName>
that is recorded in its <propertyName> table property. The auxiliary table
schema may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
]
},
"AUXILIARY_TABLE_PROPERTY_MALFORMED" : {
"message" : [
- "The auxiliary table <auxTableName> has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
+ "The internal auxiliary table has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
]
},
"AUXILIARY_TABLE_PROPERTY_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing the required
<propertyName> table property; cannot validate AutoCDC key columns. The
auxiliary table metadata may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing the required <propertyName>
table property; cannot validate AutoCDC key columns. The auxiliary table
metadata may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
]
},
"KEY_SCHEMA_DRIFT" : {
"message" : [
- "The AutoCDC flow's current key columns <expectedKeySchema> do not
match the keys recorded in the auxiliary table <auxTableName> (recorded keys
<recordedKeySchema>). AutoCDC does not support changing key columns or their
types across incremental pipeline runs. To change keys, perform a full refresh
of the target table."
+ "One or more AutoCDC flows writing to the target use key columns
<expectedKeySchema>, which are inconsistent with the keys recorded for it
(recorded <recordedKeySchema>). AutoCDC does not support changing key columns
or their types across incremental pipeline runs. Correct the conflicting
flow(s) or perform a full refresh of the target table."
+ ]
+ },
+ "SCD_TYPE_DRIFT" : {
+ "message" : [
+ "One or more AutoCDC flows writing to the target use SCD type
<expectedScdType>, which is inconsistent with the SCD type recorded for it
(recorded <recordedScdType>). AutoCDC does not support changing a target's SCD
type across incremental pipeline runs. Correct the conflicting flow(s) or
perform a full refresh of the target table."
]
}
},
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
new file mode 100644
index 000000000000..6fcf2fc4cdab
--- /dev/null
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
@@ -0,0 +1,335 @@
+/*
+ * 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 scala.util.control.NonFatal
+
+import org.json4s.JsonAST.{JArray, JString}
+import org.json4s.jackson.JsonMethods.{compact, parse}
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Table =>
CatalogTable, TableCatalog}
+import org.apache.spark.sql.pipelines.autocdc.{AutoCdcReservedNames, ScdType}
+import org.apache.spark.sql.types.{StructField, StructType}
+
+/**
+ * Helpers to construct and validate an AutoCDC flow's auxiliary table within
the context of a
+ * dataflow graph.
+ */
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. If a
+ * table exists with a name matching the name derived here, it is assumed to
be an AutoCDC
+ * auxiliary table that should be managed by the pipeline.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+
+ /**
+ * Reserved table property key set on the auxiliary table to record which
SCD strategy it
+ * serves.
+ */
+ val scdTypePropertyKey: String =
s"${PipelinesTableProperties.pipelinesPrefix}autocdc.scdType"
+
+ /**
+ * Table property recording the auxiliary table's unquoted AutoCDC key
column names as a JSON
+ * string array (e.g. `["id","region"]`). Written once when the auxiliary
table is created and is
+ * considered immutable; full-refresh is the only way to change it.
+ */
+ val keyColumnNamesProperty: String =
+ s"${PipelinesTableProperties.pipelinesPrefix}autocdc.keyColumnNames"
+
+ /**
+ * Serialize key column names to the JSON form stored at
[[keyColumnNamesProperty]].
+ * Round-trips an empty list as `[]`; callers are expected to enforce a
non-empty key set
+ * upstream.
+ */
+ private[graph] def serializeKeyColumnNames(names: Seq[String]): String = {
+ compact(JArray(names.map(JString(_)).toList))
+ }
+
+ /**
+ * Parse a [[keyColumnNamesProperty]] value. `None` if it is not a JSON
array of strings.
+ * Round-trips an empty list as `[]`; callers are expected to enforce a
non-empty key set
+ * upstream.
+ */
+ private[graph] def parseKeyColumnNames(raw: String): Option[Seq[String]] = {
+ val parsed = try Some(parse(raw)) catch { case NonFatal(_) => None }
+ parsed.flatMap {
+ case JArray(elems) =>
+ val names = elems.collect { case JString(s) => s }
+ if (names.size == elems.size) Some(names) else None
+ case _ => None
+ }
+ }
+
+ /**
+ * Build the auxiliary table spec given an AutoCdc flow and the target table
it writes to.
+ *
+ * @param targetTable the dataset that owns the auxiliary table
+ * @param targetTableSchema the AutoCDC target's evolved schema as of the
latest pipeline run
+ * (the union of all flows writing to the target
after schema
+ * evolution, NOT the target's `specifiedSchema`)
+ * @param inputAutoCdcFlow the AutoCDC flow writing to `targetTable`
+ * @return the auxiliary-table spec
+ */
+ def buildAuxiliaryTableSpecFor(
+ targetTable: Table,
+ targetTableSchema: StructType,
+ inputAutoCdcFlow: AutoCdcMergeFlow): AuxiliaryTableSpec = {
+ inputAutoCdcFlow.changeArgs.storedAsScdType match {
+ case ScdType.Type1 =>
+ buildScd1AuxiliaryTableSpecFor(
+ targetTable,
+ targetTableSchema,
+ inputAutoCdcFlow
+ )
+ case ScdType.Type2 =>
+ // SCD2 auxiliary derivation lands with SCD2 support. AutoCdcMergeFlow
rejects SCD2 at
+ // construction today, so a resolved SCD2 flow cannot exist and this
branch is unreachable.
+ throw SparkException.internalError(
+ "SCD2 auxiliary table derivation is not yet implemented."
+ )
+ }
+ }
+
+ /**
+ * Build the SCD1 auxiliary table spec given the AutoCdc flow's declared
keys and the target
+ * table it writes to.
+ *
+ * @param targetTable the dataset that owns the SCD1 auxiliary table
+ * @param targetTableSchema the AutoCDC target's evolved schema as of the
latest pipeline run
+ * (the union of all flows writing to the target
after schema
+ * evolution), from which the key and CDC metadata
fields are
+ * resolved
+ * @param inputAutoCdcFlow the AutoCDC flow writing to `targetTable`
+ * @return the SCD1 auxiliary-table spec
+ */
+ private def buildScd1AuxiliaryTableSpecFor(
+ targetTable: Table,
+ targetTableSchema: StructType,
+ inputAutoCdcFlow: AutoCdcMergeFlow
+ ): AuxiliaryTableSpec = {
+ val scd1AuxiliaryTableIdentifier = identifier(targetTable.identifier)
+
+ val resolver = inputAutoCdcFlow.df.sparkSession.sessionState.conf.resolver
+ val autoCdcKeyColumnNames = inputAutoCdcFlow.changeArgs.keys.map(_.name)
+
+ // The auxiliary table should derive its schema from the exact same
key/CDC metadata column
+ // schema in its corresponding target table. Retrieve those column schemas.
+ val keyFields = autoCdcKeyColumnNames.map { keyColumnName =>
+ findFieldInTargetSchema(
+ targetTableSchema = targetTableSchema,
+ targetTableIdentifier = targetTable.identifier,
+ autoCdcFlowIdentifier = inputAutoCdcFlow.identifier,
+ fieldName = keyColumnName,
+ resolver = resolver
+ )
+ }
+ val cdcMetadataField = findFieldInTargetSchema(
+ targetTableSchema = targetTableSchema,
+ targetTableIdentifier = targetTable.identifier,
+ autoCdcFlowIdentifier = inputAutoCdcFlow.identifier,
+ fieldName = AutoCdcReservedNames.cdcMetadataColName,
+ resolver = resolver
+ )
+
+ val scd1AuxiliaryTableSchema = StructType(keyFields :+ cdcMetadataField)
+
+ val scd1AuxiliaryTableProperties =
+ // Record which SCD strategy this auxiliary table serves so downstream
readers can identify it
+ // without inspecting the schema.
+ Map(scdTypePropertyKey -> ScdType.Type1.label) ++
+ // Persist the AutoCDC key column names as a JSON list; immutable
post-creation (full-refresh
+ // is the only way to change it).
+ Map(keyColumnNamesProperty ->
serializeKeyColumnNames(keyFields.map(_.name))) ++
+ // Inherit the target's format so MERGE semantics line up. When
unspecified, omit the provider
+ // so the catalog falls back to its default.
+ targetTable.format.map(TableCatalog.PROP_PROVIDER -> _)
+
+ AutoCdcAuxiliaryTableSpec(
+ identifier = scd1AuxiliaryTableIdentifier,
+ schema = scd1AuxiliaryTableSchema,
+ properties = scd1AuxiliaryTableProperties,
+ targetTableIdentifier = targetTable.identifier,
+ expectedKeyFields = keyFields,
+ expectedScdType = ScdType.Type1
+ )
+ }
+
+ /**
+ * Resolve the [[StructField]] named `fieldName` in `targetTableSchema` (the
AutoCDC target's
+ * evolved schema). The key columns and the CDC metadata column are always
present in that schema,
+ * so a miss is an implementation invariant and surfaces as an internal
error.
+ *
+ * @param targetTableSchema the AutoCDC target's evolved schema to resolve
against
+ * @param fieldName the column name to resolve
+ * @param resolver the session resolver used for case-sensitivity-aware
field lookups
+ * @param targetTableIdentifier the AutoCDC target's identifier, named in
the error message
+ * @param autoCdcFlowIdentifier the AutoCDC flow writing to the target,
named in the error message
+ * @return the matching field
+ */
+ private def findFieldInTargetSchema(
+ targetTableSchema: StructType,
+ fieldName: String,
+ resolver: Resolver,
+ targetTableIdentifier: TableIdentifier,
+ autoCdcFlowIdentifier: TableIdentifier): StructField = {
+ targetTableSchema.fields
+ .find(field => resolver(field.name, fieldName))
+ .getOrElse(
+ throw SparkException.internalError(
+ s"Expected but unable to find column $fieldName in target table " +
+ s"$targetTableIdentifier written to by AutoCDC flow
$autoCdcFlowIdentifier."
+ )
+ )
+ }
+
+ /**
+ * Reject an existing auxiliary table whose key columns have drifted from
`expectedKeyFields` as a
+ * set: same arity, same set of names (per `resolver`), same per-name
`dataType`s. Nullability and
+ * metadata changes are intentionally tolerated.
+ *
+ * AutoCDC cannot change keys across incremental runs; a changed key set
would otherwise be
+ * silently unioned into the schema by the additive evolve. The remedy is a
full refresh, which
+ * recreates the auxiliary table. Errors name the AutoCDC target table
rather than any single
+ * flow, since one auxiliary table is shared by every flow writing to that
target.
+ */
+ private[graph] def validateNoKeyColumnDrift(
+ existingAuxiliaryTable: CatalogTable,
+ targetTableIdentifier: TableIdentifier,
+ expectedKeyFields: Seq[StructField],
+ resolver: Resolver): Unit = {
+ val existingAuxSchema =
CatalogV2Util.v2ColumnsToStructType(existingAuxiliaryTable.columns())
+ val recordedKeyNames =
+ parseRecordedKeyColumnNames(existingAuxiliaryTable,
targetTableIdentifier)
+ // First validate the existing auxiliary table is internally consistent:
every key column name
+ // recorded in its table property must still resolve to a field in its
schema. A missing key
+ // column means the table was corrupted or modified externally, and is
rejected before any
+ // drift comparison against the expected key fields.
+ val recordedKeyFields: Seq[StructField] = recordedKeyNames.map { name =>
+ existingAuxSchema.fields
+ .find(field => resolver(field.name, name))
+ .getOrElse(
+ // Either an implementation bug or, more likely, the user has
corrupted the auxiliary
+ // table schema (e.g. dropped the key column). The remedy is
full-refresh in either case.
+ throw new AnalysisException(
+ errorClass =
"AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_KEY_COLUMN_MISSING",
+ messageParameters = Map(
+ "tableName" -> targetTableIdentifier.unquotedString,
+ "keyColumnName" -> name,
+ "propertyName" -> keyColumnNamesProperty
+ )
+ )
+ )
+ }
+
+ val drifted =
+ // Arity drift (added or dropped keys).
+ recordedKeyFields.length != expectedKeyFields.length ||
+ // Name or dataType drift: every expected key must have a same-name
(resolver-aware) recorded
+ // counterpart with an equivalent dataType. Columns changing nullability
and metadata in the
+ // schema are intentionally tolerated, although null key values during
microbatch execution
+ // will be invalidated regardless.
+ expectedKeyFields.exists { expected =>
+ recordedKeyFields.find(rf => resolver(rf.name, expected.name)) match {
+ case None => true
+ case Some(recorded) => !recorded.dataType.sameType(expected.dataType)
+ }
+ }
+
+ if (drifted) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
+ messageParameters = Map(
+ "tableName" -> targetTableIdentifier.unquotedString,
+ "expectedKeySchema" -> StructType(expectedKeyFields).toDDL,
+ "recordedKeySchema" -> StructType(recordedKeyFields).toDDL
+ )
+ )
+ }
+ }
+
+ /**
+ * Reject an existing auxiliary table whose recorded `scdType` differs from
the expected one.
+ * SCD1 and SCD2 auxiliary tables carry different state shapes, so an
in-place flip is
+ * incompatible; the remedy is a full refresh, which recreates the auxiliary
table.
+ */
+ private[graph] def validateNoScdTypeDrift(
+ existingAuxiliaryTable: CatalogTable,
+ targetTableIdentifier: TableIdentifier,
+ expectedScdType: ScdType): Unit = {
+ val recordedScdType = Option(
+ existingAuxiliaryTable.properties().get(scdTypePropertyKey)
+ ).getOrElse {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING",
+ messageParameters = Map(
+ "tableName" -> targetTableIdentifier.unquotedString,
+ "propertyName" -> scdTypePropertyKey
+ )
+ )
+ }
+ if (recordedScdType != expectedScdType.label) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_INVALID_STATE.SCD_TYPE_DRIFT",
+ messageParameters = Map(
+ "tableName" -> targetTableIdentifier.unquotedString,
+ "expectedScdType" -> expectedScdType.label,
+ "recordedScdType" -> recordedScdType
+ )
+ )
+ }
+ }
+
+ /**
+ * Read [[keyColumnNamesProperty]] off an existing auxiliary table and parse
it into the ordered
+ * list of recorded AutoCDC key column names.
+ */
+ private def parseRecordedKeyColumnNames(
+ existingAuxiliaryTable: CatalogTable,
+ targetTableIdentifier: TableIdentifier): Seq[String] = {
+ val rawKeyColumnNamesStr = Option(
+ existingAuxiliaryTable.properties().get(keyColumnNamesProperty)
+ ).getOrElse {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING",
+ messageParameters = Map(
+ "tableName" -> targetTableIdentifier.unquotedString,
+ "propertyName" -> keyColumnNamesProperty
+ )
+ )
+ }
+ parseKeyColumnNames(rawKeyColumnNamesStr).getOrElse {
+ throw new AnalysisException(
+ errorClass =
"AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MALFORMED",
+ messageParameters = Map(
+ "tableName" -> targetTableIdentifier.unquotedString,
+ "propertyName" -> keyColumnNamesProperty,
+ "rawValue" -> rawKeyColumnNamesStr
+ )
+ )
+ }
+ }
+}
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AuxiliaryTableSpec.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AuxiliaryTableSpec.scala
new file mode 100644
index 000000000000..55cd19a1f8c4
--- /dev/null
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AuxiliaryTableSpec.scala
@@ -0,0 +1,80 @@
+/*
+ * 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.catalyst.TableIdentifier
+import org.apache.spark.sql.pipelines.autocdc.ScdType
+import org.apache.spark.sql.types.{StructField, StructType}
+
+/**
+ * A specification for an internal auxiliary table whose lifecycle follows a
materialized
+ * dataset of the [[DataflowGraph]]: it is created/evolved alongside that
dataset during dataset
+ * materialization and dropped when that dataset is fully refreshed and still
accompanied by an
+ * auxiliary table spec in the current pipeline definition. If the
accompanying flow (e.g. the
+ * AutoCDC flow) is removed and the target is fully refreshed, the orphaned
auxiliary table is
+ * intentionally left in place.
+ *
+ * An [[AuxiliaryTableSpec]] describes a table that is deliberately NOT part
of the logical
+ * [[DataflowGraph]]: it is never resolved, connected, materialized as a
user-facing dataset, nor
+ * exposed as an [[Input]] to other flows, and it is intentionally outside the
[[Output]]
+ * hierarchy.
+ *
+ * An auxiliary table is owned by the dataset it accompanies and derived from
the flows that write
+ * to that dataset. This ownership coupling is intentional and part of the
definition of an
+ * auxiliary/companion table: it cannot exist independently of an owning
dataset, which is itself an
+ * artifact of the dataflow graph.
+ */
+sealed trait AuxiliaryTableSpec {
+ /** The catalog identifier of the auxiliary table. */
+ def identifier: TableIdentifier
+
+ /** The schema the auxiliary table should be created with (and evolved
towards). */
+ def schema: StructType
+
+ /** The table properties the auxiliary table should be created/altered with.
*/
+ def properties: Map[String, String]
+}
+
+/**
+ * An [[AuxiliaryTableSpec]] for the auxiliary state table owned by an AutoCDC
target. Beyond the
+ * create/evolve shape, it carries the metadata required to reject drift of an
already-materialized
+ * auxiliary table before it is evolved. The drift check itself is a stateless
method on
+ * [[AutoCdcAuxiliaryTable]]; this spec is pure data describing what that
check expects.
+ *
+ * No flow name is carried here on purpose: a single auxiliary table is shared
by every AutoCDC flow
+ * writing to its target, so drift errors name the target table rather than
any one flow.
+ *
+ * @param identifier the catalog identifier of the auxiliary table.
+ * @param schema the schema the auxiliary table should be
created with (and evolved
+ * towards).
+ * @param properties the table properties the auxiliary table should
be created/altered
+ * with.
+ * @param targetTableIdentifier the identifier of the AutoCDC target this
auxiliary table belongs
+ * to, used to name the target in drift error
messages.
+ * @param expectedKeyFields the AutoCDC key fields the auxiliary table is
expected to carry if
+ * it already exists, in order (names and types).
+ * @param expectedScdType the SCD type the auxiliary table is expected to
have recorded, if it
+ * already exists.
+ */
+final case class AutoCdcAuxiliaryTableSpec(
+ identifier: TableIdentifier,
+ schema: StructType,
+ properties: Map[String, String],
+ targetTableIdentifier: TableIdentifier,
+ expectedKeyFields: Seq[StructField],
+ expectedScdType: ScdType) extends AuxiliaryTableSpec
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 c9578ddd3b46..c5210976d3f9 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
@@ -181,6 +181,37 @@ case class DataflowGraph(
}.toMap
}
+ /**
+ * The internal auxiliary tables owned by each destination [[Table]],
derived from the resolved
+ * flows writing to it and that destination's [[inferredSchema]]. Keyed by
the destination's
+ * identifier; only destinations that actually require auxiliary tables
appear. Today only
+ * AutoCDC flow destination tables have an auxiliary table, and exactly one.
+ *
+ * Auxiliary tables are deliberately NOT part of the logical graph (they are
never resolved,
+ * connected, or exposed as [[Input]]s); this is purely a derived view used
during dataset
+ * materialization to create/evolve them alongside their owning table. The
derivation is pure and
+ * performs no catalog access.
+ */
+ lazy val auxiliaryTableSpecs: Map[TableIdentifier, AuxiliaryTableSpec] = {
+ resolvedFlowsTo.flatMap { case (destinationTableIdentifier,
flowsToDestinationTable) =>
+ table.get(destinationTableIdentifier).flatMap { destinationTable =>
+ flowsToDestinationTable
+ // A target is written by at most one AutoCDC flow today (graph
validation rejects
+ // multi-flow AutoCDC), so take the single AutoCDC flow if one
exists. Furthermore, today
+ // only AutoCDC flows produce an auxiliary table artifact.
+ .collectFirst { case f: AutoCdcMergeFlow => f }
+ .map { autoCdcFlow =>
+ val spec = AutoCdcAuxiliaryTable.buildAuxiliaryTableSpecFor(
+ targetTable = destinationTable,
+ targetTableSchema = inferredSchema(destinationTableIdentifier),
+ inputAutoCdcFlow = autoCdcFlow
+ )
+ destinationTableIdentifier -> spec
+ }
+ }
+ }.toMap
+ }
+
/** Ensure that the [[DataflowGraph]] is valid and throws errors if not. */
def validate(): DataflowGraph = {
validationFailure.toOption match {
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
index 3c813dd997de..4f96f2f70990 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
@@ -29,6 +29,7 @@ import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.connector.catalog.{
CatalogV2Util,
Identifier,
+ SupportsRowLevelOperations,
Table => V2Table,
TableCatalog,
TableChange,
@@ -38,6 +39,7 @@ import
org.apache.spark.sql.connector.catalog.CatalogV2Util.v2ColumnsToStructTyp
import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
Expressions, Transform}
import org.apache.spark.sql.execution.command.CreateViewCommand
import org.apache.spark.sql.pipelines.graph.QueryOrigin.ExceptionHelpers
+import org.apache.spark.sql.pipelines.util.PipelinesCatalogUtils
import org.apache.spark.sql.pipelines.util.SchemaInferenceUtils.diffSchemas
import org.apache.spark.sql.pipelines.util.SchemaMergingUtils
import org.apache.spark.sql.types.StructType
@@ -104,12 +106,38 @@ object DatasetManager extends Logging {
transformer.transformTables { table =>
if (tablesToMaterialize.keySet.contains(table.identifier)) {
try {
- materializeTable(
+ val isFullRefresh =
tablesToMaterialize(table.identifier).isFullRefresh
+ val (tableWithMaterializationMetadata, catalogTableEntity) =
materializeTable(
resolvedDataflowGraph = resolvedDataflowGraph,
table = table,
- isFullRefresh =
tablesToMaterialize(table.identifier).isFullRefresh,
+ isFullRefresh = isFullRefresh,
context = context
)
+ // Auxiliary tables' lifecycle should follow the table that it
is complementary to.
+ // If this table has any auxiliary tables, validate the target
can host them and
+ // materialize/full-refresh them accordingly.
+
resolvedDataflowGraph.auxiliaryTableSpecs.get(table.identifier).foreach {
+ auxiliaryTableSpec =>
+ // If this table is an AutoCDC target table, as identified
by being
+ // accompanied by an AutoCDC auxiliary table, additionally
validate that the
+ // target table supports row level mutations. This is a
relevant validation for
+ // the auxiliary table itself too, whose format for
AutoCDC is fully derived
+ // from the target table.
+ auxiliaryTableSpec match {
+ case _: AutoCdcAuxiliaryTableSpec =>
+ requireAutoCdcTargetSupportsRowLevelOps(
+ targetTable = table,
+ targetTableCatalogEntity = catalogTableEntity
+ )
+ }
+
+ materializeAuxiliaryTable(
+ auxiliaryTableSpec = auxiliaryTableSpec,
+ isFullRefresh = isFullRefresh,
+ context = context
+ )
+ }
+ tableWithMaterializationMetadata
} catch {
case NonFatal(e) =>
throw TableMaterializationException(
@@ -243,24 +271,19 @@ object DatasetManager extends Logging {
* @param table The table to be materialized.
* @param isFullRefresh Whether this table should be full refreshed or not.
* @param context The context for the pipeline update.
- * @return The materialized table (with additional metadata set).
+ * @return The materialized graph [[Table]] (with additional metadata set)
paired with the loaded
+ * DSv2 handle of the just created/evolved table.
*/
private def materializeTable(
resolvedDataflowGraph: DataflowGraph,
table: Table,
isFullRefresh: Boolean,
- context: PipelineUpdateContext): Table = {
+ context: PipelineUpdateContext): (Table, V2Table) = {
logInfo(log"Materializing metadata for table ${MDC(LogKeys.TABLE_NAME,
table.identifier)}.")
- val catalogManager = context.spark.sessionState.catalogManager
- val catalog = (table.identifier.catalog match {
- case Some(catalogName) =>
- catalogManager.catalog(catalogName)
- case None =>
- catalogManager.currentCatalog
- }).asInstanceOf[TableCatalog]
+ // Get the DSv2 catalog handler and identifier for the table.
+ val (catalog, identifier) =
+ PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
- val identifier =
- Identifier.of(Array(table.identifier.database.get),
table.identifier.identifier)
val outputSchema = table.specifiedSchema.getOrElse(
resolvedDataflowGraph.inferredSchema(table.identifier).asNullable
)
@@ -307,20 +330,6 @@ 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.
-
- // Intentionally DROP and not TRUNCATE: the auxiliary table is an
internal state store
- // that is not part of the dataflow graph, so it does not participate in
regular schema
- // evolution like user tables do. On a full refresh we want a clean
recreation against
- // the new target schema rather than carrying forward the previous
generation's layout.
-
- val auxiliaryTableId = AutoCdcAuxiliaryTable.identifier(table.identifier)
- context.spark.sql(s"DROP TABLE IF EXISTS
${auxiliaryTableId.quotedString}")
- }
-
// Create the table if absent, otherwise evolve it (schema + properties).
existingTableOpt match {
case Some(existingTable) =>
@@ -342,11 +351,127 @@ object DatasetManager extends Logging {
)
}
- table.copy(
- normalizedPath = Option(
-
catalog.loadTable(identifier).properties().get(TableCatalog.PROP_LOCATION)
+ val catalogTableEntity = catalog.loadTable(identifier)
+ val tableWithMaterializationMetadata =
+ table.copy(
+ normalizedPath =
+
Option(catalogTableEntity.properties().get(TableCatalog.PROP_LOCATION))
)
+
+ (tableWithMaterializationMetadata, catalogTableEntity)
+ }
+
+ /**
+ * Validate that the AutoCDC target table is backed by a connector
implementing
+ * [[SupportsRowLevelOperations]], the DSv2 contract for the
MERGE/UPDATE/DELETE-with-rewrite
+ * operations the AutoCDC transformation relies on. Reuses the target handle
already loaded by
+ * [[materializeTable]], so it performs no additional catalog I/O. Only
AutoCDC auxiliary specs
+ * carry a MERGE-backed target; other auxiliary tables have no such
requirement.
+ *
+ * @param targetTable the target table graph entity, source of
the identifier and
+ * declared format used in the error message.
+ * @param targetTableCatalogEntity the target table's loaded DSv2 handle.
+ */
+ private def requireAutoCdcTargetSupportsRowLevelOps(
+ targetTable: Table,
+ targetTableCatalogEntity: V2Table): Unit = {
+ if (!targetTableCatalogEntity.isInstanceOf[SupportsRowLevelOperations]) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE",
+ messageParameters = Map(
+ "tableName" -> targetTable.identifier.quotedString,
+ // Prefer the flow-declared format, falling back to the connector's
provider property.
+ "format" -> targetTable.format
+
.orElse(Option(targetTableCatalogEntity.properties.get(TableCatalog.PROP_PROVIDER)))
+ .getOrElse("<unknown>")
+ )
+ )
+ }
+ }
+
+ /**
+ * Materialize the auxiliary table according to the provided spec.
+ *
+ * @param auxiliaryTableSpec the spec describing the auxiliary table to
create/evolve.
+ * @param isFullRefresh whether the owning table is being fully refreshed.
+ * @param context the context for the pipeline update.
+ */
+ private def materializeAuxiliaryTable(
+ auxiliaryTableSpec: AuxiliaryTableSpec,
+ isFullRefresh: Boolean,
+ context: PipelineUpdateContext): Unit = {
+ // Get the DSv2 catalog handler and identifier for the aux table.
+ val (catalog, auxiliaryTableIdentifier) =
+ PipelinesCatalogUtils.resolveTableCatalog(context.spark,
auxiliaryTableSpec.identifier)
+
+ logInfo(
+ log"Materializing auxiliary table " +
+ log"${MDC(LogKeys.TABLE_NAME, auxiliaryTableSpec.identifier)}."
)
+
+ if (isFullRefresh) {
+ // Intentionally DROP and not TRUNCATE on full refresh. The auxiliary
table is an internal
+ // table whose identity does not need to be preserved on full refresh,
and has metadata
+ // (ex. table properties) that should not persist between full refreshes.
+ //
+ // DROP + CREATE (rather than an atomic REPLACE) because REPLACE is not
universally supported
+ // by DSv2 catalogs. The non-atomicity is acceptable: a CREATE that
fails after the DROP is
+ // self-healing on the next run (a full refresh re-enters here; an
incremental run recreates
+ // via the create path below).
+ logInfo(
+ log"Dropping and recreating auxiliary table " +
+ log"${MDC(LogKeys.TABLE_NAME, auxiliaryTableSpec.identifier)} as part
of full refresh."
+ )
+
+ // [[dropTable]] is a no-op if the table does not exist.
+ catalog.dropTable(auxiliaryTableIdentifier)
+
+ createTable(
+ catalog = catalog,
+ tableIdentifier = auxiliaryTableIdentifier,
+ schema = auxiliaryTableSpec.schema,
+ properties = auxiliaryTableSpec.properties,
+ transforms = Seq.empty
+ )
+ } else {
+ loadTableIfExists(catalog, auxiliaryTableIdentifier) match {
+ case Some(existingAuxiliaryTable) =>
+ auxiliaryTableSpec match {
+ case autoCdcSpec: AutoCdcAuxiliaryTableSpec =>
+ // For AutoCDC auxiliary tables specifically, we persist
metadata about the AutoCDC
+ // configuration that should be invariant for the flow's
lifetime; i.e until it is
+ // full-refreshed. Validate these configurations remain
invariant before attempting
+ // to evolve the auxiliary table's schema, to prevent corrupting
the table.
+ AutoCdcAuxiliaryTable.validateNoKeyColumnDrift(
+ existingAuxiliaryTable = existingAuxiliaryTable,
+ targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
+ expectedKeyFields = autoCdcSpec.expectedKeyFields,
+ resolver = context.spark.sessionState.conf.resolver
+ )
+ AutoCdcAuxiliaryTable.validateNoScdTypeDrift(
+ existingAuxiliaryTable = existingAuxiliaryTable,
+ targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
+ expectedScdType = autoCdcSpec.expectedScdType
+ )
+ }
+ evolveTable(
+ catalog = catalog,
+ tableIdentifier = auxiliaryTableIdentifier,
+ existingTable = existingAuxiliaryTable,
+ desiredSchema = auxiliaryTableSpec.schema,
+ properties = auxiliaryTableSpec.properties,
+ mergeWithExistingSchema = true
+ )
+ case None =>
+ createTable(
+ catalog = catalog,
+ tableIdentifier = auxiliaryTableIdentifier,
+ schema = auxiliaryTableSpec.schema,
+ properties = auxiliaryTableSpec.properties,
+ transforms = Seq.empty
+ )
+ }
+ }
}
/** Loads the table at `identifier` from `catalog`, or `None` if it does not
exist. */
@@ -358,7 +483,8 @@ object DatasetManager extends Logging {
/**
* Creates the table at `identifier` with the given schema, properties, and
partition/cluster
- * transforms. Used when no table yet exists at the identifier.
+ * transforms. Used both for graph datasets and for internal auxiliary
tables when no table yet
+ * exists at the identifier.
*
* @param schema the schema to create the table with.
* @param properties the table properties to create the table with.
@@ -381,13 +507,15 @@ object DatasetManager extends Logging {
}
/**
- * Evolves the already-existing `existingTable` at `identifier` in place by
diffing its schema
- * and properties, skipping the catalog `alterTable` entirely when nothing
actually changes.
- * Partitioning/clustering cannot change in place, so no transforms are
accepted here.
+ * Evolves the already-existing `existingTable` at `identifier` in place by
diffing its schema and
+ * properties, skipping the catalog `alterTable` entirely when nothing
actually changes.
+ * Partitioning/clustering cannot change in place, so no transforms are
accepted here. Used both
+ * for graph datasets and for internal auxiliary tables.
*
* @param existingTable the currently materialized table.
* @param desiredSchema the schema the table should have as
computed in the current
- * execution (the user-specified or inferred
schema). This is the
+ * execution (for graph datasets, the
user-specified or inferred
+ * schema; for auxiliary tables, the derived
schema). This is the
* "incoming" side and may differ from
`existingTable`'s recorded
* schema due to schema evolution across runs.
* @param properties the declared table properties to (re)set
on the table. Note
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
index 5b5ec776c074..d83526d14ab8 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
@@ -20,31 +20,18 @@ package org.apache.spark.sql.pipelines.graph
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.atomic.AtomicBoolean
-import scala.collection.mutable
import scala.concurrent.{ExecutionContext, Future}
-import scala.jdk.CollectionConverters._
import scala.util.control.NonFatal
-import org.json4s.JsonAST.{JArray, JString}
-import org.json4s.jackson.JsonMethods.{compact, parse}
-
-import org.apache.spark.SparkException
import org.apache.spark.internal.{Logging, LogKeys}
-import org.apache.spark.sql.{AnalysisException, Dataset, Row}
+import org.apache.spark.sql.{Dataset, Row}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.classic.SparkSession
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util,
SupportsRowLevelOperations, Table => CatalogTable, TableCatalog, TableInfo}
-import org.apache.spark.sql.pipelines.autocdc.{
- AutoCdcReservedNames,
- ChangeArgs,
- Scd1BatchProcessor,
- Scd1ForeachBatchHandler
-}
+import org.apache.spark.sql.pipelines.autocdc.{Scd1BatchProcessor,
Scd1ForeachBatchHandler}
import org.apache.spark.sql.pipelines.graph.QueryOrigin.ExceptionHelpers
-import org.apache.spark.sql.pipelines.util.{PipelinesCatalogUtils,
SparkSessionUtils}
+import org.apache.spark.sql.pipelines.util.SparkSessionUtils
import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery, Trigger}
-import org.apache.spark.sql.types.{StructField, StructType}
import org.apache.spark.util.ThreadUtils
/**
@@ -318,293 +305,6 @@ class SinkWrite(
}
}
-object AutoCdcAuxiliaryTable {
- /**
- * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. If a
- * table exists with a name matching the name derived here, it is assumed to
be an AutoCDC
- * auxiliary table that should be managed by the pipeline.
- */
- def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
- table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
- database = destination.database,
- catalog = destination.catalog
- )
-
- /**
- * Reserved table property key set on the auxiliary table to record which
SCD strategy it
- * serves.
- */
- val scdTypePropertyKey: String =
s"${PipelinesTableProperties.pipelinesPrefix}autocdc.scdType"
-
- /**
- * Table property recording the auxiliary table's unquoted AutoCDC key
column names as a JSON
- * string array (e.g. `["id","region"]`). Written once when the auxiliary
table is created and is
- * considered immutable; full-refresh is the only way to change it.
- */
- val keyColumnNamesProperty: String =
- s"${PipelinesTableProperties.pipelinesPrefix}autocdc.keyColumnNames"
-
- /**
- * Serialize key column names to the JSON form stored at
[[keyColumnNamesProperty]].
- * Round-trips an empty list as `[]`; callers are expected to enforce a
non-empty key set
- * upstream.
- */
- def serializeKeyColumnNames(names: Seq[String]): String = {
- compact(JArray(names.map(JString(_)).toList))
- }
-
- /**
- * Parse a [[keyColumnNamesProperty]] value. `None` if it is not a JSON
array of strings.
- * Round-trips an empty list as `[]`; callers are expected to enforce a
non-empty key set
- * upstream.
- */
- def parseKeyColumnNames(raw: String): Option[Seq[String]] = {
- val parsed = try Some(parse(raw)) catch { case NonFatal(_) => None }
- parsed.flatMap {
- case JArray(elems) =>
- val names = elems.collect { case JString(s) => s }
- if (names.size == elems.size) Some(names) else None
- case _ => None
- }
- }
-}
-
-/**
- * Base trait for AutoCDC merge-based write flows.
- *
- * Today, this trait and its children manage auxiliary table creation and
validation across
- * pipeline executions. Eventually we should evolve DatasetManager to be aware
of the concept of
- * auxiliary tables, and streamline creation/validation there.
- */
-trait AutoCdcMergeWriteBase {
- /** 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 identifier, used as `flowName` in error messages
emitted by this mixin. */
- protected def identifier: TableIdentifier
-
- /** 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
-
- /**
- * Create the auxiliary table for [[destination]] if it does not already
exist and return its
- * [[TableIdentifier]].
- *
- * When the aux table already exists, its schema and properties are left
untouched. For SCD1
- * the keys must be invariant across executions and the CDC metadata is
always present, so
- * this is correct; drift validation reads the recorded
`keyColumnNamesProperty` to enforce
- * the invariant before this method is called.
- */
- protected def createAuxiliaryTableIfNotExists(spark: SparkSession):
TableIdentifier = {
- val auxIdent = AutoCdcAuxiliaryTable.identifier(destination.identifier)
- val (catalog, v2Identifier) =
PipelinesCatalogUtils.resolveTableCatalog(spark, auxIdent)
-
- if (!catalog.tableExists(v2Identifier)) {
- val properties = mutable.Map.empty[String, String]
-
- // Inherit the target's format so MERGE semantics line up. When
unspecified, omit the
- // provider so the catalog falls back to its default.
- destination.format.foreach { fmt =>
properties(TableCatalog.PROP_PROVIDER) = fmt }
-
- // Record which SCD strategy this auxiliary table serves so downstream
readers can
- // identify it without having to inspect the schema.
- properties(AutoCdcAuxiliaryTable.scdTypePropertyKey) =
changeArgs.storedAsScdType.label
-
- // Persist the AutoCDC key column names as a JSON list on first
creation. The value
- // is stored verbatim by the catalog.
- properties(AutoCdcAuxiliaryTable.keyColumnNamesProperty) =
- AutoCdcAuxiliaryTable.serializeKeyColumnNames(auxiliaryKeyColumnNames)
-
- // Table creation is not atomic with the table exists check, and
[[createTable]] will fail
- // with TableAlreadyExistsException if some asynchronous process creates
the table between
- // the [[tableExists]] check and [[createTable]]. This is both rare (we
don't support
- // multi-AutoCDC-flow targets so there are no race conditions within a
single pipeline) and
- // acceptable - users can cleanly retry the failed flow when this
happens. SQL offers an
- // atomic CREATE IF NOT EXISTS, but would require special casing of the
table properties
- // in DDL and we would lose compile-time syntax and type safety.
- catalog.createTable(
- v2Identifier,
- new TableInfo.Builder()
-
.withColumns(CatalogV2Util.structTypeToV2Columns(auxiliaryTableSchema))
- .withProperties(properties.asJava)
- .build()
- )
- }
- auxIdent
- }
-
- /**
- * Resolves each AutoCDC key in `changeArgs.keys` to its [[StructField]] in
- * [[auxiliaryTableSchema]], preserving `changeArgs.keys` declaration order.
This is the
- * expected (flow-declared) side of drift validation, distinct from the keys
recorded on an
- * existing auxiliary table.
- *
- * [[AutoCdcMergeFlow]] should have validated that all `changeArgs.keys`
exist in the deduced
- * aux/target schemas by now, so a missing key is an internal error rather
than a user-facing
- * condition.
- */
- private lazy val expectedAuxiliaryKeyFields: Seq[StructField] = {
- val resolver = spark.sessionState.conf.resolver
- changeArgs.keys.map { key =>
- auxiliaryTableSchema.fields
- .find(field => resolver(field.name, key.name))
- .getOrElse(
- throw SparkException.internalError(
- s"AutoCDC key column '${key.name}' is missing from the auxiliary
table schema " +
- s"for flow ${identifier.unquotedString} writing to target " +
- s"${destination.identifier.quotedString}."
- )
- )
- }
- }
-
- /**
- * Returns the resolved AutoCDC key column names as they appear in the
auxiliary schema, in
- * `changeArgs.keys` declaration order.
- */
- private lazy val auxiliaryKeyColumnNames: Seq[String] =
expectedAuxiliaryKeyFields.map(_.name)
-
- /**
- * 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.
- */
- protected def requireDestinationSupportsRowLevelOps(): Unit = {
- val (catalog, v2Identifier) =
- PipelinesCatalogUtils.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.orElse(
- Option(
- destinationTable.properties.get(TableCatalog.PROP_PROVIDER)
- )
- )
- .getOrElse("<unknown>")
- )
- )
- }
- }
-
- /**
- * If the auxiliary table for this flow's destination already exists,
validate that the
- * AutoCDC keys the flow expects line up with the keys recorded in the
auxiliary
- * table. On a fresh pipeline (or after a full refresh dropped the
auxiliary), the
- * auxiliary is absent and there's nothing to drift from, so this is a no-op.
- */
- protected def validateNoAutoCdcKeyDriftIfAuxTableExists(): Unit = {
- val auxIdent = AutoCdcAuxiliaryTable.identifier(destination.identifier)
- val (catalog, v2Identifier) =
PipelinesCatalogUtils.resolveTableCatalog(spark, auxIdent)
- if (catalog.tableExists(v2Identifier)) {
- validateNoAutoCdcKeyDrift(catalog.loadTable(v2Identifier), auxIdent)
- }
- }
-
- /**
- * Validate that the AutoCDC key columns the flow expects match the keys
recorded in the
- * existing auxiliary table at [[auxIdent]] as a set: same arity, same set
of names (per the
- * session resolver), same per-name `dataType`s.
- */
- private def validateNoAutoCdcKeyDrift(
- existingAuxTable: CatalogTable,
- auxIdent: TableIdentifier): Unit = {
- val resolver = spark.sessionState.conf.resolver
- val existingAuxSchema =
CatalogV2Util.v2ColumnsToStructType(existingAuxTable.columns())
-
- // Resolve the flow-declared (expected) keys from
[[auxiliaryTableSchema]]. We deliberately
- // do not look them up in [[existingAuxSchema]] - that's the recorded
side, and conflating
- // the two sides would mask drift. See [[expectedAuxiliaryKeyFields]].
- val expectedKeyFields: Seq[StructField] = expectedAuxiliaryKeyFields
- val recordedKeyNames = parseRecordedKeyColumnNames(existingAuxTable,
auxIdent)
- val recordedKeyFields: Seq[StructField] = recordedKeyNames.map { name =>
- existingAuxSchema.fields
- .find(field => resolver(field.name, name))
- .getOrElse(
- // Either an implementation bug or, more likely, the user has
corrupted the auxiliary
- // table schema (e.g. dropped the key column). The remedy is
full-refresh in either
- // case.
- throw new AnalysisException(
- errorClass =
"AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_KEY_COLUMN_MISSING",
- messageParameters = Map(
- "flowName" -> identifier.unquotedString,
- "auxTableName" -> auxIdent.unquotedString,
- "keyColumnName" -> name,
- "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty
- )
- )
- )
- }
-
- val drifted =
- // Arity drift (added or dropped keys).
- recordedKeyFields.length != expectedKeyFields.length ||
- // Name or dataType drift: every expected key must have a same-name
(resolver-aware)
- // recorded counterpart with an equivalent dataType. Columns changing
nullability and
- // metadata in the schema are intentionally tolerated, although null key
values during
- // microbatch execution will be invalidated regardless.
- expectedKeyFields.exists { expected =>
- recordedKeyFields.find(rf => resolver(rf.name, expected.name)) match {
- case None => true
- case Some(recorded) => !recorded.dataType.sameType(expected.dataType)
- }
- }
-
- if (drifted) {
- throw new AnalysisException(
- errorClass = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
- messageParameters = Map(
- "flowName" -> identifier.unquotedString,
- "auxTableName" -> auxIdent.unquotedString,
- "expectedKeySchema" -> StructType(expectedKeyFields).toDDL,
- "recordedKeySchema" -> StructType(recordedKeyFields).toDDL
- )
- )
- }
- }
-
- /**
- * Read the [[AutoCdcAuxiliaryTable.keyColumnNamesProperty]] off an existing
auxiliary table
- * and parse it into the ordered list of recorded AutoCDC key column names.
- */
- private def parseRecordedKeyColumnNames(
- existingAuxTable: CatalogTable,
- auxIdent: TableIdentifier): Seq[String] = {
- val rawKeyColumnNamesStr = Option(
-
existingAuxTable.properties().get(AutoCdcAuxiliaryTable.keyColumnNamesProperty)
- ).getOrElse {
- throw new AnalysisException(
- errorClass = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING",
- messageParameters = Map(
- "flowName" -> identifier.unquotedString,
- "auxTableName" -> auxIdent.unquotedString,
- "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty
- )
- )
- }
- AutoCdcAuxiliaryTable.parseKeyColumnNames(rawKeyColumnNamesStr).getOrElse {
- throw new AnalysisException(
- errorClass =
"AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MALFORMED",
- messageParameters = Map(
- "flowName" -> identifier.unquotedString,
- "auxTableName" -> auxIdent.unquotedString,
- "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty,
- "rawValue" -> rawKeyColumnNamesStr
- )
- )
- }
- }
-}
-
/**
* A [[StreamingFlowExecution]] that applies a CDC event stream to a target
[[Table]] via
* SCD Type 1 MERGE semantics.
@@ -618,26 +318,17 @@ class Scd1MergeStreamingWrite(
val trigger: Trigger,
val destination: Table,
val sqlConf: Map[String, String]
-) extends StreamingFlowExecution with AutoCdcMergeWriteBase {
-
- requireDestinationSupportsRowLevelOps()
- validateNoAutoCdcKeyDriftIfAuxTableExists()
+) extends StreamingFlowExecution {
override def getOrigin: QueryOrigin = flow.origin
- override protected def changeArgs: ChangeArgs = flow.changeArgs
-
override def startStream(): StreamingQuery = {
val sourceChangeDataFeed = graph.reanalyzeFlow(flow).df
- // The auxiliary table is created here (at flow execution) rather than
during flow resolution
- // or dataset materialization for two reasons:
- // 1. It is an internal state store: we deliberately keep it out of the
graph registration
- // context's table set so that it is invisible to other flows and the
[[DatasetManager]]
- // will never materialize it.
- // 2. Its format must match the target table's, which only exists after
the target is
- // materialized. Flow resolution must also stay side-effect free
(e.g. for dry runs).
- val auxiliaryTableIdentifier = createAuxiliaryTableIfNotExists(spark =
updateContext.spark)
+ // The auxiliary table is created and evolved during dataset
materialization (see
+ // [[DatasetManager]]), so it already exists by the time this flow
executes; resolve its
+ // identifier to hand to the foreachBatch handler.
+ val auxiliaryTableIdentifier =
AutoCdcAuxiliaryTable.identifier(destination.identifier)
val foreachBatchHandler = Scd1ForeachBatchHandler(
batchProcessor = Scd1BatchProcessor(
@@ -657,43 +348,4 @@ class Scd1MergeStreamingWrite(
})
.start()
}
-
- override protected lazy val auxiliaryTableSchema: StructType =
- // SCD1's auxiliary table is just keys + the CDC metadata struct; no user
data columns. Keys
- // come first, in `changeArgs.keys` declaration order, to anchor the
per-key sequence
- // watermark used to gate out-of-order events.
- StructType(autoCdcKeyFields :+ cdcMetadataField)
-
- /**
- * AutoCDC key columns resolved out of the flow's augmented schema, in
- * `changeArgs.keys` declaration order. Keys are guaranteed to be present in
the schema
- * because [[AutoCdcMergeFlow.schema]] validates that.
- */
- private lazy val autoCdcKeyFields: Seq[StructField] = {
- val resolver = updateContext.spark.sessionState.conf.resolver
- val targetTableSchema = flow.schema
- flow.changeArgs.keys.map { key =>
- targetTableSchema.fields
- .find(field => resolver(field.name, key.name))
- .getOrElse(
- throw SparkException.internalError(
- s"Key column '${key.name}' was not found in the AutoCDC flow's
selected schema."
- )
- )
- }
- }
-
- /** CDC metadata field resolved out of the flow's augmented schema. */
- private lazy val cdcMetadataField: StructField = {
- val resolver = updateContext.spark.sessionState.conf.resolver
- val cdcMetadataColName = AutoCdcReservedNames.cdcMetadataColName
- flow.schema.fields
- .find(field => resolver(field.name, cdcMetadataColName))
- .getOrElse(
- throw SparkException.internalError(
- s"CDC metadata column '$cdcMetadataColName' was not found in the " +
- s"AutoCDC flow's target table schema."
- )
- )
- }
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTableSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTableSuite.scala
index 9fb6070c01e7..fb368198f72d 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTableSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTableSuite.scala
@@ -17,12 +17,16 @@
package org.apache.spark.sql.pipelines.graph
+import scala.jdk.CollectionConverters._
+
import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.connector.catalog.{Table, TableCapability}
+import org.apache.spark.sql.pipelines.autocdc.ScdType
/**
- * Unit tests for the [[AutoCdcAuxiliaryTable]] companion object, in
particular the
- * `serializeKeyColumnNames` / `parseKeyColumnNames` round-trip helpers used
to persist the
- * AutoCDC key column names as a JSON-encoded reserved table property on the
auxiliary table.
+ * Unit tests for the [[AutoCdcAuxiliaryTable]] companion object.
*
* These tests are intentionally session-less: the helpers are pure functions
on `String` and
* `Seq[String]`, and verifying their byte-for-byte round-trip contract
requires no Spark
@@ -47,6 +51,14 @@ class AutoCdcAuxiliaryTableSuite extends SparkFunSuite {
)
}
+ /** Minimal [[Table]] stub exposing only the properties map the SCD-type
validator reads. */
+ private def auxTableWithProperties(props: Map[String, String]): Table = new
Table {
+ override def name(): String = "aux"
+ override def capabilities(): java.util.Set[TableCapability] =
+ Set.empty[TableCapability].asJava
+ override def properties(): java.util.Map[String, String] = props.asJava
+ }
+
test("serializeKeyColumnNames/parseKeyColumnNames round-trip preserves plain
ASCII names") {
assertKeyColumnNamesRoundTrip(Seq("id"))
assertKeyColumnNamesRoundTrip(Seq("id", "region"))
@@ -97,4 +109,51 @@ class AutoCdcAuxiliaryTableSuite extends SparkFunSuite {
assert(AutoCdcAuxiliaryTable.parseKeyColumnNames("[\"id\", null]").isEmpty)
assert(AutoCdcAuxiliaryTable.parseKeyColumnNames("[[\"id\"]]").isEmpty)
// nested array
}
+
+ test("validateNoScdTypeDrift accepts an auxiliary table whose recorded SCD
type matches") {
+ val existing =
+ auxTableWithProperties(Map(AutoCdcAuxiliaryTable.scdTypePropertyKey ->
ScdType.Type1.label))
+ // Must not throw.
+ AutoCdcAuxiliaryTable.validateNoScdTypeDrift(
+ existingAuxiliaryTable = existing,
+ targetTableIdentifier = TableIdentifier("target", Some("ns"),
Some("cat")),
+ expectedScdType = ScdType.Type1)
+ }
+
+ test("validateNoScdTypeDrift throws SCD_TYPE_DRIFT when the recorded SCD
type differs") {
+ val existing =
+ auxTableWithProperties(Map(AutoCdcAuxiliaryTable.scdTypePropertyKey ->
ScdType.Type2.label))
+ val ex = intercept[AnalysisException] {
+ AutoCdcAuxiliaryTable.validateNoScdTypeDrift(
+ existingAuxiliaryTable = existing,
+ targetTableIdentifier = TableIdentifier("target", Some("ns"),
Some("cat")),
+ expectedScdType = ScdType.Type1)
+ }
+ checkError(
+ exception = ex,
+ condition = "AUTOCDC_INVALID_STATE.SCD_TYPE_DRIFT",
+ sqlState = "42000",
+ parameters = Map(
+ "tableName" -> TableIdentifier("target", Some("ns"),
Some("cat")).unquotedString,
+ "expectedScdType" -> ScdType.Type1.label,
+ "recordedScdType" -> ScdType.Type2.label))
+ }
+
+ test("validateNoScdTypeDrift throws AUXILIARY_TABLE_PROPERTY_MISSING when
scdType is absent") {
+ // Simulates corrupt/externally-modified metadata (e.g. `ALTER TABLE ...
UNSET TBLPROPERTIES`).
+ val existing = auxTableWithProperties(Map.empty)
+ val ex = intercept[AnalysisException] {
+ AutoCdcAuxiliaryTable.validateNoScdTypeDrift(
+ existingAuxiliaryTable = existing,
+ targetTableIdentifier = TableIdentifier("target", Some("ns"),
Some("cat")),
+ expectedScdType = ScdType.Type1)
+ }
+ checkError(
+ exception = ex,
+ condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING",
+ sqlState = "42000",
+ parameters = Map(
+ "tableName" -> TableIdentifier("target", Some("ns"),
Some("cat")).unquotedString,
+ "propertyName" -> AutoCdcAuxiliaryTable.scdTypePropertyKey))
+ }
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
index bbd8f7c8dbf9..10b8d0dfc5ae 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala
@@ -181,6 +181,31 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite
assert(getAuxTableKeyColumnNames(target = "target") == Seq("region", "id"))
}
+ test("a dry run resolves and validates the graph without provisioning the
auxiliary " +
+ "table") {
+ val session = spark
+ import session.implicits._
+
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+ )
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 1L))
+ val ctx = singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = stream.toDF().toDF("id", "version"),
+ keys = Seq("id"),
+ sequencing = functions.col("version"))
+
+ val updateCtx = TestPipelineUpdateContext(spark, ctx.toDataflowGraph,
storageRoot)
+ updateCtx.pipelineExecution.dryRunPipeline()
+
+ assert(!spark.catalog.tableExists(auxTableNameFor("target")))
+ }
+
test("if the AutoCDC auxiliary table is dropped between runs, it is
transparently " +
"recreated") {
val session = spark
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
index fc7706c84e3e..1e97ab2c2cd2 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
@@ -68,9 +68,8 @@ class AutoCdcScd1KeyDriftSuite
condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
sqlState = Some("42000"),
parameters = Map(
- "flowName" ->
- fullyQualifiedIdentifier("flow_v2", Some(catalog),
Some(namespace)).unquotedString,
- "auxTableName" -> auxTableNameFor("target"),
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
// `region` is nullable here because Scala `String` is a reference
type and the
// [[MemoryStream]] tuple encoder treats reference types as nullable.
Only Scala
// primitives (`Int`, `Long`, ...) yield `NOT NULL` columns.
@@ -105,9 +104,8 @@ class AutoCdcScd1KeyDriftSuite
condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
sqlState = Some("42000"),
parameters = Map(
- "flowName" ->
- fullyQualifiedIdentifier("flow_v2", Some(catalog),
Some(namespace)).unquotedString,
- "auxTableName" -> auxTableNameFor("target"),
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
"expectedKeySchema" -> "id INT NOT NULL",
// `region` is nullable here because Scala `String` is a reference
type; see the
// analogous comment in the "adds a key column" test above.
@@ -144,9 +142,8 @@ class AutoCdcScd1KeyDriftSuite
condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
sqlState = Some("42000"),
parameters = Map(
- "flowName" ->
- fullyQualifiedIdentifier("flow_v2", Some(catalog),
Some(namespace)).unquotedString,
- "auxTableName" -> auxTableNameFor("target"),
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
// `country` and `region` are nullable here because Scala `String` is
a reference type;
// see the analogous comment in the "adds a key column" test above.
"expectedKeySchema" -> "id INT NOT NULL,country STRING",
@@ -176,9 +173,8 @@ class AutoCdcScd1KeyDriftSuite
condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
sqlState = Some("42000"),
parameters = Map(
- "flowName" ->
- fullyQualifiedIdentifier("flow", Some(catalog),
Some(namespace)).unquotedString,
- "auxTableName" -> auxTableNameFor("target"),
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
"expectedKeySchema" -> "id INT NOT NULL",
"recordedKeySchema" -> "id BIGINT NOT NULL"
)
@@ -302,9 +298,8 @@ class AutoCdcScd1KeyDriftSuite
condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
sqlState = Some("42000"),
parameters = Map(
- "flowName" ->
- fullyQualifiedIdentifier("flow_v2", Some(catalog),
Some(namespace)).unquotedString,
- "auxTableName" -> auxTableNameFor("target"),
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
"expectedKeySchema" -> "Id INT NOT NULL",
"recordedKeySchema" -> "id INT NOT NULL"
)
@@ -362,9 +357,8 @@ class AutoCdcScd1KeyDriftSuite
condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MISSING",
sqlState = Some("42000"),
parameters = Map(
- "flowName" ->
- fullyQualifiedIdentifier("flow", Some(catalog),
Some(namespace)).unquotedString,
- "auxTableName" -> auxTableNameFor("target"),
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
"propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty
)
)
@@ -395,9 +389,8 @@ class AutoCdcScd1KeyDriftSuite
condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_PROPERTY_MALFORMED",
sqlState = Some("42000"),
parameters = Map(
- "flowName" ->
- fullyQualifiedIdentifier("flow", Some(catalog),
Some(namespace)).unquotedString,
- "auxTableName" -> auxTableNameFor("target"),
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
"propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty,
"rawValue" -> malformedKeysArray
)
@@ -430,9 +423,8 @@ class AutoCdcScd1KeyDriftSuite
condition = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_KEY_COLUMN_MISSING",
sqlState = Some("42000"),
parameters = Map(
- "flowName" ->
- fullyQualifiedIdentifier("flow", Some(catalog),
Some(namespace)).unquotedString,
- "auxTableName" -> auxTableNameFor("target"),
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
"keyColumnName" -> "region",
"propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty
)
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
index 2100928bc68a..f86c592e3cf7 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1MultiPipelineSuite.scala
@@ -298,9 +298,8 @@ class AutoCdcScd1MultiPipelineSuite
condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
sqlState = Some("42000"),
parameters = Map(
- "flowName" ->
- fullyQualifiedIdentifier("flow_v2", Some(catalog),
Some(namespace)).unquotedString,
- "auxTableName" -> auxTableNameFor("shared_target"),
+ "tableName" ->
+ fullyQualifiedIdentifier("shared_target", Some(catalog),
Some(namespace)).unquotedString,
// Pipeline #2's AutoCDC key resolves from the source DF, where
`MemoryStream[(Int, String,
// Long)]` produces a nullable StringType for `name`.
"expectedKeySchema" -> "name STRING",
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
index b6c8f2179b7f..fb3094aa2eb4 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
@@ -24,6 +24,7 @@ import
org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.functions
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.pipelines.autocdc.{
+ AutoCdcReservedNames,
ColumnSelection,
UnqualifiedColumnName
}
@@ -221,6 +222,56 @@ class AutoCdcScd1SchemaEvolutionSuite
)
}
+ test("additive target-column evolution leaves the SCD1 auxiliary table
schema unchanged") {
+ val session = spark
+ import session.implicits._
+
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)"
+ )
+
+ // Shared (id, name, version) stream so the streaming checkpoint resumes
cleanly across
+ // runs; run #1 projects away `name` (target starts at (id, version)), run
#2 keeps it so
+ // the target additively gains `name`.
+ val stream = MemoryStream[(Int, String, Long)]
+ def buildCtx(includeName: Boolean): TestGraphRegistrationContext = {
+ val sourceDf = stream.toDF().toDF("id", "name", "version")
+ val projectedDf = if (includeName) sourceDf else sourceDf.drop("name")
+ singleAutoCdcFlowPipeline(
+ flowName = "auto_cdc_flow",
+ target = "target",
+ sourceDf = projectedDf,
+ keys = Seq("id"),
+ sequencing = functions.col("version"))
+ }
+
+ val expectedAuxSchema = Seq("id", AutoCdcReservedNames.cdcMetadataColName)
+
+ // Run #1: target is (id, version); aux is (id, _cdc_metadata).
+ stream.addData((1, "ignored", 1L))
+ runPipeline(buildCtx(includeName = false))
+ assert(
+ spark.table(auxTableNameFor("target")).schema.fieldNames.toSeq ==
expectedAuxSchema,
+ "auxiliary schema after run #1 should be the keys plus the CDC metadata
column"
+ )
+
+ // Run #2: `name` is added to the target but the auxiliary schema must
remain unchanged, since
+ // the SCD keys remain unchanged. For SCD1, the auxiliary table only
contains key and CDC
+ // metadata columns.
+ stream.addData((2, "bob", 2L))
+ runPipeline(buildCtx(includeName = true))
+
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, 1L, cdcMeta(None, Some(1L)), null),
+ Row(2, 2L, cdcMeta(None, Some(2L)), "bob")
+ )
+ )
+ assert(spark.table(auxTableNameFor("target")).schema.fieldNames.toSeq ==
expectedAuxSchema)
+ }
+
test("broadening the column selection between runs adds the newly-included
column to " +
"the target") {
val session = spark
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
