szehon-ho commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3692319590
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -453,6 +470,12 @@ object DatasetManager extends Logging {
targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
expectedScdType = autoCdcSpec.expectedScdType
)
+ AutoCdcAuxiliaryTable.validateNoTrackHistoryDrift(
Review Comment:
**By the time this fires, the target has already been altered - and the
error's own remedy then wedges the pipeline.**
`materializeTable` runs the target's `evolveTable` before
`materializeAuxiliaryTable` runs these validators (`DatasetManager.scala:110`,
then `:134`). So when an SCD2 flow adds a source column, the target is ALTERed
to add `city` and *then* this throws `TRACK_HISTORY_DRIFT`. `diffSchemas` only
adds columns, so the target keeps `city` for good; the aux table never gets it,
since its `evolveTable` is below this throw.
The message says "Correct the conflicting flow(s) or perform a full
refresh." Full refresh works. Correcting the flow does not: reverting the
source makes this check pass (recorded `[name, seq]` == expected `[name,
seq]`), but `Scd2ForeachBatchHandler.reconcileMicrobatch` unions the batch, the
aux rows, and the target rows with `unionByName`, and
`findAffectedRowsFromTargetTable` passes all target columns through. Six
columns against seven, so the run dies on a column-count mismatch that mentions
neither AutoCDC nor drift. The error steers users onto the one remedy that
doesn't work.
Suggested fix, mirroring what you already do for sequencing type: run this
check before the target's `evolveTable`, so a rejected run leaves the target
untouched. `validateNoKeyColumnDrift` has the same ordering, so consider moving
the group. Worth a test too - neither new end-to-end test asserts the target's
schema after the rejected run.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala:
##########
@@ -227,13 +239,27 @@ object AutoCdcAuxiliaryTable {
StructField(Scd2BatchProcessor.deletedByBatchIdColName, LongType,
nullable = true)
val scd2AuxiliaryTableSchema = StructType(targetTableSchema.fields :+
deletedByBatchIdField)
+ // The effective track-history column set, resolved by the flow from its
user-selected source
+ // schema (see [[AutoCdcMergeFlow.trackHistoryColumnNames]]) -- NOT
recomputed here from the
+ // evolved target schema, which would keep tracking columns the flow no
longer selects and would
+ // miss implicit (default / `* EXCEPT`) tracked-set changes. A change in
this set reinterprets
+ // which transitions open a new historical record, so it is drift-checked.
+ val trackHistoryColumnNames =
inputAutoCdcFlow.trackHistoryColumnNames.getOrElse(
+ throw SparkException.internalError(
+ "SCD2 AutoCDC flow is missing its resolved track-history column set."
+ )
+ )
+
val scd2AuxiliaryTableProperties =
// Record which SCD strategy this auxiliary table serves so downstream
readers can identify it
// without inspecting the schema.
Map(scdTypePropertyKey -> ScdType.Type2.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))) ++
+ // Persist the resolved track-history column names; a change
reinterprets already-reconciled
+ // history, so it is immutable post-creation (full-refresh is the only
way to change it).
+ Map(trackHistoryColumnNamesProperty ->
serializeKeyColumnNames(trackHistoryColumnNames)) ++
Review Comment:
`serializeKeyColumnNames` / `parseKeyColumnNames` now serve two properties,
but their scaladoc still describes only keys: "callers are expected to enforce
a non-empty key set upstream" (`AutoCdcAuxiliaryTable.scala:74`).
That clause is wrong for this caller - an empty tracked set is legitimate
when every eligible source column is a key, and `computeTrackedHistoryColumns`
returns `Seq.empty` there. Someone trusting the comment could add a non-empty
assertion and break it.
They're really just a JSON string-array codec, so `serializeColumnNames` /
`parseColumnNames` with a property-neutral comment would fit.
`AUXILIARY_TABLE_PROPERTY_MALFORMED` is already generic, so nothing else
changes.
--
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]
