szehon-ho commented on code in PR #56160: URL: https://github.com/apache/spark/pull/56160#discussion_r3314522306
########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/PipelinesCatalogUtils.scala: ########## @@ -0,0 +1,49 @@ +/* + * 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.util + +import org.apache.spark.SparkException +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog} + +/** Catalog-resolution helpers shared across the pipelines module. */ +object PipelinesCatalogUtils { + + /** + * Resolve a v1 [[TableIdentifier]] to a `(TableCatalog, Identifier)` pair usable against the + * v2 connector APIs. If `ident.catalog` is unset, falls back to the session's + * `currentCatalog`. The catalog is required to be a [[TableCatalog]]; namespace must be + * non-empty. + */ + def resolveTableCatalog( + spark: SparkSession, + ident: TableIdentifier): (TableCatalog, Identifier) = { + val catalogManager = spark.sessionState.catalogManager + val catalog = ident.catalog + .map(catalogManager.catalog) + .getOrElse(catalogManager.currentCatalog) + .asInstanceOf[TableCatalog] Review Comment: The original helper this is extracted from did: ```scala val catalog = catalogPlugin match { case t: TableCatalog => t case _ => throw QueryCompilationErrors.missingCatalogTablesAbilityError(catalogPlugin) } ``` The new `.asInstanceOf[TableCatalog]` will throw a raw `ClassCastException` on a non-`TableCatalog` plugin instead of the structured `QueryCompilationError`. Unrelated to the drift-validation change and looks like a silent regression introduced by the extract — can we keep the `match` so the structured error is preserved? ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala: ########## @@ -331,6 +331,33 @@ object AutoCdcAuxiliaryTable { * serves. */ val scdTypePropertyKey: String = s"${PipelinesTableProperties.pipelinesPrefix}autocdc.scd_type" + + /** + * Table property recording the auxiliary table's 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 = + PipelinesTableProperties.pipelinesPrefix + "autoCdc.keyColumnNames" Review Comment: Worth catching before this ships: the existing reserved property right above uses `pipelines.autocdc.scd_type` (lowercase + snake_case), but the new one uses `pipelines.autoCdc.keyColumnNames` (camelCase). Two different conventions on the same auxiliary table. Suggest renaming to `pipelines.autocdc.key_column_names` for parity with `scdTypePropertyKey`. This ends up in user-visible `SHOW TBLPROPERTIES` output, so the rename is much cheaper now than later. ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala: ########## @@ -380,4 +388,124 @@ class AutoCdcMergeFlow( ) } } + + /** + * If the auxiliary table for this flow's destination already exists, validate that the + * AutoCDC key columns 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. + */ + private def validateNoAutoCdcKeyDriftIfAuxTableExists(): Unit = { + val auxIdent = AutoCdcAuxiliaryTable.identifier(flow.destinationIdentifier) + 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: org.apache.spark.sql.connector.catalog.Table, + auxIdent: TableIdentifier): Unit = { + val existingAuxSchema = CatalogV2Util.v2ColumnsToStructType(existingAuxTable.columns()) + val resolver = spark.sessionState.conf.resolver + + val expectedKeyFields: Seq[StructField] = changeArgs.keys.map { key => + userSelectedSchema.fields + .find(field => resolver(field.name, key.name)) + .getOrElse( + // Construction of [[userSelectedSchema]] already enforces all of the user-specified + // keys are indeed in the selected schema, so if we don't find a key it is truly an + // internal error. + throw SparkException.internalError( + s"Key column '${key.name}' was not found in the AutoCDC flow's selected schema." + ) + ) + } + val expectedKeySchema = StructType(expectedKeyFields) + + 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 SparkException( + errorClass = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_KEY_COLUMN_MISSING", + messageParameters = Map( + "flowName" -> flow.identifier.unquotedString, + "auxTableName" -> auxIdent.unquotedString, + "keyColumnName" -> name, + "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty + ), + cause = null + ) + ) + } + + val drifted = + // Arity drift (added or dropped keys). + recordedKeyFields.length != expectedKeySchema.length || + // Name or dataType drift: every expected key must have a same-name (resolver-aware) + // recorded counterpart with a matching dataType. An expected name that is absent from + // the recorded set indicates the key set has changed; a same-name recorded counterpart + // with a different dataType indicates a key dataType change. + expectedKeyFields.exists { expected => + recordedKeyFields.find(rf => resolver(rf.name, expected.name)) match { + case None => true + case Some(recorded) => recorded.dataType != expected.dataType + } + } Review Comment: Two notes on the drift predicate: 1. `recorded.dataType != expected.dataType` compares `DataType` only; `StructField.nullable` and `StructField.metadata` are intentionally tolerated. Worth a one-line comment documenting that, since the error message below renders both schemas via `StructType.toDDL`, which *does* include `NOT NULL` — so a future bug that lets nullability diverge would render a confusing message ("expected `id INT NOT NULL`, recorded `id INT`") without the validator actually flagging it. 2. Could we add a positive test (`AutoCdcMergeFlow allows nullable mismatch`) to lock the tolerance in? Right now no test exercises nullability divergence, so the contract is implicit. ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala: ########## @@ -271,6 +275,10 @@ class AutoCdcMergeFlow( selectedSchema } + // If the auxiliary table corresponding to the target already exists, verify the user is trigging Review Comment: Typo: `trigging` → `triggering`. ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala: ########## @@ -331,6 +331,33 @@ object AutoCdcAuxiliaryTable { * serves. */ val scdTypePropertyKey: String = s"${PipelinesTableProperties.pipelinesPrefix}autocdc.scd_type" + + /** + * Table property recording the auxiliary table's 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 = + PipelinesTableProperties.pipelinesPrefix + "autoCdc.keyColumnNames" + + /** Serialize key column names to the JSON form stored at [[keyColumnNamesProperty]]. */ + def serializeKeyColumnNames(names: Seq[String]): String = { + import org.json4s.JsonAST.{JArray, JString} + import org.json4s.jackson.JsonMethods.compact + compact(JArray(names.map(JString(_)).toList)) + } + + /** Parse a [[keyColumnNamesProperty]] value. `None` if it is not a JSON array of strings. */ Review Comment: The test `round-trip preserves the empty list` correctly notes that `Seq.empty` is not user-reachable — callers (`AutoCdcMergeFlow`) reject empty key sets upstream — but `serializeKeyColumnNames` / `parseKeyColumnNames` are public helpers on `AutoCdcAuxiliaryTable`. Worth one line in each Scaladoc: "Round-trips an empty list as `[]`; callers are expected to enforce a non-empty key set upstream." That way the contract is discoverable from the helper itself, not only from the test. ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala: ########## @@ -343,44 +370,91 @@ trait AutoCdcMergeWriteBase { /** 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 Review Comment: The docstring claims `identifier` is "used as `flowName` in error messages emitted by this mixin", but I can't find any method in `AutoCdcMergeWriteBase` that consumes it — `createAuxiliaryTableIfNotExists`, `auxiliaryKeyColumnNames`, and `requireDestinationSupportsRowLevelOps` all reference `destination.identifier`, not `this.identifier`. Two options: - Remove the abstract member (it adds a contract obligation for subclasses for no benefit), or - Actually wire it in — e.g. the internal-error in `auxiliaryKeyColumnNames` would be more diagnosable with a `flowName`. ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala: ########## @@ -271,6 +275,10 @@ class AutoCdcMergeFlow( selectedSchema } + // If the auxiliary table corresponding to the target already exists, verify the user is trigging + // an AutoCDC transformation to that target using the same keys. + validateNoAutoCdcKeyDriftIfAuxTableExists() Review Comment: This puts a catalog read into `AutoCdcMergeFlow`'s constructor, which is invoked from `CoreDataflowNodeProcessor.transformAutoCdcFlowToResolvedFlow` — i.e. during graph resolution, not execution. `Scd1MergeStreamingWrite.startStream` explicitly states the resolution-is-side-effect-free invariant: > `// Flow resolution must also stay side-effect free (e.g. for dry runs).` The sibling check `requireDestinationSupportsRowLevelOps()` lives in the `Scd1MergeStreamingWrite` constructor for precisely this reason. Drift validation has the same shape (depends only on catalog metadata of the destination + its aux table) — could we move it next to `requireDestinationSupportsRowLevelOps()` so both pre-flight checks live at the execution layer and dry-runs stay catalog-free? If there's a reason it has to be at resolution time, a comment here explaining why would help. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala: ########## @@ -19,13 +19,17 @@ package org.apache.spark.sql.pipelines.autocdc import java.util.Locale +import org.scalatest.BeforeAndAfterEach Review Comment: Import group ordering is wrong here: `org.scalatest.BeforeAndAfterEach` is placed before `scala.util.Success`, but Spark's convention is `java.*` → `scala.*` → third-party → `org.apache.spark.*`. `AutoCdcGraphExecutionTestMixin.scala` in this same PR gets it right. Suggest: ```scala import java.util.Locale import scala.util.Success import org.scalatest.BeforeAndAfterEach import org.apache.spark.sql.{functions => F, AnalysisException, Column, QueryTest} ``` ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala: ########## @@ -380,4 +388,124 @@ class AutoCdcMergeFlow( ) } } + + /** + * If the auxiliary table for this flow's destination already exists, validate that the + * AutoCDC key columns 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. + */ + private def validateNoAutoCdcKeyDriftIfAuxTableExists(): Unit = { + val auxIdent = AutoCdcAuxiliaryTable.identifier(flow.destinationIdentifier) + 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: org.apache.spark.sql.connector.catalog.Table, + auxIdent: TableIdentifier): Unit = { + val existingAuxSchema = CatalogV2Util.v2ColumnsToStructType(existingAuxTable.columns()) + val resolver = spark.sessionState.conf.resolver + + val expectedKeyFields: Seq[StructField] = changeArgs.keys.map { key => + userSelectedSchema.fields + .find(field => resolver(field.name, key.name)) + .getOrElse( + // Construction of [[userSelectedSchema]] already enforces all of the user-specified + // keys are indeed in the selected schema, so if we don't find a key it is truly an + // internal error. + throw SparkException.internalError( + s"Key column '${key.name}' was not found in the AutoCDC flow's selected schema." + ) + ) + } + val expectedKeySchema = StructType(expectedKeyFields) + + 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 SparkException( + errorClass = "AUTOCDC_INVALID_STATE.AUXILIARY_TABLE_KEY_COLUMN_MISSING", + messageParameters = Map( + "flowName" -> flow.identifier.unquotedString, + "auxTableName" -> auxIdent.unquotedString, + "keyColumnName" -> name, + "propertyName" -> AutoCdcAuxiliaryTable.keyColumnNamesProperty + ), + cause = null + ) + ) + } + + val drifted = + // Arity drift (added or dropped keys). + recordedKeyFields.length != expectedKeySchema.length || + // Name or dataType drift: every expected key must have a same-name (resolver-aware) + // recorded counterpart with a matching dataType. An expected name that is absent from + // the recorded set indicates the key set has changed; a same-name recorded counterpart + // with a different dataType indicates a key dataType change. + expectedKeyFields.exists { expected => + recordedKeyFields.find(rf => resolver(rf.name, expected.name)) match { + case None => true + case Some(recorded) => recorded.dataType != expected.dataType + } + } + + if (drifted) { + throw new AnalysisException( + errorClass = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT", + messageParameters = Map( + "flowName" -> flow.identifier.unquotedString, + "auxTableName" -> auxIdent.unquotedString, + "expectedKeySchema" -> expectedKeySchema.toDDL, + "recordedKeySchema" -> StructType(recordedKeyFields).toDDL + ) + ) Review Comment: All four sub-classes of `AUTOCDC_INVALID_STATE` share the same parent error class and `sqlState = "42000"`, but `KEY_SCHEMA_DRIFT` is thrown as `AnalysisException` while `AUXILIARY_TABLE_PROPERTY_MISSING`, `AUXILIARY_TABLE_PROPERTY_MALFORMED`, and `AUXILIARY_TABLE_KEY_COLUMN_MISSING` are thrown as `SparkException`. The tests `intercept` on each accordingly, so the asymmetry gets locked in. All four are surfaced during construction-time analysis, so `AnalysisException` for the whole family seems most consistent and lets callers catch a single type. At minimum a comment explaining the rationale would help; ideally just unify on one. ########## common/utils/src/main/resources/error/error-conditions.json: ########## @@ -209,6 +209,34 @@ ], "sqlState" : "22023" }, + "AUTOCDC_INVALID_STATE" : { + "message" : [ + "AutoCDC flow <flowName> detected an invalid state:" + ], + "subClass" : { + "AUXILIARY_TABLE_KEY_COLUMN_MISSING" : { + "message" : [ + "The auxiliary table <auxTableName> is missing key column `<keyColumnName>` that is recorded in its <propertyName> table property. The auxiliary table schema may be corrupted or have been modified externally. Perform a full refresh of the target table to recreate the auxiliary table." Review Comment: Minor cosmetic asymmetry: this sub-class wraps `<keyColumnName>` in backticks (`` `<keyColumnName>` ``), but the `KEY_SCHEMA_DRIFT` message inlines `<expectedKeySchema>` / `<recordedKeySchema>` via `toDDL` (which only backticks identifiers that need quoting). Consider dropping the backticks here for consistent rendering across the sub-classes — or alternatively, backtick the rendered schemas in `KEY_SCHEMA_DRIFT` for parity in the other direction. ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala: ########## @@ -331,6 +331,33 @@ object AutoCdcAuxiliaryTable { * serves. */ val scdTypePropertyKey: String = s"${PipelinesTableProperties.pipelinesPrefix}autocdc.scd_type" + + /** + * Table property recording the auxiliary table's 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 = + PipelinesTableProperties.pipelinesPrefix + "autoCdc.keyColumnNames" + + /** Serialize key column names to the JSON form stored at [[keyColumnNamesProperty]]. */ + def serializeKeyColumnNames(names: Seq[String]): String = { + import org.json4s.JsonAST.{JArray, JString} + import org.json4s.jackson.JsonMethods.compact + compact(JArray(names.map(JString(_)).toList)) + } + + /** Parse a [[keyColumnNamesProperty]] value. `None` if it is not a JSON array of strings. */ + def parseKeyColumnNames(raw: String): Option[Seq[String]] = { + import org.json4s.JsonAST.{JArray, JString} + import org.json4s.jackson.JsonMethods.parse + scala.util.Try(parse(raw)).toOption.flatMap { Review Comment: `scala.util.Try` catches every `NonFatal` *and* some throwables that `Try` happens to allow (`InterruptedException`, etc. depending on Scala version), which is broader than needed — `JsonMethods.parse` throws `JsonParseException` on malformed input. A tighter `try { Some(parse(raw)) } catch { case NonFatal(_) => None }` would avoid swallowing programming errors. The Spark codebase uses the `Try` idiom elsewhere too, so this is a nit, not a blocker. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
