szehon-ho commented on code in PR #57488:
URL: https://github.com/apache/spark/pull/57488#discussion_r3648742822
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala:
##########
@@ -375,6 +377,52 @@ class AutoCdcMergeFlow(
}
}
+ /**
+ * Validate that the resolved source dataframe does not contain any column
whose name collides
+ * (by exact name, resolver-aware) with an SCD-type-specific reserved
framework column that is
+ * NOT covered by [[requireReservedPrefixAbsentInSourceColumns]].
+ *
+ * The prefix guard above only rejects names starting with
[[AutoCdcReservedNames.prefix]].
+ * SCD2 additionally persists the framework columns
[[Scd2BatchProcessor.startAtColName]] and
+ * [[Scd2BatchProcessor.endAtColName]], which do NOT carry that prefix, so a
colliding source
+ * column would otherwise be silently overwritten during preprocessing
(SPARK-57251). SCD1
Review Comment:
Nit; can we get rid of the SPARK JIRA in these comments (they show up quite
a few times and dont bring up much value.) its quite easy now to see which pr
did it
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -1089,11 +1089,13 @@ object Scd2BatchProcessor {
* eventually persisted in the target table. If the user's source dataframe
contains any of
* these columns, SCD2 reconciliation will fail.
*
- * TODO(SPARK-57251): validate at [[AutoCdcMergeFlow]] construction time
that the source
- * schema and column selection do not collide with these reserved names,
so we fail fast
- * with a user-actionable error instead of silently overwriting them at
preprocess time.
+ * Note [[startAtColName]] and [[endAtColName]] do NOT carry the reserved
+ * [[AutoCdcReservedNames.prefix]], so a source-column collision with them
is not caught by the
+ * prefix-based guard;
[[org.apache.spark.sql.pipelines.graph.AutoCdcMergeFlow]] validates the
+ * source schema against this set at construction time (SPARK-57251) to fail
fast with a
Review Comment:
Nit: This says the flow validates against this entire set, but the new check
filters it to the non-prefixed names; the prefixed member is handled separately
by `requireReservedPrefixAbsentInSourceColumns`. Could we say "validates the
source schema against the non-prefixed names in this set" for precision?
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala:
##########
@@ -375,6 +377,52 @@ class AutoCdcMergeFlow(
}
}
+ /**
+ * Validate that the resolved source dataframe does not contain any column
whose name collides
+ * (by exact name, resolver-aware) with an SCD-type-specific reserved
framework column that is
+ * NOT covered by [[requireReservedPrefixAbsentInSourceColumns]].
+ *
+ * The prefix guard above only rejects names starting with
[[AutoCdcReservedNames.prefix]].
+ * SCD2 additionally persists the framework columns
[[Scd2BatchProcessor.startAtColName]] and
+ * [[Scd2BatchProcessor.endAtColName]], which do NOT carry that prefix, so a
colliding source
+ * column would otherwise be silently overwritten during preprocessing
(SPARK-57251). SCD1
+ * targets carry no such non-prefixed framework columns, so this guard is a
no-op for SCD1.
+ *
+ * Runs in the constructor before [[schema]] is forced, so it surfaces this
actionable error
+ * ahead of the (temporary) [[AUTOCDC_SCD2_NOT_SUPPORTED]] gate, and remains
correct once SCD2
+ * support lands.
+ */
+ private def requireReservedFrameworkColumnsAbsentInSourceColumns(): Unit = {
+ val resolver = spark.sessionState.conf.resolver
+ val reservedPrefix = AutoCdcReservedNames.prefix
+
+ // Only the non-prefixed reserved names need checking here; prefixed ones
are already rejected
+ // by [[requireReservedPrefixAbsentInSourceColumns]].
+ val reservedNames: Set[String] = changeArgs.storedAsScdType match {
+ case ScdType.Type2 =>
+
Scd2BatchProcessor.reservedFrameworkColNames.filterNot(_.startsWith(reservedPrefix))
+ case ScdType.Type1 =>
+ Set.empty
+ }
+
Review Comment:
Optional: `find` reports only the first collision, so if the source contains
both `__START_AT` and `__END_AT`, the error identifies only one as present.
Would it be useful to collect all conflicts and report them together? The
current message does list all reserved names, and this matches the prefix guard
above, so keeping this as-is also seems reasonable.
--
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]
