spark-test commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r4356765634
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -317,9 +318,36 @@ object DatasetManager extends Logging {
val (catalog, identifier) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
- val outputSchema = table.specifiedSchema.getOrElse(
- inferredSchemas(table.identifier).asNullable
- )
+ // A table is an AUTO CDC target iff it has an AUTO CDC auxiliary table
spec. Compute the
+ // effective (flow-aware) resolver once here and reuse it for both the
output schema below and
+ // the drift checks further down; each `effectiveCaseSensitivityFor` call
walks the flows and
+ // can throw on a conflicting per-flow conf, so computing it a single time
also avoids doing
+ // that work (and that potential throw) twice.
+ val autoCdcAuxTableSpecOpt = auxiliaryTableSpecOpt.collect {
+ case autoCdcSpec: AutoCdcAuxiliaryTableSpec => autoCdcSpec
+ }
+ val effectiveCaseSensitive =
+ effectiveCaseSensitivityFor(resolvedDataflowGraph, table.identifier,
context)
+ val effectiveResolver =
SchemaInferenceUtils.resolverFor(effectiveCaseSensitive)
+
+ val outputSchema = table.specifiedSchema match {
+ case Some(ss) if autoCdcAuxTableSpecOpt.isDefined =>
+ // The user schema describes the logical AUTO CDC target; the engine
owns the reserved
+ // metadata column(s). Drop whatever reserved column(s) the user
declared and append the
Review Comment:
**Nit (P3):** This comment and the matching one in `VirtualTableInput.load`
(elements.scala:187-188) say the declared reserved column is dropped and the
engine-owned shape appended, so the table "always has exactly the reserved
column(s) the AUTO CDC MERGE writes". `appendEngineOwnedReservedFields` does
something different: it rebuilds the declaration in place. A declared reserved
column keeps its declared outer and nested spelling and position and takes only
the engine type and nullability, and a prefixed declared field with no engine
counterpart is kept. The upgrade-path tests and the casing-drift check depend
on that spelling preservation. A follow-up that trusted this comment and
canonicalized the name would reintroduce the mismatch this PR fixes.
Separately, AutoCdcReservedColumnMaterializationSuite.scala:143 calls the
engine column `nullable=false`, but the helper makes engine-owned reserved
fields nullable, which the suite's own nullability test asserts.
**Verification:**
- **Inspection:** The comments on the declared AUTO CDC branches in
materialization and in the analysis-time read path match the helper's in-place,
spelling-preserving reconciliation, and the test comment matches the nullable
engine column.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -294,6 +294,11 @@
"One or more AutoCDC flows writing to the target use key columns
<expectedKeySchema>, which are inconsistent with the keys recorded for it
(recorded <recordedKeySchema>). AutoCDC does not support changing key columns
or their types across incremental pipeline runs. Correct the conflicting
flow(s) or perform a full refresh of the target table."
]
},
+ "RESERVED_METADATA_COLUMN_CASING_DRIFT" : {
+ "message" : [
+ "The target table already has the reserved AutoCDC metadata column
as `<existingColumnName>`, but the current table definition resolves it to
`<resolvedColumnName>`. AutoCDC does not support changing the casing of the
reserved metadata column across incremental pipeline runs: the target keeps its
existing spelling while downstream consumers are planned against the new one,
so a case-sensitive read of the target could fail to match them. Declare the
reserved column with its existing name `<existingColumnName>`, or perform a
full refresh of the target table."
Review Comment:
**Nit (P3):** `reservedFieldCasingDrifts` also reports nested drift, as
dot-qualified paths: the nested rejection test expects `existingColumnName =
__spark_autocdc_metadata.DELETESEQUENCE`. With this template the message then
says the target "already has the reserved AutoCDC metadata column as
`__spark_autocdc_metadata.DELETESEQUENCE`" and asks the user to "Declare the
reserved column with its existing name
`__spark_autocdc_metadata.DELETESEQUENCE`". That reads as one top-level column
with a dot in its name. The actual remedy is to declare the metadata column
with its nested field spelled `DELETESEQUENCE`, or to perform a full refresh.
When the nested name comes from a struct-typed SEQUENCE BY expression, the user
never declared that field at all. Crossing an array element adds no path
segment, so array cases are even harder to map back to the declaration.
This extends an [existing
discussion](https://github.com/apache/spark/pull/57644#issuecomment-6087653115).
**Verification:**
- **Behavior:** The nested casing-drift rejection renders a message naming
the nested field path within the metadata column and a remedy that refers to
declaring the metadata column with that existing nested spelling or performing
a full refresh.
- **Compatibility:** The top-level casing-drift rejection still reports the
same error condition and identifies the existing and resolved column spellings.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -271,24 +265,40 @@ trait GraphValidations extends Logging {
// table's); for a named flow (e.g. `CREATE FLOW <name> AS AUTO CDC INTO
<target>`) they
// differ, and keying on the flow identifier would silently skip
validation.
flowsTo.keys.flatMap(table.get).foreach { t: TableElement =>
+ val flows = flowsTo(t.identifier).map(f => resolvedFlow(f.identifier))
// The output inferred schema of a table is the declared schema merged
with the
// schema of all incoming flows. This must be equivalent to the declared
schema.
val inferredSchema = SchemaInferenceUtils
.inferSchemaFromFlows(
tableIdentifier = t.identifier,
- flowsTo(t.identifier).map(f => resolvedFlow(f.identifier)),
+ flows,
userSpecifiedSchema = t.specifiedSchema,
sessionCaseSensitive = sessionCaseSensitive
)
+ // Match reserved columns with the flow's effective case sensitivity,
not just the session:
+ // a case-sensitivity conf set on the flow overrides the session, the
same way the rest of
+ // AUTO CDC derives its resolver.
+ val resolver = SchemaInferenceUtils.resolverFor(
+ SchemaInferenceUtils.effectiveCaseSensitivity(t.identifier, flows,
sessionCaseSensitive))
t.specifiedSchema.foreach { ss =>
- // Check the inferred schema matches the specified schema. Used to
catch errors where the
- // inferred user-facing schema has columns that are not in the
specified one.
- if (inferredSchema != ss) {
- val datasetType = GraphElementTypeUtils
- .getDatasetTypeForMaterializedViewOrStreamingTable(
- flowsTo(t.identifier).map(f => resolvedFlow(f.identifier))
- )
+ // A declared schema must match the schema inferred from the incoming
flows. Only for an
+ // AUTO CDC target are the engine-owned reserved metadata column(s)
set aside on both sides
+ // first: the user may omit them (the engine appends them at
materialization) or declare
+ // them, and comparing with the reserved columns removed accepts
either while still catching
+ // a genuine mismatch in the remaining columns, and stays correct if
more than one reserved
+ // column is ever added. For any other table the reserved prefix
carries no special meaning,
+ // so the comparison is exact -- otherwise a reserved-prefixed column
a non-CDC flow
+ // produces but the declaration omits would be wrongly accepted (then
dropped later).
+ val schemasMatch = if
(flows.exists(GraphElementTypeUtils.isAutoCdcFlow)) {
+ AutoCdcReservedNames.stripReservedFields(inferredSchema, resolver) ==
Review Comment:
**Non-blocking (P2):** Because every `__spark_autocdc_`-prefixed field is
stripped from both sides, validation can't reject the two declaration shapes
that `AutoCdcReservedNames.appendEngineOwnedReservedFields` says are rejected
upstream:
- A reserved-prefixed column with no engine counterpart, such as a typo
`__spark_autocdc_metdata STRUCT<...>` on an otherwise data-only declaration.
Before this PR it failed validation because the real metadata column was
missing. Now both sides reduce to the data columns. The helper keeps the
misspelled column (`case None => df`), so the created target and the
analysis-time read path get an extra reserved-namespace column the AUTO CDC
MERGE never writes. If that column is declared NOT NULL, the MERGE fails at
runtime instead.
- Both `__spark_autocdc_metadata` and `__SPARK_AUTOCDC_METADATA` under a
case-insensitive flow. `StructType.merge` folds both onto the single flow
field, the stripped comparison passes, and the user gets the helper's
`SparkException.internalError`. `VirtualTableInput.load` calls the same helper
while downstream flows are analyzed (`FlowAnalysis` calls `input.load` before
`DataflowGraph.validate` runs), so this can surface as INTERNAL_ERROR during
graph resolution.
Neither the SQL registration path
(`StructType(cst.columns.map(_.toV1Column))`) nor the Scala registration API
checks declared columns for the reserved prefix or for case-insensitive
duplicates. The only prefix guard,
`requireReservedPrefixAbsentInSourceColumns`, checks the change-feed source.
Both shapes therefore reach the helper unrejected, and its comments that
justify keeping or throwing on them are not true today.
This extends an [existing
discussion](https://github.com/apache/spark/pull/57644#issuecomment-5483130351).
**Recommended change:** Make AUTO CDC declared-schema validation set aside
only the engine-owned reserved columns the flow produces. Reject a
reserved-prefixed declared column that has no engine-owned counterpart, and a
declaration with more than one column matching the same engine-owned reserved
column, with a user-facing error. Have the shared reconciliation helper raise
that same user-facing error instead of an internal error when it meets such a
declaration on the analysis-time read path.
**Why this works:** Derive the engine-owned reserved names from the
flow-only inferred schema (the same source materialization uses) under the
flow-effective resolver. In validation, strip only declared fields that match
one of those names. Report AUTOCDC_RESERVED_COLUMN_NAME_PREFIX_CONFLICT, or an
equivalent user-facing condition, for any remaining reserved-prefixed declared
field and for any second declared field matching the same engine-owned column.
Put this check in one AutoCdcReservedNames helper and also call it from
appendEngineOwnedReservedFields, so VirtualTableInput.load, which runs during
graph resolution before validation, reports the same error. Update the helper
comments that currently claim upstream rejection.
**Scope:** Tighten the AUTO CDC declared-schema check in graph validation
and in the shared reserved-column helper so unknown or ambiguous reserved
declarations fail with a user-facing error on every route, and make the helper
comments match.
**Compatibility:** Declarations that omit the reserved column or declare it
exactly once keep their current materialized and analysis-time schema,
including the preserved declared spelling.
**Risks:** A declaration that also lists the real metadata column alongside
an extra `__spark_autocdc_*` column was accepted before this PR and would now
fail validation. Treat this as an intentional tightening and mention it in the
PR description.
**Constraints:** Reserved-name matching must use the flow-effective case
sensitivity. Under a case-sensitive flow, a differently cased user column such
as `__SPARK_AUTOCDC_METADATA` remains an ordinary user column and must not be
rejected.
**Exclusions:**
- Validating the nested shape of a declared reserved column beyond the
existing merge-compatibility check, such as rejecting extra nested fields
inside the metadata struct, stays out of scope.
**Success:** A data-only AUTO CDC declaration that adds a reserved-prefixed
column with no engine counterpart fails validation with a user-facing
reserved-prefix error. A declaration that matches the engine metadata column
more than once under the flow's case sensitivity fails with the same
user-facing error, both when a downstream dataset is analyzed and at
validation, and never with INTERNAL_ERROR. Data-only declarations, declarations
that include the reserved column exactly once, and case-sensitive flows with a
distinct differently cased user column behave as they do today.
**Verification:**
- **Behavior:** A data-only declaration on an AUTO CDC target that adds an
unmatched reserved-prefixed column is rejected at validation with the
user-facing reserved-prefix error.
- **Regression:** A declaration containing both case variants of the
metadata column under a case-insensitive flow, with a same-graph downstream
reader of the target, fails with the user-facing error rather than
INTERNAL_ERROR.
- **Compatibility:** The existing outcomes still hold: data-only and full
declarations are accepted, the wrong-type metadata declaration is rejected, the
case-sensitive distinct-column case is kept, and non-AUTO CDC tables use exact
comparison.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcReservedNames.scala:
##########
@@ -40,4 +44,192 @@ 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, and reconcile the
types recursively
+ * (see `reconcileReservedType`) so declared nested spelling survives
through structs and arrays.
+ */
+ private def reconcileReservedField(
+ declared: StructField,
+ engine: StructField,
+ resolver: Resolver): StructField =
+ engine.copy(
Review Comment:
**Nit (P3):** `engine.copy(name = declared.name, ...)` keeps the engine
field's metadata, which is empty because it comes from
`AutoCdcMergeFlow.schema`, and discards the declared field's. The same happens
for nested fields through `reconcileReservedStruct`. The helper doc above says
only the type and nullability are replaced, and `SchemaInferenceUtils` merges
the declared schema in specifically to carry user metadata such as comments.
Before this PR, a declaration like `__spark_autocdc_metadata STRUCT<...>
COMMENT 'cdc bookkeeping'` was materialized verbatim. Now the comment never
reaches the created table or `VirtualTableInput.load`, and on a full refresh of
an existing target `diffSchemas` removes the persisted comment. Building from
the declared field instead, with its name and metadata plus the engine's
reconciled type and nullability, would match the documented contract.
**Verification:**
- **Behavior:** A declared AUTO CDC target whose reserved metadata column
carries a COMMENT is materialized with that comment while still receiving the
engine's type and nullability.
- **Regression:** A full refresh of an existing AUTO CDC target with a
commented declared reserved column leaves the comment in place.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcReservedColumnMaterializationSuite.scala:
##########
@@ -0,0 +1,506 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.graph
+
+import java.util.Locale
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.pipelines.autocdc.{AutoCdcReservedNames,
Scd1BatchProcessor}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{ArrayType, IntegerType, LongType,
StringType, StructType}
+
+/**
+ * Materialization-level tests for AUTO CDC's engine-owned reserved metadata
column (SPARK-58118).
+ *
+ * The reserved `__spark_autocdc_metadata` column is engine-owned: a
user-declared schema may omit
+ * it, and materialization appends the engine-owned shape so the created table
matches what the
+ * AUTO CDC MERGE writes at runtime. Reserved-column matching goes through the
flow's effective
+ * case sensitivity -- a pipeline-level `SET spark.sql.caseSensitive` can
differ from the session --
+ * so these tests inspect the created table's schema rather than only
validation.
+ */
+class AutoCdcReservedColumnMaterializationSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ import testImplicits._
+
+ /** The engine's SCD1 metadata struct shape: `deleteSequence` /
`upsertSequence`, both long. */
+ private def scd1MetadataType: StructType = new StructType()
+ .add(Scd1BatchProcessor.cdcDeleteSequenceFieldName, LongType)
+ .add(Scd1BatchProcessor.cdcUpsertSequenceFieldName, LongType)
+
+ test("materialization appends the engine-owned reserved metadata column when
the user " +
+ "schema omits it") {
+ // The user declares only the logical data columns and omits the
engine-owned reserved
+ // metadata column. Materialization must append it so the created target
has exactly what the
+ // AUTO CDC MERGE writes; otherwise the MERGE fails with an unresolved
metadata column.
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("name", StringType)
+ .add("version", LongType, nullable = false)
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 5L))
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable(
+ "target",
+ catalog = Some(catalog),
+ database = Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")))
+ }
+ runPipeline(ctx)
+
+ val targetSchema = spark.table(s"$catalog.$namespace.target").schema
+ assert(
+
targetSchema.fieldNames.contains(AutoCdcReservedNames.cdcMetadataColName),
+ "target should carry the engine-owned reserved metadata column, got " +
+ targetSchema.fieldNames.mkString(", "))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 5L, cdcMeta(None, Some(5L))))
+ )
+ }
+
+ test("materialization matches the reserved metadata column through the
flow's effective case " +
+ "sensitivity, not the session's") {
+ // The session is case-sensitive, but the flow sets
spark.sql.caseSensitive=false, so the
+ // flow's effective analysis is case-insensitive -- the same resolver the
AUTO CDC MERGE uses.
+ // A user who declares the reserved column as __SPARK_AUTOCDC_METADATA
(upper case) is then
+ // declaring the engine-owned column, so materialization must recognize it
as reserved and end
+ // up with a single engine-owned column, not keep the upper-case one and
append a lower-case
+ // duplicate. Matching against the session resolver (case-sensitive) here
would leave two.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("version", LongType, nullable = false)
+ .add(AutoCdcReservedNames.cdcMetadataColName.toUpperCase(Locale.ROOT),
scd1MetadataType)
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 5L))
+ val ctx = new TestGraphRegistrationContext(
+ spark,
+ Map(SQLConf.CASE_SENSITIVE.key -> "false")) {
+ registerTable(
+ "target",
+ catalog = Some(catalog),
+ database = Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")))
+ }
+ runPipeline(ctx)
+
+ val reserved =
spark.table(s"$catalog.$namespace.target").schema.fieldNames
+
.filter(_.toLowerCase(Locale.ROOT).startsWith(AutoCdcReservedNames.prefix))
+ // Exactly one reserved column (no duplicate), which is the point of the
effective-resolver
+ // fix. Its spelling is the one the user declared -- the engine
substitutes only the type,
+ // keeping the declared casing and position.
+ assert(reserved.length == 1,
+ "expected exactly one engine-owned metadata column, got " +
reserved.mkString(", "))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, 5L, cdcMeta(None, Some(5L))))
+ )
+ }
+ }
+
+ test("materialization keeps a differently-cased user column distinct from
the engine's " +
+ "reserved column when the flow is case-sensitive") {
+ // Reverse of the previous test: the session is case-insensitive but the
flow sets
+ // spark.sql.caseSensitive=true. Under the flow's case-sensitive analysis
a user-declared
+ // __SPARK_AUTOCDC_METADATA is a genuinely distinct column from the
engine's lower-case
+ // __spark_autocdc_metadata, so both survive. The engine still owns and
writes only the
+ // lower-case column (nullable=false); the user's upper-case column is
left unpopulated.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val userColName =
AutoCdcReservedNames.cdcMetadataColName.toUpperCase(Locale.ROOT)
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("version", LongType, nullable = false)
+ .add(userColName, scd1MetadataType)
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 5L))
+ val ctx = new TestGraphRegistrationContext(
+ spark,
+ Map(SQLConf.CASE_SENSITIVE.key -> "true")) {
+ registerTable(
+ "target",
+ catalog = Some(catalog),
+ database = Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version")))
+ }
+ runPipeline(ctx)
+
+ val schema = spark.table(s"$catalog.$namespace.target").schema
+ // Both the user column and the engine column survive; case-sensitive
analysis keeps them
+ // distinct.
+ assert(schema.fieldNames.contains(userColName),
schema.fieldNames.mkString(", "))
+ assert(
+ schema.fieldNames.contains(AutoCdcReservedNames.cdcMetadataColName),
+ schema.fieldNames.mkString(", "))
+ // Read under case-sensitive analysis so the two case-differing columns
stay distinct. The
+ // engine writes its metadata to the lower-case column; the user's
upper-case column is left
+ // unpopulated (null).
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, 5L, null, cdcMeta(None, Some(5L))))
+ )
+ }
+ }
+ }
+
+ test("a same-graph downstream SELECT * sees the target's materialized
schema, including the " +
+ "engine-owned reserved column") {
+ // The target declares a data-only schema (omits the reserved metadata
column). A downstream
+ // dataset in the same graph reads it with SELECT *. Its plan-time view of
the target (through
+ // VirtualTableInput) must match the schema the target is actually
materialized with, otherwise
+ // the plan carries the 3-column declaration while execution reads the
4-column table.
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("name", StringType)
+ .add("version", LongType, nullable = false)
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 5L))
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(name = "writer", target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"), sequencing = functions.col("version")))
+ registerMaterializedView("enriched", catalog = Some(catalog), database =
Some(namespace),
+ query = readFlowFunc(s"$catalog.$namespace.target"))
+ }
+ runPipeline(ctx)
+
+ val targetFields =
spark.table(s"$catalog.$namespace.target").schema.fieldNames.toSeq
+ val enrichedFields =
spark.table(s"$catalog.$namespace.enriched").schema.fieldNames.toSeq
+ assert(
+ enrichedFields == targetFields,
+ s"downstream consumer schema ${enrichedFields.mkString(",")} should
match the materialized " +
+ s"target schema ${targetFields.mkString(",")}")
+ assert(enrichedFields.contains(AutoCdcReservedNames.cdcMetadataColName))
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.enriched"),
+ Seq(Row(1, "alice", 5L, cdcMeta(None, Some(5L))))
+ )
+ }
+
+ test("the engine-owned reserved column is nullable when the user declares a
data-only schema, " +
+ "matching the omitted-schema path") {
+ // A target created from a data-only declaration must get the same
nullable metadata column as
+ // one created with no declared schema (that path materializes the
inferred schema asNullable),
+ // so adding or removing a declaration between runs does not surface as a
nullability change.
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("version", LongType, nullable = false)
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 5L))
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(name = "auto_cdc_flow", target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "version")),
+ keys = Seq("id"), sequencing = functions.col("version")))
+ }
+ runPipeline(ctx)
+ val metaField = spark.table(s"$catalog.$namespace.target")
+ .schema(AutoCdcReservedNames.cdcMetadataColName)
+ assert(metaField.nullable, "engine-owned reserved metadata column should
be nullable")
+ }
+
+ test("a data-only declaration on an already-materialized target is stable
across an " +
+ "incremental re-run") {
+ // The second run reaches the target through
evolveTable/mergeWithExistingSchema rather than a
+ // fresh create. Because the declared-schema path produces the same
engine-owned shape the
+ // target already carries, the schema must not churn between runs.
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("version", LongType, nullable = false)
+ def buildCtx(rows: Seq[(Int, Long)]): TestGraphRegistrationContext = {
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData(rows: _*)
+ new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(name = "auto_cdc_flow", target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "version")),
+ keys = Seq("id"), sequencing = functions.col("version")))
+ }
+ }
+ runPipeline(buildCtx(Seq((1, 5L))))
+ val schemaAfterRun1 = spark.table(s"$catalog.$namespace.target").schema
+ runPipeline(buildCtx(Seq((2, 3L))))
+ val schemaAfterRun2 = spark.table(s"$catalog.$namespace.target").schema
+ assert(
+ schemaAfterRun2 == schemaAfterRun1,
+ s"schema changed across incremental runs: $schemaAfterRun1 ->
$schemaAfterRun2")
+ }
+
+ test("an already-materialized target's reserved-column casing is preserved
for a " +
+ "case-sensitive downstream SELECT *") {
+ // Upgrade path: a target already exists in the catalog with the reserved
metadata column
+ // declared upper-cased (allowed under case-insensitive AUTO CDC).
evolveTable merges that
+ // existing spelling in, so the read path must report the SAME casing,
otherwise a
+ // case-sensitive downstream `SELECT *` plans the canonical lower-case
name and cannot resolve
+ // the upper-case column the target actually has.
+ val upperMeta =
AutoCdcReservedNames.cdcMetadataColName.toUpperCase(Locale.ROOT)
+ val del = Scd1BatchProcessor.cdcDeleteSequenceFieldName
+ val ups = Scd1BatchProcessor.cdcUpsertSequenceFieldName
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, " +
+ s"$upperMeta STRUCT<$del:BIGINT,$ups:BIGINT> NOT NULL)")
+
+ val metadataType = new StructType().add(del, LongType).add(ups, LongType)
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("version", LongType, nullable = false)
+ .add(upperMeta, metadataType, nullable = false)
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 5L))
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(name = "auto_cdc_flow", target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "version")),
+ keys = Seq("id"), sequencing = functions.col("version")))
+ registerMaterializedView("copy", catalog = Some(catalog), database =
Some(namespace),
+ sqlConf = Map(SQLConf.CASE_SENSITIVE.key -> "true"),
+ query = readFlowFunc(s"$catalog.$namespace.target"))
+ }
+ runPipeline(ctx)
+
+ val targetFields =
spark.table(s"$catalog.$namespace.target").schema.fieldNames.toSeq
+ val copyFields =
spark.table(s"$catalog.$namespace.copy").schema.fieldNames.toSeq
+ assert(targetFields.contains(upperMeta),
+ "target should keep the declared upper-case reserved column, got " +
+ targetFields.mkString(","))
+ assert(copyFields == targetFields,
+ s"downstream copy ${copyFields.mkString(",")} should match " +
+ s"target ${targetFields.mkString(",")}")
+ }
+
+ test("an already-materialized target with an upper-cased reserved column
rejects a data-only " +
+ "declaration that omits it") {
+ // Upgrade path, the omitted-declaration case: a target already exists
with the reserved
+ // metadata column upper-cased (allowed under case-insensitive AUTO CDC).
The user adopts the
+ // new data-only declaration and drops the column, so this run resolves it
to the canonical
+ // lower-case name while the incremental merge keeps the existing
upper-case one. A
+ // case-sensitive downstream `SELECT *` could not match the two names, so
the run is rejected
+ // up front with an actionable error rather than left to fail opaquely
during execution.
+ // Reconciling at analysis time is not possible: the read path cannot see
the existing catalog
+ // casing, nor whether this run keeps it (incremental) or rewrites it to
canonical (refresh).
+ val upperMeta =
AutoCdcReservedNames.cdcMetadataColName.toUpperCase(Locale.ROOT)
+ val del = Scd1BatchProcessor.cdcDeleteSequenceFieldName
+ val ups = Scd1BatchProcessor.cdcUpsertSequenceFieldName
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, " +
+ s"$upperMeta STRUCT<$del:BIGINT,$ups:BIGINT> NOT NULL)")
+
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("version", LongType, nullable = false)
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 5L))
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(name = "auto_cdc_flow", target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "version")),
+ keys = Seq("id"), sequencing = functions.col("version")))
+ }
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition =
"AUTOCDC_INVALID_STATE.RESERVED_METADATA_COLUMN_CASING_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
+ "existingColumnName" -> upperMeta,
+ "resolvedColumnName" -> AutoCdcReservedNames.cdcMetadataColName
+ )
+ )
+ }
+
+ test("an already-materialized target's reserved-column nested casing is
preserved for a " +
+ "case-sensitive downstream SELECT *") {
+ // Upgrade path at the nested level: the target already carries the
reserved metadata column
+ // with its nested fields declared upper-cased (allowed under
case-insensitive AUTO CDC).
+ // evolveTable merges that existing spelling in, so the read path must
report the same nested
+ // casing, otherwise a case-sensitive downstream `SELECT *` plans the
canonical lower-case
+ // nested names and cannot resolve the upper-case fields the target
actually has.
+ val meta = AutoCdcReservedNames.cdcMetadataColName
+ val upperDel =
Scd1BatchProcessor.cdcDeleteSequenceFieldName.toUpperCase(Locale.ROOT)
+ val upperUps =
Scd1BatchProcessor.cdcUpsertSequenceFieldName.toUpperCase(Locale.ROOT)
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, " +
+ s"$meta STRUCT<$upperDel:BIGINT,$upperUps:BIGINT> NOT NULL)")
+
+ val metadataType = new StructType().add(upperDel, LongType).add(upperUps,
LongType)
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("version", LongType, nullable = false)
+ .add(meta, metadataType, nullable = false)
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 5L))
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(name = "auto_cdc_flow", target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "version")),
+ keys = Seq("id"), sequencing = functions.col("version")))
+ registerMaterializedView("copy", catalog = Some(catalog), database =
Some(namespace),
+ sqlConf = Map(SQLConf.CASE_SENSITIVE.key -> "true"),
+ query = readFlowFunc(s"$catalog.$namespace.target"))
+ }
+ runPipeline(ctx)
+
+ def nestedNames(table: String): Seq[String] =
+
spark.table(table).schema(meta).dataType.asInstanceOf[StructType].fieldNames.toSeq
+ val targetNested = nestedNames(s"$catalog.$namespace.target")
+ val copyNested = nestedNames(s"$catalog.$namespace.copy")
+ assert(targetNested == Seq(upperDel, upperUps),
+ s"target should keep the declared upper-case nested names, got
${targetNested.mkString(",")}")
+ assert(copyNested == targetNested,
+ s"downstream copy nested ${copyNested.mkString(",")} should match " +
+ s"target ${targetNested.mkString(",")}")
+ }
+
+ test("an already-materialized target with upper-cased reserved nested fields
rejects a " +
+ "data-only declaration that omits the metadata column") {
+ // Upgrade path, the omitted-declaration case at the nested level: the
target already carries
+ // the reserved metadata column with upper-cased nested fields. The user
adopts a data-only
+ // declaration and drops the column, so this run resolves the nested
fields to their canonical
+ // lower-case names while the incremental merge keeps the existing
upper-case ones. A
+ // case-sensitive downstream `SELECT *` could not match the two, so the
run is rejected up front
+ // with the same actionable error used for the outer column.
+ val meta = AutoCdcReservedNames.cdcMetadataColName
+ val del = Scd1BatchProcessor.cdcDeleteSequenceFieldName
+ val upperDel = del.toUpperCase(Locale.ROOT)
+ val upperUps =
Scd1BatchProcessor.cdcUpsertSequenceFieldName.toUpperCase(Locale.ROOT)
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, " +
+ s"$meta STRUCT<$upperDel:BIGINT,$upperUps:BIGINT> NOT NULL)")
+
+ val declaredSchema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("version", LongType, nullable = false)
+
+ val stream = MemoryStream[(Int, Long)]
+ stream.addData((1, 5L))
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace),
+ specifiedSchema = Some(declaredSchema))
+ registerFlow(autoCdcFlow(name = "auto_cdc_flow", target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "version")),
+ keys = Seq("id"), sequencing = functions.col("version")))
+ }
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition =
"AUTOCDC_INVALID_STATE.RESERVED_METADATA_COLUMN_CASING_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
+ "existingColumnName" -> s"$meta.$upperDel",
+ "resolvedColumnName" -> s"$meta.$del"
+ )
+ )
+ }
+
+ test("a declared reserved column's array-of-struct nested casing is
preserved for a " +
Review Comment:
**Nit (P3):** This is the only array-of-struct case, and it creates the
target fresh, so `existingTableOpt` is empty and the casing-drift check never
runs. That includes the `ArrayType` branch of `typeCasingDrifts` added in the
latest commit. Both drift-rejection tests use plain struct nesting, so deleting
the array branch leaves the suite green. Two other new paths also lack a
failure signal. No test runs a full refresh over a drifted target, even though
that is the remedy the new error advertises and it depends on the check staying
behind `isTableIncrementallyUpdated`. And SCD2 data-only declarations are only
validated in UserSpecifiedSchemaValidationSuite; nothing materializes such a
target or runs the SCD2 MERGE against it.
This extends an [existing
discussion](https://github.com/apache/spark/pull/57644#discussion_r4330516972).
**Verification:**
- **Regression:** An existing target whose metadata sequencing field is an
array of structs with a differently cased element field, rerun with a data-only
declaration, fails with the casing-drift error.
- **Behavior:** A full refresh over a target with a drifted reserved
spelling succeeds, leaves the canonical spelling, and a case-sensitive
downstream SELECT * matches the target.
- **Behavior:** An SCD2 AUTO CDC target with a data-only declaration is
created with the engine-owned metadata column and produces the expected SCD2
rows.
--
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]
