szehon-ho commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r4330516972
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcReservedNames.scala:
##########
@@ -40,4 +44,165 @@ private[pipelines] object AutoCdcReservedNames {
* Note that the schema of the CDC metadata column however can and does
differ on the SCD-type.
*/
val cdcMetadataColName: String = s"${prefix}metadata"
+
+ // A field is engine-owned when its name carries the reserved prefix.
Matching goes through the
+ // caller's `resolver` so a user-declared column is still recognized as
reserved under
+ // case-insensitive analysis, and the length guard keeps `substring` safe
for shorter names.
+ private def isReservedFieldName(name: String, resolver: Resolver): Boolean =
+ name.length >= prefix.length && resolver(name.substring(0, prefix.length),
prefix)
+
+ /** The engine-owned reserved AUTO CDC column(s) present in `schema`, if
any. */
+ private[pipelines] def reservedFields(schema: StructType, resolver:
Resolver): Seq[StructField] =
+ schema.fields.toSeq.filter(f => isReservedFieldName(f.name, resolver))
+
+ /** `schema` with the engine-owned reserved AUTO CDC column(s) removed. */
+ private[pipelines] def stripReservedFields(schema: StructType, resolver:
Resolver): StructType =
+ StructType(schema.fields.filterNot(f => isReservedFieldName(f.name,
resolver)))
+
+ /**
+ * The schema a materialized AUTO CDC target carries when the user declares
a schema: the user's
+ * columns with the engine-owned type/nullability substituted into any
declared reserved column,
+ * followed by the engine-owned reserved column(s) the declaration omitted.
`flowInferredSchema`
+ * must be flow-derived (no user-declared columns merged in), so its
reserved fields are exactly
+ * the engine-produced ones and never a user column that happens to carry
the prefix.
+ *
+ * A declared reserved field keeps its own spelling and position; only its
type and nullability
+ * are replaced with the engine's. That matters on the upgrade path: an
already-materialized
+ * target may carry the reserved column under a different casing (e.g. an
older definition
+ * declared it upper-cased under case-insensitive analysis), `evolveTable`
merges that existing
+ * spelling in, and this schema -- which also feeds the analysis-time read
path -- must agree with
+ * it so a downstream `SELECT *` plans against the column the target
actually has.
+ *
+ * The engine-owned reserved fields are made nullable so a target created
from a declared schema
+ * matches one created from an omitted schema (that path materializes the
inferred schema
+ * `asNullable`), avoiding a spurious nullability diff when a declaration is
added or removed.
+ */
+ private[pipelines] def appendEngineOwnedReservedFields(
+ declaredSchema: StructType,
+ flowInferredSchema: StructType,
+ resolver: Resolver): StructType = {
+ val engineReserved = StructType(reservedFields(flowInferredSchema,
resolver)).asNullable.fields
+
+ // A declaration may carry at most one field matching each engine-owned
reserved column; more
+ // than one (e.g. the metadata column declared twice under
case-insensitive analysis) is
+ // ambiguous and schema validation should have rejected it upstream.
+ engineReserved.foreach { ef =>
+ val matches = declaredSchema.fields.filter(df => resolver(df.name,
ef.name))
+ if (matches.length > 1) {
+ throw SparkException.internalError(
+ s"Ambiguous reserved AUTO CDC field:
${matches.map(_.name).mkString(", ")} all match " +
+ s"the engine-owned column ${ef.name}.")
+ }
+ }
+
+ // Rebuild the declared schema in place: a declared reserved field keeps
its spelling and
+ // position but takes the engine-owned type/nullability. A
reserved-prefixed declared field with
+ // no engine counterpart is left as-is (validation rejects it upstream for
an AUTO CDC target).
+ val usedEngineNames = scala.collection.mutable.Set.empty[String]
+ val rebuilt = declaredSchema.fields.map { df =>
+ if (isReservedFieldName(df.name, resolver)) {
+ engineReserved.find(ef => resolver(df.name, ef.name)) match {
+ case Some(ef) =>
+ usedEngineNames += ef.name
+ reconcileReservedField(df, ef, resolver)
+ case None => df
+ }
+ } else {
+ df
+ }
+ }
+
+ // Append engine-owned reserved fields the declaration omitted, in
canonical form.
+ val appended = engineReserved.filterNot(ef =>
usedEngineNames.contains(ef.name))
+ StructType(rebuilt ++ appended)
+ }
+
+ /**
+ * Reconcile a declared reserved field against its engine-owned counterpart:
keep the declared
+ * spelling, take the engine field's type and nullability. When both are
structs -- the AUTO CDC
+ * metadata column is one -- recurse so the declared nested spelling and
ordering are kept too,
+ * with the engine's nested types and nullability, mirroring the
outer-column policy one level
+ * down.
+ */
+ private def reconcileReservedField(
+ declared: StructField,
+ engine: StructField,
+ resolver: Resolver): StructField =
+ (declared.dataType, engine.dataType) match {
+ case (declaredStruct: StructType, engineStruct: StructType) =>
+ engine.copy(
+ name = declared.name,
+ dataType = reconcileReservedStruct(declaredStruct, engineStruct,
resolver))
+ case _ =>
+ engine.copy(name = declared.name)
Review Comment:
Could we also recurse through array element types in this helper and in
`reservedFieldCasingDrifts`? The current fallback replaces the declared array
element schema with the engine's, losing any declared nested spelling.
AUTO CDC accepts `ARRAY<STRUCT<ordinal: BIGINT>>` as a sequencing type:
`ScdBatchValidator` calls `RowOrdering.isOrderable`, which accepts arrays of
orderable structs. SCD1 then uses `max_by(row, sequencing)` to select the
greatest sequence per key. For example, `[{ordinal: 2}]` sorts after
`[{ordinal: 1}]`.
Here is an end-to-end example. Assume a configured Iceberg catalog named
`lakehouse`, namespace `demo`, and a connector compatible with this Spark
version. First create an append-only source using ordinary Spark SQL:
```sql
CREATE TABLE lakehouse.demo.events (
id INT,
name STRING,
version ARRAY<STRUCT<ordinal: BIGINT>>
) USING iceberg;
INSERT INTO lakehouse.demo.events
SELECT 1, 'alice', array(named_struct('ordinal', CAST(1 AS BIGINT)))
UNION ALL
SELECT 1, 'alicia', array(named_struct('ordinal', CAST(2 AS BIGINT)));
```
Then use this pipeline SQL source:
```sql
SET spark.sql.caseSensitive = false;
CREATE STREAMING TABLE lakehouse.demo.target (
id INT,
name STRING,
version ARRAY<STRUCT<ordinal: BIGINT>>,
__spark_autocdc_metadata STRUCT<
deleteSequence: ARRAY<STRUCT<ORDINAL: BIGINT>>,
upsertSequence: ARRAY<STRUCT<ORDINAL: BIGINT>>
>
) USING iceberg;
CREATE FLOW writer AS AUTO CDC INTO lakehouse.demo.target
FROM STREAM(lakehouse.demo.events)
KEYS (id)
SEQUENCE BY version
STORED AS SCD TYPE 1;
SET spark.sql.caseSensitive = true;
CREATE MATERIALIZED VIEW lakehouse.demo.latest_version
USING iceberg
AS SELECT
id,
__spark_autocdc_metadata.upsertSequence[0].ORDINAL AS latest_ordinal
FROM lakehouse.demo.target;
```
Before this PR, `VirtualTableInput` uses the declared metadata schema, so
the case-sensitive `ORDINAL` reference resolves. The case-insensitive AUTO CDC
flow can write the source's lowercase `ordinal` into the declared uppercase
metadata field; the expected downstream result is `(1, 2)`.
With this PR, reconciliation stops at the array and substitutes the engine's
`ARRAY<STRUCT<ordinal: BIGINT>>`. The downstream reference to `ORDINAL`
therefore fails during graph analysis, even though the declaration and query
are unchanged. This variant does not require an existing target or an
incremental upgrade.
Could we preserve compatible declared names recursively through arrays,
apply the engine's nullability, and extend the drift check through arrays too?
A regression test using the case-sensitive downstream field reference above
would cover the retained-declaration case.
I traced this through the code; I have not executed this example locally.
--
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]
