szehon-ho commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r4225729744
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcReservedNames.scala:
##########
@@ -40,4 +44,98 @@ 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
+ ef.copy(name = df.name)
Review Comment:
Could we extend the casing preservation and early-rejection policy to nested
metadata fields too? `ef.copy(name = df.name)` preserves the outer column name,
but replaces its entire struct type, including the declared nested names.
For example, before this PR, a case-insensitive pipeline can create a target
using the following definition (assuming a streamable `source(id INT, version
BIGINT)` and a catalog that supports AUTO CDC's MERGE):
```sql
SET spark.sql.caseSensitive = false;
CREATE STREAMING TABLE target (
id INT,
version BIGINT,
__spark_autocdc_metadata STRUCT<
DELETESEQUENCE: BIGINT,
UPSERTSEQUENCE: BIGINT
>
);
CREATE FLOW writer AS AUTO CDC INTO target
FROM STREAM(source)
KEYS (id)
SEQUENCE BY version
STORED AS SCD TYPE 1;
```
The target retains `DELETESEQUENCE` / `UPSERTSEQUENCE`, and the
case-insensitive AUTO CDC writer can resolve them.
After upgrading to this PR, keep that definition unchanged and add a
downstream MV in the same graph:
```sql
SET spark.sql.caseSensitive = true;
CREATE MATERIALIZED VIEW copy AS
SELECT * FROM target;
```
On an incremental run:
1. `VirtualTableInput` uses this helper and exposes `deleteSequence` /
`upsertSequence`, so `copy` is planned and created with those nested names.
2. `evolveTable` merges the existing target schema first, preserving
`DELETESEQUENCE` / `UPSERTSEQUENCE` in the actual target.
3. `reservedFieldCasingDrifts` passes because the outer name,
`__spark_autocdc_metadata`, is identical; it does not inspect the struct's
fields.
4. At execution, `copy` reanalyzes its read against the actual target. Its
case-sensitive by-name write cannot match the uppercase nested names to the
canonical names in its destination schema.
This can therefore affect users who retain their original metadata
declaration, as well as users who omit it. I traced this through the code; I
have not run this variant locally.
For an explicit declaration, could the helper preserve compatible nested
names and ordering recursively, while validating types with the AUTO CDC flow's
effective resolver and applying the engine's nullability? For an omitted
declaration, could the existing drift check also reject nested casing
differences against the persisted schema? That follows the same policy already
implemented for the outer column. A regression test could use the target above
and a case-sensitive downstream `SELECT *`, covering both retained and omitted
metadata declarations.
--
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]
