AnishMahto commented on code in PR #56122:
URL: https://github.com/apache/spark/pull/56122#discussion_r3306855873
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala:
##########
@@ -301,3 +312,176 @@ class SinkWrite(
.start()
}
}
+
+object AutoCdcAuxiliaryTable {
+ /**
+ * Helper for deriving the auxiliary AutoCDC catalog table identifier from a
target table. The
+ * derived name is anchored on [[AutoCdcReservedNames.prefix]] so it is
unambiguously
+ * AutoCDC-managed and cannot collide with a user-managed table.
+ */
+ def identifier(destination: TableIdentifier): TableIdentifier =
TableIdentifier(
+ table = s"${AutoCdcReservedNames.prefix}aux_state_${destination.table}",
+ database = destination.database,
+ catalog = destination.catalog
+ )
+}
+
+/**
+ * Helper mixin for AutoCDC merge-based write flows.
+ */
+trait AutoCdcMergeWriteMixin {
+ /** The spark session the AutoCDC flow is going to be planned in. */
+ protected def spark: SparkSession
+
+ /** The destination (target) table entity the AutoCDC flow will be writing
to. */
+ protected def destination: Table
+
+ /** The AutoCDC flow's [[ChangeArgs]] (keys, sequencing, columnSelection,
...). */
+ protected def changeArgs: ChangeArgs
+
+ /** Full schema of the auxiliary table for this SCD type. */
+ protected def auxiliaryTableSchema: StructType
+
+ // Eagerly validate at construction time that the destination supports
row-level ops.
+ requireDestinationSupportsRowLevelOps()
+
+ /**
+ * Idempotently create the auxiliary table for [[destination]] if it does
not already exist
+ * and return its [[TableIdentifier]].
+ */
+ protected def createAuxiliaryTableIfNotExists(spark: SparkSession):
TableIdentifier = {
Review Comment:
(2) is definitely the intended behavior - any key changes must require a
full refresh, it is semantically invalid to change the keys mid-way through and
SCD decomposition.
I'm choosing not to do the `tableExists` check though because:
1. This is a very small cost incurred once when the streaming query
executes; anyway processing the microbatch will take many orders of magnitude
more wall clock time
2. Very small edge case, but we'd still need to do `CREATE TABLE IF NOT
EXISTS` in the DDL, because there could otherwise be a TOCTOU race condition
between checking and creating table
3. `catalog.tableExists` is still going to incur some cost
----------
Btw something else that I'm tangentially thinking about is for SCD2, the aux
table will additionally contain data columns (it must, for to support
track-history columns), which are subject to schema evolution.
Or even for SCD1, when we eventually support `ignoreNull`, it should be
possible for users to upgrade their SCD1 flows to start using `ignoreNull`,
which would require evolving the CDC metadata column.
What this means is eventually we will have to add schema evolution for the
aux table, even if the table is not being created for the first time.
But we don't need to do that now, lets do it when either SCD2 or
`ignoreNull` lands. And in all cases, its still not valid for key columns to
change without full refresh.
--
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]
