anew commented on code in PR #57625:
URL: https://github.com/apache/spark/pull/57625#discussion_r3688709867
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala:
##########
@@ -227,13 +238,28 @@ object AutoCdcAuxiliaryTable {
StructField(Scd2BatchProcessor.deletedByBatchIdColName, LongType,
nullable = true)
val scd2AuxiliaryTableSchema = StructType(targetTableSchema.fields :+
deletedByBatchIdField)
+ // Resolve the effective track-history columns (an explicit TRACK HISTORY
selection, or the
+ // default of every eligible non-key/non-framework column) against the
target schema, using the
+ // same single source of truth the reconciler uses. A change in this set
reinterprets which
+ // transitions open a new historical record, so it is drift-checked.
+ val caseSensitive =
+ inputAutoCdcFlow.df.sparkSession.sessionState.conf.caseSensitiveAnalysis
Review Comment:
Fixed both. validateNoTargetSequencingTypeDrift now takes expectedScdType
and looks the sequencing-typed inner field up by name —
Scd1BatchProcessor.cdcUpsertSequenceFieldName for SCD1,
Scd2BatchProcessor.recordStartAtFieldName for SCD2 — instead of fields.head, so
a metadata field added at position 0 later can't silently compare an unrelated
type. And the _cdc_metadata column and the inner field are both matched through
the session Resolver now, consistent with findFieldInTargetSchema, so a
case-differing hand-written target DDL no longer skips the check.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala:
##########
@@ -373,6 +401,100 @@ object AutoCdcAuxiliaryTable {
}
}
+ /**
+ * Reject an incremental update to an existing AutoCDC target table whose
sequencing type has
+ * drifted. The AutoCDC sequencing *expression* may legitimately change
across runs (e.g. a new
+ * timestamp parse format), but its resolved result type must not: the
target persists the
+ * sequencing type inside its `_cdc_metadata` struct (and, for SCD2, in the
interval columns), so
+ * a changed type would make new events incomparable with the persisted
history and would
+ * otherwise surface only as a generic CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE
during schema
+ * evolution. Runs against the target table (before its schema is evolved),
not the auxiliary
+ * table. The remedy is a full refresh.
+ *
+ * @param existingTargetSchema the schema of the already-materialized target
table.
+ * @param expectedSequencingType the resolved sequencing type of the
incoming AutoCDC flow.
+ */
+ private[graph] def validateNoTargetSequencingTypeDrift(
+ existingTargetSchema: StructType,
+ targetTableIdentifier: TableIdentifier,
+ expectedSequencingType: DataType): Unit = {
+ // The sequencing type is embedded as the inner field type(s) of the
reserved _cdc_metadata
+ // struct, for both SCD1 (delete/upsert sequence fields) and SCD2
(recordStartAt field). Read it
+ // from the first inner field. If the metadata column is absent or not a
struct, this is not a
+ // recognizable AutoCDC target state; skip rather than misreport (schema
evolution will surface
+ // any genuine incompatibility).
+ val recordedSequencingType: Option[DataType] = existingTargetSchema.fields
+ .find(_.name == AutoCdcReservedNames.cdcMetadataColName)
+ .map(_.dataType)
+ .collect { case s: StructType if s.nonEmpty => s.fields.head.dataType }
Review Comment:
Fixed both. validateNoTargetSequencingTypeDrift now takes expectedScdType
and looks the sequencing-typed inner field up by name —
Scd1BatchProcessor.cdcUpsertSequenceFieldName for SCD1,
Scd2BatchProcessor.recordStartAtFieldName for SCD2 — instead of fields.head, so
a metadata field added at position 0 later can't silently compare an unrelated
type. And the _cdc_metadata column and the inner field are both matched through
the session Resolver now, consistent with findFieldInTargetSchema, so a
case-differing hand-written target DDL no longer skips the check.
--
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]
