anew commented on code in PR #56686:
URL: https://github.com/apache/spark/pull/56686#discussion_r3531968868
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -211,27 +211,32 @@
},
"AUTOCDC_INVALID_STATE" : {
"message" : [
- "AutoCDC flow <flowName> detected an invalid state:"
+ "Detected an invalid AutoCDC state for target table <tableName>:"
],
"subClass" : {
"AUXILIARY_TABLE_KEY_COLUMN_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing key column
<keyColumnName> that is recorded in its <propertyName> table property. The
auxiliary table schema may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing key column <keyColumnName>
that is recorded in its <propertyName> table property. The auxiliary table
schema may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
Review Comment:
should this message have the name of the target table? Without any table
name in the message it is very hard to know.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -211,27 +211,32 @@
},
"AUTOCDC_INVALID_STATE" : {
"message" : [
- "AutoCDC flow <flowName> detected an invalid state:"
+ "Detected an invalid AutoCDC state for target table <tableName>:"
],
"subClass" : {
"AUXILIARY_TABLE_KEY_COLUMN_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing key column
<keyColumnName> that is recorded in its <propertyName> table property. The
auxiliary table schema may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing key column <keyColumnName>
that is recorded in its <propertyName> table property. The auxiliary table
schema may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
]
},
"AUXILIARY_TABLE_PROPERTY_MALFORMED" : {
"message" : [
- "The auxiliary table <auxTableName> has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
+ "The internal auxiliary table has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
Review Comment:
ditto
##########
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]] = {
Review Comment:
isn't there a standard way to parse a Json string? Why do we need this
custom way?
##########
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
Review Comment:
ditto
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -211,27 +211,32 @@
},
"AUTOCDC_INVALID_STATE" : {
"message" : [
- "AutoCDC flow <flowName> detected an invalid state:"
+ "Detected an invalid AutoCDC state for target table <tableName>:"
],
"subClass" : {
"AUXILIARY_TABLE_KEY_COLUMN_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing key column
<keyColumnName> that is recorded in its <propertyName> table property. The
auxiliary table schema may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing key column <keyColumnName>
that is recorded in its <propertyName> table property. The auxiliary table
schema may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
]
},
"AUXILIARY_TABLE_PROPERTY_MALFORMED" : {
"message" : [
- "The auxiliary table <auxTableName> has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
+ "The internal auxiliary table has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
]
},
"AUXILIARY_TABLE_PROPERTY_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing the required
<propertyName> table property; cannot validate AutoCDC key columns. The
auxiliary table metadata may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing the required <propertyName>
table property; cannot validate AutoCDC key columns. The auxiliary table
metadata may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
]
},
"KEY_SCHEMA_DRIFT" : {
"message" : [
- "The AutoCDC flow's current key columns <expectedKeySchema> do not
match the keys recorded in the auxiliary table <auxTableName> (recorded keys
<recordedKeySchema>). AutoCDC does not support changing key columns or their
types across incremental pipeline runs. To change keys, perform a full refresh
of the target table."
+ "One or more AutoCDC flows writing to the target use key columns
<expectedKeySchema>, which are inconsistent with the keys recorded for it
(recorded <recordedKeySchema>). AutoCDC does not support changing key columns
or their types across incremental pipeline runs. Correct the conflicting
flow(s) or perform a full refresh of the target table."
+ ]
+ },
+ "SCD_TYPE_DRIFT" : {
+ "message" : [
+ "One or more AutoCDC flows writing to the target use SCD type
<expectedScdType>, which is inconsistent with the SCD type recorded for it
(recorded <recordedScdType>). AutoCDC does not support changing a target's SCD
type across incremental pipeline runs. Correct the conflicting flow(s) or
perform a full refresh of the target table."
Review Comment:
ditto
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -211,27 +211,32 @@
},
"AUTOCDC_INVALID_STATE" : {
"message" : [
- "AutoCDC flow <flowName> detected an invalid state:"
+ "Detected an invalid AutoCDC state for target table <tableName>:"
],
"subClass" : {
"AUXILIARY_TABLE_KEY_COLUMN_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing key column
<keyColumnName> that is recorded in its <propertyName> table property. The
auxiliary table schema may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing key column <keyColumnName>
that is recorded in its <propertyName> table property. The auxiliary table
schema may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
]
},
"AUXILIARY_TABLE_PROPERTY_MALFORMED" : {
"message" : [
- "The auxiliary table <auxTableName> has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
+ "The internal auxiliary table has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
]
},
"AUXILIARY_TABLE_PROPERTY_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing the required
<propertyName> table property; cannot validate AutoCDC key columns. The
auxiliary table metadata may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing the required <propertyName>
table property; cannot validate AutoCDC key columns. The auxiliary table
metadata may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
]
},
"KEY_SCHEMA_DRIFT" : {
"message" : [
- "The AutoCDC flow's current key columns <expectedKeySchema> do not
match the keys recorded in the auxiliary table <auxTableName> (recorded keys
<recordedKeySchema>). AutoCDC does not support changing key columns or their
types across incremental pipeline runs. To change keys, perform a full refresh
of the target table."
+ "One or more AutoCDC flows writing to the target use key columns
<expectedKeySchema>, which are inconsistent with the keys recorded for it
(recorded <recordedKeySchema>). AutoCDC does not support changing key columns
or their types across incremental pipeline runs. Correct the conflicting
flow(s) or perform a full refresh of the target table."
Review Comment:
ditto
##########
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"
Review Comment:
would it make sense to move this to `PipelinesTableProperties`? That would
keep all such properties defined in a central place
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -211,27 +211,32 @@
},
"AUTOCDC_INVALID_STATE" : {
"message" : [
- "AutoCDC flow <flowName> detected an invalid state:"
+ "Detected an invalid AutoCDC state for target table <tableName>:"
],
"subClass" : {
"AUXILIARY_TABLE_KEY_COLUMN_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing key column
<keyColumnName> that is recorded in its <propertyName> table property. The
auxiliary table schema may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing key column <keyColumnName>
that is recorded in its <propertyName> table property. The auxiliary table
schema may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
]
},
"AUXILIARY_TABLE_PROPERTY_MALFORMED" : {
"message" : [
- "The auxiliary table <auxTableName> has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
+ "The internal auxiliary table has a malformed <propertyName>
property with raw value '<rawValue>'. The property must be a JSON array of
strings (e.g. '[\"id\",\"region\"]'). The auxiliary table metadata may be
corrupted or have been modified externally. Perform a full refresh of the
target table to recreate the auxiliary table."
]
},
"AUXILIARY_TABLE_PROPERTY_MISSING" : {
"message" : [
- "The auxiliary table <auxTableName> is missing the required
<propertyName> table property; cannot validate AutoCDC key columns. The
auxiliary table metadata may be corrupted or have been modified externally.
Perform a full refresh of the target table to recreate the auxiliary table."
+ "The internal auxiliary table is missing the required <propertyName>
table property; cannot validate AutoCDC key columns. The auxiliary table
metadata may be corrupted or have been modified externally. Perform a full
refresh of the target table to recreate the auxiliary table."
Review Comment:
ditto
##########
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 =
Review Comment:
ditto
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -104,12 +106,38 @@ object DatasetManager extends Logging {
transformer.transformTables { table =>
if (tablesToMaterialize.keySet.contains(table.identifier)) {
try {
- materializeTable(
+ val isFullRefresh =
tablesToMaterialize(table.identifier).isFullRefresh
+ val (tableWithMaterializationMetadata, catalogTableEntity) =
materializeTable(
resolvedDataflowGraph = resolvedDataflowGraph,
table = table,
- isFullRefresh =
tablesToMaterialize(table.identifier).isFullRefresh,
+ isFullRefresh = isFullRefresh,
context = context
)
+ // Auxiliary tables' lifecycle should follow the table that it
is complimentary to.
+ // If this table has any auxiliary tables, validate the target
can host them and
+ // materialize/full-refresh them accordingly.
+
resolvedDataflowGraph.auxiliaryTableSpecs.get(table.identifier).foreach {
+ auxiliaryTableSpec =>
+ // If this table is an AutoCDC target table, as identified
by being
+ // accompanied by an AutoCDC auxiliary table, additionally
validate that the
+ // target table supports row level mutations. This is a
relevant validation for
Review Comment:
shouldn't that be validated before the target table is materialized?
##########
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 }
Review Comment:
shouldn't this be an error rather than silently returning None?
##########
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(
Review Comment:
Is this the right way to format an error message. Shouldn't we have an error
class definition for it?
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -342,11 +351,127 @@ object DatasetManager extends Logging {
)
}
- table.copy(
- normalizedPath = Option(
-
catalog.loadTable(identifier).properties().get(TableCatalog.PROP_LOCATION)
+ val catalogTableEntity = catalog.loadTable(identifier)
+ val tableWithMaterializationMetadata =
+ table.copy(
+ normalizedPath =
+
Option(catalogTableEntity.properties().get(TableCatalog.PROP_LOCATION))
)
+
+ (tableWithMaterializationMetadata, catalogTableEntity)
+ }
+
+ /**
+ * Validate that the AutoCDC target table is backed by a connector
implementing
+ * [[SupportsRowLevelOperations]], the DSv2 contract for the
MERGE/UPDATE/DELETE-with-rewrite
+ * operations the AutoCDC transformation relies on. Reuses the target handle
already loaded by
+ * [[materializeTable]], so it performs no additional catalog I/O. Only
AutoCDC auxiliary specs
+ * carry a MERGE-backed target; other auxiliary tables have no such
requirement.
+ *
+ * @param targetTable the target table graph entity, source of
the identifier and
+ * declared format used in the error message.
+ * @param targetTableCatalogEntity the target table's loaded DSv2 handle.
+ */
+ private def requireAutoCdcTargetSupportsRowLevelOps(
+ targetTable: Table,
+ targetTableCatalogEntity: V2Table): Unit = {
+ if (!targetTableCatalogEntity.isInstanceOf[SupportsRowLevelOperations]) {
+ throw new AnalysisException(
+ errorClass = "AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE",
+ messageParameters = Map(
+ "tableName" -> targetTable.identifier.quotedString,
+ // Prefer the flow-declared format, falling back to the connector's
provider property.
+ "format" -> targetTable.format
+
.orElse(Option(targetTableCatalogEntity.properties.get(TableCatalog.PROP_PROVIDER)))
+ .getOrElse("<unknown>")
+ )
+ )
+ }
+ }
+
+ /**
+ * Materialize the auxiliary table according to the provided spec.
+ *
+ * @param auxiliaryTableSpec the spec describing the auxiliary table to
create/evolve.
+ * @param isFullRefresh whether the owning table is being fully refreshed.
+ * @param context the context for the pipeline update.
+ */
+ private def materializeAuxiliaryTable(
+ auxiliaryTableSpec: AuxiliaryTableSpec,
+ isFullRefresh: Boolean,
+ context: PipelineUpdateContext): Unit = {
+ // Get the DSv2 catalog handler and identifier for the aux table.
+ val (catalog, auxiliaryTableIdentifier) =
+ PipelinesCatalogUtils.resolveTableCatalog(context.spark,
auxiliaryTableSpec.identifier)
+
+ logInfo(
+ log"Materializing auxiliary table " +
+ log"${MDC(LogKeys.TABLE_NAME, auxiliaryTableSpec.identifier)}."
)
+
+ if (isFullRefresh) {
+ // Intentionally DROP and not TRUNCATE on full refresh. The auxiliary
table is an internal
Review Comment:
however, truncate would preserve history for table formats that support
this. Which means, the full refresh could be undone if necessary.
##########
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
Review Comment:
if this is the target table, why not name it targetTable, and targetSchema
below?
##########
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 {
Review Comment:
I wonder why this is an object and not a class. It feels as if the methods
here are taking in and passing down the same parameters over and over again.
##########
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 =
Review Comment:
I would add a comment that this first validates that the existing aux table
is consistent within itself, that is, all key columns exists.
##########
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]] = {
Review Comment:
and it looks as if the only place this is used is from
`parseRecordedKeyColumnNames`. Should it be defined as private and moved next
to its caller?
--
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]
