AnishMahto commented on code in PR #56160:
URL: https://github.com/apache/spark/pull/56160#discussion_r3315009900
##########
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:
Converted all the `AnalysisException` for consistency.
--
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]
