szehon-ho commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3693574074
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -333,6 +333,26 @@ object DatasetManager extends Logging {
// Create the table if absent, otherwise evolve it (schema + properties).
existingTableOpt match {
case Some(existingTable) =>
+ // For an incrementally-updated AutoCDC target, validate that the
AutoCDC configuration has
+ // not drifted from what the auxiliary table recorded, BEFORE evolving
the target's schema.
+ // This ordering is load-bearing: `evolveTable` below ALTERs the
target (additively) in
+ // place, so a check that ran afterwards would leave the target
already mutated by a run it
+ // then rejects -- and the drift remedy ("correct the flow") could not
undo that. Running
+ // first means a rejected run leaves the target untouched. It also
surfaces a
+ // sequencing-type change as an actionable SEQUENCING_TYPE_DRIFT
rather than a generic
+ // CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE from the schema merge.
+ if (isTableIncrementallyUpdated) {
+
resolvedDataflowGraph.auxiliaryTableSpecs.get(table.identifier).collect {
+ case autoCdcSpec: AutoCdcAuxiliaryTableSpec => autoCdcSpec
+ }.foreach { autoCdcSpec =>
+ validateNoAutoCdcConfigDrift(
Review Comment:
Moving all four checks in here also narrowed *when* they run, and I think
that drops a guard we had before.
This block sits inside `case Some(existingTable) =>` (line 335), so every
check now requires the **target** table to already exist. But the three
aux-owned checks (key columns, SCD type, track history) used to run in
`materializeAuxiliaryTable` whenever the *auxiliary* table existed, regardless
of the target.
So: a user drops and recreates the target to reset it, not knowing to also
drop the internal `__aux_state_<target>` table. On the next run:
- `existingTableOpt` is `None`, so we take the `case None => createTable`
branch and no drift check runs at all.
- `materializeAuxiliaryTable` then finds the surviving aux table and goes
straight to `evolveTable(mergeWithExistingSchema = true)`, which merges the new
schema in and overwrites `keyColumnNamesProperty`, `scdTypePropertyKey`, and
`trackHistoryColumnNamesProperty` with this run's values.
That is the case `validateNoKeyColumnDrift`'s own scaladoc says it exists to
prevent: "a changed key set would otherwise be silently unioned into the schema
by the additive evolve."
I traced the downstream impact, and to be fair it isn't target corruption.
With changed keys the stale aux rows have NULL in the new key column so they
never join - they just sit there accumulating. With a changed SCD type or
track-history set, the merged aux schema stops lining up with the reconciled
rows and the run fails later in `unionByName` or on the aux MERGE with an
opaque schema error. Either way we lose what the guard was for: one clear error
telling the user to full refresh.
Suggested fix: keep the sequencing-type check inside the
`Some(existingTable)` branch, since it needs the target schema, and hoist the
other three out of it so they run whenever the aux table exists, still gated on
`isTableIncrementallyUpdated`. Worth a test in the durability suite mirroring
`AutoCdcScd1AuxiliaryTableDurabilitySuite:209` ("if the AutoCDC auxiliary table
is dropped between runs, it is transparently recreated") - that's the same
situation with the two tables swapped, and it's the half that now has neither a
guard nor a test.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -474,6 +481,61 @@ object DatasetManager extends Logging {
}
}
+ /**
+ * Validate that an incrementally-updated AutoCDC target's configuration has
not drifted from
+ * what its auxiliary table recorded. Called from [[materializeTable]]
BEFORE the target's
+ * `evolveTable`, so a rejected run leaves the target's schema untouched (a
check that ran after
+ * the target evolve would leave it already ALTERed by a run it then
rejects, and the drift
+ * remedy could not undo that).
+ *
+ * The sequencing-type check reads the existing target schema directly; the
key-column, SCD-type,
+ * and track-history checks read the recorded properties off the existing
auxiliary table, so this
+ * loads it. If the auxiliary table does not exist yet (first AutoCDC run
over a pre-existing
+ * target), those three checks are skipped -- there is no recorded
configuration to drift from.
+ *
+ * @param existingTargetSchema the schema of the already-materialized target
table.
+ * @param autoCdcSpec the auxiliary-table spec carrying this run's expected
AutoCDC configuration.
+ * @param context the context for the pipeline update.
+ */
+ private def validateNoAutoCdcConfigDrift(
+ existingTargetSchema: StructType,
+ autoCdcSpec: AutoCdcAuxiliaryTableSpec,
+ context: PipelineUpdateContext): Unit = {
+ val resolver = context.spark.sessionState.conf.resolver
+
+ // Independent of the auxiliary table: the sequencing type is embedded in
the target's schema.
+ AutoCdcAuxiliaryTable.validateNoTargetSequencingTypeDrift(
+ existingTargetSchema = existingTargetSchema,
+ targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
+ expectedScdType = autoCdcSpec.expectedScdType,
+ expectedSequencingType = autoCdcSpec.expectedSequencingType,
+ resolver = resolver
+ )
+
+ // The remaining checks read the recorded configuration off the existing
auxiliary table.
+ val (auxCatalog, auxIdentifier) =
+ PipelinesCatalogUtils.resolveTableCatalog(context.spark,
autoCdcSpec.identifier)
+ loadTableIfExists(auxCatalog, auxIdentifier).foreach {
existingAuxiliaryTable =>
Review Comment:
Minor, and fine as a follow-up rather than something to block on: this loads
the auxiliary table, and then `materializeAuxiliaryTable` resolves and loads it
again at line 457. `loadTableIfExists` is itself two catalog calls
(`tableExists` then `loadTable`), so that's four round trips per AutoCDC table
per run where there used to be two. Fine for an in-JVM metastore, less so for a
remote one.
Two independent ways to improve it whenever you get to it:
- Make `loadTableIfExists` one call instead of two.
`CatalogV2Util.loadTable` already does exactly that (`try` / `catch
NoSuchTableException`) and returns `Option[Table]`, which is what `V2Table`
aliases here. That helps all three call sites including the target load at line
306, and closes a small hole where the table disappearing between `tableExists`
and `loadTable` throws an uncaught `NoSuchTableException`. It also swallows
`NoSuchDatabaseException`, so if you'd rather not widen that for the target, a
narrow local `try` / `catch` does the job.
- Or load the aux table once in the `transformTables` block and pass the
handle into both `materializeTable` and `materializeAuxiliaryTable`. Nothing
between them mutates the aux table, so a single snapshot is safe - and it
happens to make the aux-owned checks independent of the target's existence,
which is the other comment.
--
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]
