szehon-ho commented on code in PR #56160:
URL: https://github.com/apache/spark/pull/56160#discussion_r3315366807
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -330,11 +330,51 @@ object AutoCdcAuxiliaryTable {
* Reserved table property key set on the auxiliary table to record which
SCD strategy it
* serves.
*/
- val scdTypePropertyKey: String =
s"${PipelinesTableProperties.pipelinesPrefix}autocdc.scd_type"
+ 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 = {
Review Comment:
Nit (non-blocking): the `org.json4s.*` imports are scoped inside both
`serializeKeyColumnNames` and `parseKeyColumnNames` rather than declared at the
top of the file. The rest of `FlowExecution.scala` (and the Spark codebase
generally) keeps imports file-top; scoping them locally is unusual enough to
draw the eye on a future read. Worth hoisting for consistency.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -343,44 +383,92 @@ 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
+
/** 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
/**
- * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
- * and return its [[TableIdentifier]].
+ * Create the auxiliary table for [[destination]] if it does not already
exist and return its
+ * [[TableIdentifier]].
*
- * Note that this is `CREATE TABLE IF NOT EXISTS`: when the aux table
already exists, its
- * schema is left untouched and `auxiliaryTableSchema` is ignored. For SCD1,
they keys must be
- * invariant across executions and the CDC metadata will always be present,
so this is correct.
+ * 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)
- // The auxiliary table inherits the target's format so MERGE semantics
line up. When the
- // target's format is unspecified (None), omit the USING clause and fall
back to the
- // session's default source provider.
- val usingClause = destination.format.map(fmt => s"USING
$fmt").getOrElse("")
- val tblPropertiesClause =
- s"TBLPROPERTIES ('${AutoCdcAuxiliaryTable.scdTypePropertyKey}' = " +
- s"'${changeArgs.storedAsScdType.label}')"
- spark.sql(
- s"""CREATE TABLE IF NOT EXISTS
- |${auxIdent.quotedString}
- |(${auxiliaryTableSchema.toDDL}) $usingClause
$tblPropertiesClause""".stripMargin
- )
+ val (catalog, v2Identifier) =
PipelinesCatalogUtils.resolveTableCatalog(spark, auxIdent)
+
+ if (!catalog.tableExists(v2Identifier)) {
+ val properties = scala.collection.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
}
+ /**
+ * Returns the resolved AutoCDC key column names as they appear in the
auxiliary schema, in
+ * `changeArgs.keys` declaration order.
+ */
+ private def auxiliaryKeyColumnNames: Seq[String] = {
Review Comment:
Nit (non-blocking): `auxiliaryKeyColumnNames` and the `expectedKeyFields`
lookup in `validateNoAutoCdcKeyDrift` both walk `changeArgs.keys` against
`auxiliaryTableSchema.fields` with the same resolver, just returning different
shapes (`Seq[String]` vs `Seq[StructField]`). ~10 lines duplicated, with the
same `internalError` invariant on both sides. Not worth a refactor for two
callers, but if a third one shows up, factor them into a shared helper.
--
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]
