From 378a98f89b15dbdb0f5d189b135074adfed1dd7e Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Mon, 22 Jun 2026 23:46:21 +0000 Subject: [PATCH 1/8] extract table creation and evolution into helper functions --- .../sql/pipelines/graph/DatasetManager.scala | 126 +++++++++++++----- 1 file changed, 94 insertions(+), 32 deletions(-) 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 456edca8d1e22..38e5d3e66e517 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,16 +29,18 @@ import org.apache.spark.sql.classic.SparkSession import org.apache.spark.sql.connector.catalog.{ CatalogV2Util, Identifier, + Table => CatalogTable, TableCatalog, TableChange, TableInfo } import org.apache.spark.sql.connector.catalog.CatalogV2Util.v2ColumnsToStructType -import org.apache.spark.sql.connector.expressions.{ClusterByTransform, Expressions} +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.SchemaInferenceUtils.diffSchemas import org.apache.spark.sql.pipelines.util.SchemaMergingUtils +import org.apache.spark.sql.types.StructType /** * `DatasetManager` is responsible for materializing tables in the catalog based on the given @@ -278,11 +280,7 @@ object DatasetManager extends Logging { val allTransforms = partitioning ++ clustering - val existingTableOpt = if (catalog.tableExists(identifier)) { - Some(catalog.loadTable(identifier)) - } else { - None - } + val existingTableOpt = loadTableIfExists(catalog, identifier) // Error if partitioning/clustering doesn't match existingTableOpt.foreach { existingTable => @@ -298,8 +296,14 @@ object DatasetManager extends Logging { } } + // A streaming table on a non-full-refresh run is maintained incrementally: its existing data is + // preserved and its schema is merged with (not replaced by) the schema computed in this run. + // Every other case (materialized views, and any full refresh) is recomputed from scratch: + // existing data is wiped and the schema is taken directly from this run's computed schema. + val isTableIncrementallyUpdated = table.isStreamingTable && !isFullRefresh + // Wipe the data if we need to - if ((isFullRefresh || !table.isStreamingTable) && existingTableOpt.isDefined) { + if (existingTableOpt.isDefined && !isTableIncrementallyUpdated) { context.spark.sql(s"TRUNCATE TABLE ${table.identifier.quotedString}") } @@ -317,31 +321,25 @@ object DatasetManager extends Logging { context.spark.sql(s"DROP TABLE IF EXISTS ${auxiliaryTableId.quotedString}") } - // Alter the table if we need to - existingTableOpt.foreach { existingTable => - val existingSchema = v2ColumnsToStructType(existingTable.columns()) - - val targetSchema = if (table.isStreamingTable && !isFullRefresh) { - SchemaMergingUtils.mergeSchemas(existingSchema, outputSchema) - } else { - outputSchema - } - - val columnChanges = diffSchemas(existingSchema, targetSchema) - val setProperties = mergedProperties.map { case (k, v) => TableChange.setProperty(k, v) } - catalog.alterTable(identifier, (columnChanges ++ setProperties).toArray: _*) - } - - // Create the table if we need to - if (existingTableOpt.isEmpty) { - catalog.createTable( - identifier, - new TableInfo.Builder() - .withProperties(mergedProperties.asJava) - .withColumns(CatalogV2Util.structTypeToV2Columns(outputSchema)) - .withPartitions(allTransforms.toArray) - .build() - ) + // Create the table if absent, otherwise evolve it (schema + properties). + existingTableOpt match { + case Some(existingTable) => + evolveTableInCatalog( + catalog = catalog, + tableIdentifier = identifier, + existingTable = existingTable, + desiredSchema = outputSchema, + properties = mergedProperties, + mergeWithExistingSchema = isTableIncrementallyUpdated + ) + case None => + createTableInCatalog( + catalog = catalog, + tableIdentifier = identifier, + schema = outputSchema, + properties = mergedProperties, + transforms = allTransforms + ) } table.copy( @@ -351,6 +349,70 @@ object DatasetManager extends Logging { ) } + /** Loads the table at `identifier` from `catalog`, or `None` if it does not exist. */ + private def loadTableIfExists( + catalog: TableCatalog, + identifier: Identifier): Option[CatalogTable] = { + Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier)) + } + + /** + * Creates the table at `identifier` with the given schema, properties, and partition/cluster + * transforms. Used 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. + * @param transforms the partition/cluster transforms to create the table with. + */ + private def createTableInCatalog( + catalog: TableCatalog, + tableIdentifier: Identifier, + schema: StructType, + properties: Map[String, String], + transforms: Seq[Transform]): Unit = { + catalog.createTable( + tableIdentifier, + new TableInfo.Builder() + .withProperties(properties.asJava) + .withColumns(CatalogV2Util.structTypeToV2Columns(schema)) + .withPartitions(transforms.toArray) + .build() + ) + } + + /** + * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema and + * (re)setting its properties. Partitioning/clustering cannot change in place, so no transforms are + * accepted here. + * + * @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 + * "incoming" side and may differ from `existingTable`'s recorded + * schema due to schema evolution across runs. + * @param properties the table properties to (re)set on evolve. + * @param mergeWithExistingSchema whether the effective schema is the merge of the existing and + * desired schemas (additive evolution) rather than the desired + * schema as-is. + */ + private def evolveTableInCatalog( + catalog: TableCatalog, + tableIdentifier: Identifier, + existingTable: CatalogTable, + desiredSchema: StructType, + properties: Map[String, String], + mergeWithExistingSchema: Boolean): Unit = { + val currentSchema = v2ColumnsToStructType(existingTable.columns()) + val targetSchema = if (mergeWithExistingSchema) { + SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema) + } else { + desiredSchema + } + val columnChanges = diffSchemas(currentSchema, targetSchema) + val setProperties = properties.map { case (k, v) => TableChange.setProperty(k, v) } + catalog.alterTable(tableIdentifier, (columnChanges ++ setProperties).toArray: _*) + } + /** * Some fields on the [[Table]] object are represented as reserved table properties by the catalog * APIs. This method creates a table properties map that merges the user-provided table properties From 27ebee3f220f853025a22595d1c5159195d634c0 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Tue, 23 Jun 2026 00:29:32 +0000 Subject: [PATCH 2/8] manage auxiliary table in DatasetManager --- .../resources/error/error-conditions.json | 15 +- .../graph/AutoCdcAuxiliaryTable.scala | 331 ++++++++++++++++ .../pipelines/graph/AuxiliaryTableSpec.scala | 77 ++++ .../sql/pipelines/graph/DataflowGraph.scala | 31 ++ .../sql/pipelines/graph/DatasetManager.scala | 113 +++++- .../sql/pipelines/graph/FlowExecution.scala | 364 +----------------- ...CdcScd1AuxiliaryTableDurabilitySuite.scala | 32 ++ .../graph/AutoCdcScd1KeyDriftSuite.scala | 40 +- .../graph/AutoCdcScd1MultiPipelineSuite.scala | 5 +- .../AutoCdcScd1SchemaEvolutionSuite.scala | 59 +++ 10 files changed, 660 insertions(+), 407 deletions(-) create mode 100644 sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala create mode 100644 sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AuxiliaryTableSpec.scala diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index b1ee046a28ea1..69803bd20e830 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 detected an invalid state:" + "Detected an invalid AutoCDC state for target table :" ], "subClass" : { "AUXILIARY_TABLE_KEY_COLUMN_MISSING" : { "message" : [ - "The auxiliary table is missing key column that is recorded in its 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 that is recorded in its 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 has a malformed property with raw value ''. 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 property with raw value ''. 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 is missing the required 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 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 do not match the keys recorded in the auxiliary table (recorded keys ). 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 , which are inconsistent with the keys recorded for it (recorded ). 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 , which is inconsistent with the SCD type recorded for it (recorded ). 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 0000000000000..397f0cb9cab99 --- /dev/null +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala @@ -0,0 +1,331 @@ +/* + * 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. + */ + 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 + } + } + + /** + * Build the auxiliary table spec given an AutoCdc flow and the destination table it writes to. + * + * @param destinationTable the dataset that owns the auxiliary table + * @param destinationTableSchema 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 `destinationTable` + * @return the auxiliary-table spec + */ + def buildAuxiliaryTableSpecFor( + destinationTable: Table, + destinationTableSchema: StructType, + inputAutoCdcFlow: AutoCdcMergeFlow): AuxiliaryTableSpec = { + inputAutoCdcFlow.changeArgs.storedAsScdType match { + case ScdType.Type1 => + buildScd1AuxiliaryTableSpecFor( + destinationTable, + destinationTableSchema, + 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 destination + * table it writes to. + * + * @param destinationTable the dataset that owns the SCD1 auxiliary table + * @param destinationTableSchema 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 `destinationTable` + * @return the SCD1 auxiliary-table spec + */ + private def buildScd1AuxiliaryTableSpecFor( + destinationTable: Table, + destinationTableSchema: StructType, + inputAutoCdcFlow: AutoCdcMergeFlow + ): AuxiliaryTableSpec = { + val scd1AuxiliaryTableIdentifier = identifier(destinationTable.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 destination table. Retrieve those column schemas. + val keyFields = autoCdcKeyColumnNames.map { keyColumnName => + findFieldInDestinationSchema( + destinationTableSchema = destinationTableSchema, + destinationTableIdentifier = destinationTable.identifier, + autoCdcFlowIdentifier = inputAutoCdcFlow.identifier, + fieldName = keyColumnName, + resolver = resolver + ) + } + val cdcMetadataField = findFieldInDestinationSchema( + destinationTableSchema = destinationTableSchema, + destinationTableIdentifier = destinationTable.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. + destinationTable.format.map(TableCatalog.PROP_PROVIDER -> _) + + AutoCdcAuxiliaryTableSpec( + identifier = scd1AuxiliaryTableIdentifier, + schema = scd1AuxiliaryTableSchema, + properties = scd1AuxiliaryTableProperties, + targetTableIdentifier = destinationTable.identifier, + expectedKeyFields = keyFields, + expectedScdType = ScdType.Type1 + ) + } + + /** + * Resolve the [[StructField]] named `fieldName` in `destinationTableSchema` (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 destinationTableSchema 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 destinationTableIdentifier 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 findFieldInDestinationSchema( + destinationTableSchema: StructType, + fieldName: String, + resolver: Resolver, + destinationTableIdentifier: TableIdentifier, + autoCdcFlowIdentifier: TableIdentifier): StructField = { + destinationTableSchema.fields + .find(field => resolver(field.name, fieldName)) + .getOrElse( + throw SparkException.internalError( + s"Expected but unable to find column $fieldName in target table " + + s"$destinationTableIdentifier 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) + 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 0000000000000..cd3a02412bc5f --- /dev/null +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AuxiliaryTableSpec.scala @@ -0,0 +1,77 @@ +/* + * 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. + * + * 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 c9578ddd3b469..c3d1a8a03256b 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( + destinationTable = destinationTable, + destinationTableSchema = 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 38e5d3e66e517..17e37ab143fd5 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 @@ -38,6 +38,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 +105,25 @@ object DatasetManager extends Logging { transformer.transformTables { table => if (tablesToMaterialize.keySet.contains(table.identifier)) { try { - materializeTable( + val isFullRefresh = tablesToMaterialize(table.identifier).isFullRefresh + val materializedTable = materializeTable( resolvedDataflowGraph = resolvedDataflowGraph, table = table, - isFullRefresh = tablesToMaterialize(table.identifier).isFullRefresh, + isFullRefresh = isFullRefresh, context = context ) + // Auxiliary tables' lifecycle should follow the table that it is complimentary to. + // If this table has any auxiliary tables, materialize/full-refresh them + // accordingly. + resolvedDataflowGraph.auxiliaryTableSpecs.get(table.identifier).foreach { + auxiliaryTableSpec => + materializeAuxiliaryTable( + auxiliaryTableSpec = auxiliaryTableSpec, + isFullRefresh = isFullRefresh, + context = context + ) + } + materializedTable } catch { case NonFatal(e) => throw TableMaterializationException( @@ -307,20 +321,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) => @@ -349,6 +349,79 @@ object DatasetManager extends Logging { ) } + /** + * 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 = { + val auxiliaryTableCatalystIdentifier = auxiliaryTableSpec.identifier + + // Get the DSv2 catalog handler and identifier for the aux table. + val (catalog, auxiliaryTableIdentifier) = + PipelinesCatalogUtils.resolveTableCatalog(context.spark, auxiliaryTableCatalystIdentifier) + + if (isFullRefresh) { + // Intentionally DROP and not TRUNCATE on full refresh. The auxiliary table is an internal + // table whose identity does not need to be perserved on full refresh, and has metadata + // (ex. table properties) that should not persist between full refreshes. After the drop the + // table is recreated from scratch. + context.spark.sql( + s"DROP TABLE IF EXISTS ${auxiliaryTableCatalystIdentifier.quotedString}" + ) + createTableInCatalog( + 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 + ) + } + evolveTableInCatalog( + catalog = catalog, + tableIdentifier = auxiliaryTableIdentifier, + existingTable = existingAuxiliaryTable, + desiredSchema = auxiliaryTableSpec.schema, + properties = auxiliaryTableSpec.properties, + mergeWithExistingSchema = true + ) + case None => + createTableInCatalog( + 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. */ private def loadTableIfExists( catalog: TableCatalog, @@ -358,7 +431,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. @@ -383,11 +457,12 @@ object DatasetManager extends Logging { /** * Evolves the already-existing `existingTable` at `identifier` in place by diffing its schema and * (re)setting its properties. Partitioning/clustering cannot change in place, so no transforms are - * accepted here. + * 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 table properties to (re)set on evolve. 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 5b5ec776c0742..d83526d14ab89 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("") - ) - ) - } - } - - /** - * 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/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala index bbd8f7c8dbf90..40d76202b07da 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,38 @@ 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._ + + // Auxiliary-table provisioning moved out of flow execution and into the materialization + // phase, which a dry run deliberately skips (`dryRunPipeline` only resolves + validates + // the graph; it never calls `DatasetManager.materializeDatasets`). So a dry run must + // leave no auxiliary table behind -- it stays free of catalog side effects. + 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")), + "a dry run must not provision the AutoCDC auxiliary table" + ) + } + 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 fc7706c84e3ee..1e97ab2c2cd20 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 2100928bc68af..f86c592e3cf71 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 b6c8f2179b7f1..e11154164470e 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,64 @@ class AutoCdcScd1SchemaEvolutionSuite ) } + test("additive target-column evolution leaves the SCD1 auxiliary table schema unchanged") { + val session = spark + import session.implicits._ + + // The SCD1 auxiliary table carries only the AutoCDC key columns plus the CDC metadata + // column -- never any data columns. So when the target evolves additively (a new data + // column appears between runs), the target grows but the auxiliary schema must stay + // (keys + _cdc_metadata). This is the SCD1 counterpart to the SCD2 hidden-aux + // schema-evolution contract (`AutoCdcScd2SchemaEvolutionSuite`): SCD1 is immune to the + // motivating union-by-name bug precisely because its aux carries no data columns. The + // materialization-time aux create/evolve path must preserve this invariant across reruns. + 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 (appended last by mergeSchemas). The auxiliary + // schema must be byte-for-byte unchanged. + 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, + "additive target evolution must not alter the SCD1 auxiliary schema" + ) + } + test("broadening the column selection between runs adds the newly-included column to " + "the target") { val session = spark From 46d8c6a2f2d861ef3a4e5ce3430a9383f66fd236 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 1 Jul 2026 21:21:47 +0000 Subject: [PATCH 3/8] move destination table compat check to datasetmanager --- .../sql/pipelines/graph/DatasetManager.scala | 69 +++++++++++++++---- 1 file changed, 57 insertions(+), 12 deletions(-) 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 17e37ab143fd5..6b61b1901f796 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 => CatalogTable, TableCatalog, TableChange, @@ -106,24 +107,36 @@ object DatasetManager extends Logging { if (tablesToMaterialize.keySet.contains(table.identifier)) { try { val isFullRefresh = tablesToMaterialize(table.identifier).isFullRefresh - val materializedTable = materializeTable( + val (tableWithMaterializationMetadata, catalogTableEntity) = materializeTable( resolvedDataflowGraph = resolvedDataflowGraph, table = table, isFullRefresh = isFullRefresh, context = context ) // Auxiliary tables' lifecycle should follow the table that it is complimentary to. - // If this table has any auxiliary tables, materialize/full-refresh them - // accordingly. - resolvedDataflowGraph.auxiliaryTableSpecs.get(table.identifier).foreach { + // 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. + if (auxiliaryTableSpec.isInstanceOf[AutoCdcAuxiliaryTableSpec]) { + requireAutoCdcTargetSupportsRowLevelOps( + targetTable = table, + targetTableCatalogEntity = catalogTableEntity + ) + } + materializeAuxiliaryTable( auxiliaryTableSpec = auxiliaryTableSpec, isFullRefresh = isFullRefresh, context = context ) - } - materializedTable + } + tableWithMaterializationMetadata } catch { case NonFatal(e) => throw TableMaterializationException( @@ -257,13 +270,14 @@ 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, CatalogTable) = { logInfo(log"Materializing metadata for table ${MDC(LogKeys.TABLE_NAME, table.identifier)}.") val catalogManager = context.spark.sessionState.catalogManager val catalog = (table.identifier.catalog match { @@ -342,11 +356,42 @@ 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: CatalogTable): 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("") + ) + ) + } } /** From 9bee87da74578a0440724559fc387002e82aea05 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Thu, 2 Jul 2026 18:35:02 +0000 Subject: [PATCH 4/8] self-review --- .../sql/pipelines/graph/DatasetManager.scala | 52 ++++++++------- .../graph/AutoCdcAuxiliaryTableSuite.scala | 65 ++++++++++++++++++- 2 files changed, 92 insertions(+), 25 deletions(-) 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 0bea566f6df4d..8404fc357cc4d 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 @@ -123,13 +123,14 @@ object DatasetManager extends Logging { // 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. - if (auxiliaryTableSpec.isInstanceOf[AutoCdcAuxiliaryTableSpec]) { - requireAutoCdcTargetSupportsRowLevelOps( - targetTable = table, - targetTableCatalogEntity = catalogTableEntity - ) + auxiliaryTableSpec match { + case _: AutoCdcAuxiliaryTableSpec => + requireAutoCdcTargetSupportsRowLevelOps( + targetTable = table, + targetTableCatalogEntity = catalogTableEntity + ) } - + materializeAuxiliaryTable( auxiliaryTableSpec = auxiliaryTableSpec, isFullRefresh = isFullRefresh, @@ -279,16 +280,10 @@ object DatasetManager extends Logging { isFullRefresh: Boolean, 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 ) @@ -405,20 +400,33 @@ object DatasetManager extends Logging { auxiliaryTableSpec: AuxiliaryTableSpec, isFullRefresh: Boolean, context: PipelineUpdateContext): Unit = { - val auxiliaryTableCatalystIdentifier = auxiliaryTableSpec.identifier - // Get the DSv2 catalog handler and identifier for the aux table. val (catalog, auxiliaryTableIdentifier) = - PipelinesCatalogUtils.resolveTableCatalog(context.spark, auxiliaryTableCatalystIdentifier) + 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 perserved on full refresh, and has metadata + // 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. After the drop the - // table is recreated from scratch. - context.spark.sql( - s"DROP TABLE IF EXISTS ${auxiliaryTableCatalystIdentifier.quotedString}" + // table is recreated from scratch. Use the catalog API (rather than a SQL DROP) to stay + // consistent with the create/evolve calls around it; dropTable is a no-op if absent. + // + // 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." ) + + catalog.dropTable(auxiliaryTableIdentifier) + createTable( catalog = catalog, tableIdentifier = auxiliaryTableIdentifier, 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 9fb6070c01e7a..fb368198f72d3 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)) + } } From ab5e1226b9b368d6d1f1fda73acdae7a0a561182 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Thu, 2 Jul 2026 18:54:16 +0000 Subject: [PATCH 5/8] test cleanup --- ...oCdcScd1AuxiliaryTableDurabilitySuite.scala | 4 ---- .../AutoCdcScd1SchemaEvolutionSuite.scala | 18 +++++------------- 2 files changed, 5 insertions(+), 17 deletions(-) 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 40d76202b07da..fa1062193f3f5 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 @@ -186,10 +186,6 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite val session = spark import session.implicits._ - // Auxiliary-table provisioning moved out of flow execution and into the materialization - // phase, which a dry run deliberately skips (`dryRunPipeline` only resolves + validates - // the graph; it never calls `DatasetManager.materializeDatasets`). So a dry run must - // leave no auxiliary table behind -- it stays free of catalog side effects. spark.sql( s"CREATE TABLE $catalog.$namespace.target " + s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)" 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 e11154164470e..fb3094aa2eb45 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 @@ -226,13 +226,6 @@ class AutoCdcScd1SchemaEvolutionSuite val session = spark import session.implicits._ - // The SCD1 auxiliary table carries only the AutoCDC key columns plus the CDC metadata - // column -- never any data columns. So when the target evolves additively (a new data - // column appears between runs), the target grows but the auxiliary schema must stay - // (keys + _cdc_metadata). This is the SCD1 counterpart to the SCD2 hidden-aux - // schema-evolution contract (`AutoCdcScd2SchemaEvolutionSuite`): SCD1 is immune to the - // motivating union-by-name bug precisely because its aux carries no data columns. The - // materialization-time aux create/evolve path must preserve this invariant across reruns. spark.sql( s"CREATE TABLE $catalog.$namespace.target " + s"(id INT NOT NULL, version BIGINT NOT NULL, $cdcMetadataDdl)" @@ -263,10 +256,12 @@ class AutoCdcScd1SchemaEvolutionSuite "auxiliary schema after run #1 should be the keys plus the CDC metadata column" ) - // Run #2: `name` is added to the target (appended last by mergeSchemas). The auxiliary - // schema must be byte-for-byte unchanged. + // 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( @@ -274,10 +269,7 @@ class AutoCdcScd1SchemaEvolutionSuite Row(2, 2L, cdcMeta(None, Some(2L)), "bob") ) ) - assert( - spark.table(auxTableNameFor("target")).schema.fieldNames.toSeq == expectedAuxSchema, - "additive target evolution must not alter the SCD1 auxiliary schema" - ) + assert(spark.table(auxTableNameFor("target")).schema.fieldNames.toSeq == expectedAuxSchema) } test("broadening the column selection between runs adds the newly-included column to " + From e4a3d90bb3e17136a7fb4ce82a8a9bd3cb9d4ad2 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Thu, 2 Jul 2026 19:11:17 +0000 Subject: [PATCH 6/8] linting --- .../sql/pipelines/graph/AutoCdcAuxiliaryTable.scala | 6 +++--- .../spark/sql/pipelines/graph/AuxiliaryTableSpec.scala | 4 ++-- .../spark/sql/pipelines/graph/DatasetManager.scala | 9 ++++----- .../graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala | 5 +---- 4 files changed, 10 insertions(+), 14 deletions(-) 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 index 397f0cb9cab99..34f092070be50 100644 --- 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 @@ -268,9 +268,9 @@ object AutoCdcAuxiliaryTable { } /** - * 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. + * 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, 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 index cd3a02412bc5f..14edae86db3fb 100644 --- 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 @@ -61,8 +61,8 @@ sealed trait AuxiliaryTableSpec { * 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 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 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 8404fc357cc4d..0dc808a5d3ba9 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 @@ -412,9 +412,7 @@ object DatasetManager extends Logging { 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. After the drop the - // table is recreated from scratch. Use the catalog API (rather than a SQL DROP) to stay - // consistent with the create/evolve calls around it; dropTable is a no-op if absent. + // (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 @@ -424,9 +422,10 @@ object DatasetManager extends Logging { 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, 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 fa1062193f3f5..588ebfe5192a5 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 @@ -203,10 +203,7 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite val updateCtx = TestPipelineUpdateContext(spark, ctx.toDataflowGraph, storageRoot) updateCtx.pipelineExecution.dryRunPipeline() - assert( - !spark.catalog.tableExists(auxTableNameFor("target")), - "a dry run must not provision the AutoCDC auxiliary table" - ) + assert(spark.catalog.tableExists(auxTableNameFor("target"))) } test("if the AutoCDC auxiliary table is dropped between runs, it is transparently " + From b9bbf539eec7eeb9117c1f6e032f99bb1c5573e7 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Fri, 3 Jul 2026 17:31:58 +0000 Subject: [PATCH 7/8] fix test --- .../graph/AutoCdcScd1AuxiliaryTableDurabilitySuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 588ebfe5192a5..10b8d0dfc5ae5 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 @@ -203,7 +203,7 @@ class AutoCdcScd1AuxiliaryTableDurabilitySuite val updateCtx = TestPipelineUpdateContext(spark, ctx.toDataflowGraph, storageRoot) updateCtx.pipelineExecution.dryRunPipeline() - assert(spark.catalog.tableExists(auxTableNameFor("target"))) + assert(!spark.catalog.tableExists(auxTableNameFor("target"))) } test("if the AutoCDC auxiliary table is dropped between runs, it is transparently " + From 07f3252e6bf6b76eeca9f0d79b4c4a9bc83921b3 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Thu, 9 Jul 2026 17:02:41 +0000 Subject: [PATCH 8/8] address review comments: rename destination->target, clarify docs, tighten visibility - Rename destination{Table,TableSchema} -> target{Table,TableSchema} and findFieldInDestinationSchema -> findFieldInTargetSchema in the aux-table spec builders, aligning with the spec's targetTableIdentifier field and the "target table" error wording. Update the DataflowGraph call site. - Fix stale scaladoc on AuxiliaryTableSpec: aux tables are only dropped on full refresh when the target still has a companion spec in the current pipeline definition; orphaned aux tables are intentionally left in place. - Fix "complimentary" -> "complementary" typo in DatasetManager. - Add a comment in validateNoKeyColumnDrift noting it first validates the existing aux table is internally consistent before checking drift. - Make serializeKeyColumnNames/parseKeyColumnNames private[graph]. Co-authored-by: Isaac --- .../graph/AutoCdcAuxiliaryTable.scala | 82 ++++++++++--------- .../pipelines/graph/AuxiliaryTableSpec.scala | 5 +- .../sql/pipelines/graph/DataflowGraph.scala | 4 +- .../sql/pipelines/graph/DatasetManager.scala | 2 +- 4 files changed, 50 insertions(+), 43 deletions(-) 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 index 34f092070be50..6fcf2fc4cdab2 100644 --- 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 @@ -65,7 +65,7 @@ object AutoCdcAuxiliaryTable { * Round-trips an empty list as `[]`; callers are expected to enforce a non-empty key set * upstream. */ - def serializeKeyColumnNames(names: Seq[String]): String = { + private[graph] def serializeKeyColumnNames(names: Seq[String]): String = { compact(JArray(names.map(JString(_)).toList)) } @@ -74,7 +74,7 @@ object AutoCdcAuxiliaryTable { * Round-trips an empty list as `[]`; callers are expected to enforce a non-empty key set * upstream. */ - def parseKeyColumnNames(raw: String): Option[Seq[String]] = { + private[graph] def parseKeyColumnNames(raw: String): Option[Seq[String]] = { val parsed = try Some(parse(raw)) catch { case NonFatal(_) => None } parsed.flatMap { case JArray(elems) => @@ -85,24 +85,24 @@ object AutoCdcAuxiliaryTable { } /** - * Build the auxiliary table spec given an AutoCdc flow and the destination table it writes to. + * Build the auxiliary table spec given an AutoCdc flow and the target table it writes to. * - * @param destinationTable the dataset that owns the auxiliary table - * @param destinationTableSchema 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 `destinationTable` + * @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( - destinationTable: Table, - destinationTableSchema: StructType, + targetTable: Table, + targetTableSchema: StructType, inputAutoCdcFlow: AutoCdcMergeFlow): AuxiliaryTableSpec = { inputAutoCdcFlow.changeArgs.storedAsScdType match { case ScdType.Type1 => buildScd1AuxiliaryTableSpecFor( - destinationTable, - destinationTableSchema, + targetTable, + targetTableSchema, inputAutoCdcFlow ) case ScdType.Type2 => @@ -115,41 +115,41 @@ object AutoCdcAuxiliaryTable { } /** - * Build the SCD1 auxiliary table spec given the AutoCdc flow's declared keys and the destination + * Build the SCD1 auxiliary table spec given the AutoCdc flow's declared keys and the target * table it writes to. * - * @param destinationTable the dataset that owns the SCD1 auxiliary table - * @param destinationTableSchema 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 `destinationTable` + * @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( - destinationTable: Table, - destinationTableSchema: StructType, + targetTable: Table, + targetTableSchema: StructType, inputAutoCdcFlow: AutoCdcMergeFlow ): AuxiliaryTableSpec = { - val scd1AuxiliaryTableIdentifier = identifier(destinationTable.identifier) + 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 destination table. Retrieve those column schemas. + // schema in its corresponding target table. Retrieve those column schemas. val keyFields = autoCdcKeyColumnNames.map { keyColumnName => - findFieldInDestinationSchema( - destinationTableSchema = destinationTableSchema, - destinationTableIdentifier = destinationTable.identifier, + findFieldInTargetSchema( + targetTableSchema = targetTableSchema, + targetTableIdentifier = targetTable.identifier, autoCdcFlowIdentifier = inputAutoCdcFlow.identifier, fieldName = keyColumnName, resolver = resolver ) } - val cdcMetadataField = findFieldInDestinationSchema( - destinationTableSchema = destinationTableSchema, - destinationTableIdentifier = destinationTable.identifier, + val cdcMetadataField = findFieldInTargetSchema( + targetTableSchema = targetTableSchema, + targetTableIdentifier = targetTable.identifier, autoCdcFlowIdentifier = inputAutoCdcFlow.identifier, fieldName = AutoCdcReservedNames.cdcMetadataColName, resolver = resolver @@ -166,42 +166,42 @@ object AutoCdcAuxiliaryTable { 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. - destinationTable.format.map(TableCatalog.PROP_PROVIDER -> _) + targetTable.format.map(TableCatalog.PROP_PROVIDER -> _) AutoCdcAuxiliaryTableSpec( identifier = scd1AuxiliaryTableIdentifier, schema = scd1AuxiliaryTableSchema, properties = scd1AuxiliaryTableProperties, - targetTableIdentifier = destinationTable.identifier, + targetTableIdentifier = targetTable.identifier, expectedKeyFields = keyFields, expectedScdType = ScdType.Type1 ) } /** - * Resolve the [[StructField]] named `fieldName` in `destinationTableSchema` (the AutoCDC target's + * 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 destinationTableSchema the AutoCDC target's evolved schema to resolve against + * @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 destinationTableIdentifier the AutoCDC target's identifier, named in the error message + * @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 findFieldInDestinationSchema( - destinationTableSchema: StructType, + private def findFieldInTargetSchema( + targetTableSchema: StructType, fieldName: String, resolver: Resolver, - destinationTableIdentifier: TableIdentifier, + targetTableIdentifier: TableIdentifier, autoCdcFlowIdentifier: TableIdentifier): StructField = { - destinationTableSchema.fields + targetTableSchema.fields .find(field => resolver(field.name, fieldName)) .getOrElse( throw SparkException.internalError( s"Expected but unable to find column $fieldName in target table " + - s"$destinationTableIdentifier written to by AutoCDC flow $autoCdcFlowIdentifier." + s"$targetTableIdentifier written to by AutoCDC flow $autoCdcFlowIdentifier." ) ) } @@ -224,6 +224,10 @@ object AutoCdcAuxiliaryTable { 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)) 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 index 14edae86db3fb..55cd19a1f8c43 100644 --- 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 @@ -24,7 +24,10 @@ 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. + * 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 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 c3d1a8a03256b..c5210976d3f98 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 @@ -202,8 +202,8 @@ case class DataflowGraph( .collectFirst { case f: AutoCdcMergeFlow => f } .map { autoCdcFlow => val spec = AutoCdcAuxiliaryTable.buildAuxiliaryTableSpecFor( - destinationTable = destinationTable, - destinationTableSchema = inferredSchema(destinationTableIdentifier), + targetTable = destinationTable, + targetTableSchema = inferredSchema(destinationTableIdentifier), inputAutoCdcFlow = autoCdcFlow ) destinationTableIdentifier -> spec 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 0dc808a5d3ba9..4f96f2f709908 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 @@ -113,7 +113,7 @@ object DatasetManager extends Logging { isFullRefresh = isFullRefresh, context = context ) - // Auxiliary tables' lifecycle should follow the table that it is complimentary to. + // 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 {