szehon-ho commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3798212968
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -317,9 +318,23 @@ object DatasetManager extends Logging {
val (catalog, identifier) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
- val outputSchema = table.specifiedSchema.getOrElse(
- inferredSchemas(table.identifier).asNullable
- )
+ val outputSchema = table.specifiedSchema match {
+ case Some(ss) =>
+ // The user schema describes the logical table; the engine owns the
reserved AUTO CDC
+ // metadata column(s). Drop whatever reserved column(s) the user
declared and append the
+ // engine-owned shape from the inferred schema, so the created table
always has exactly the
+ // reserved column(s) the AUTO CDC MERGE writes, even if the user
declared one with a
+ // different type or nullability. Matching goes through the flow's
effective resolver (the
+ // same one the rest of AUTO CDC uses, which honors a case-sensitivity
conf set on the flow,
+ // not just the session).
+ val resolver = SchemaInferenceUtils.resolverFor(
+ effectiveCaseSensitivityFor(resolvedDataflowGraph, table.identifier,
context))
+ StructType(
+ AutoCdcReservedNames.stripReservedFields(ss, resolver).fields ++
Review Comment:
This runs for every table with a declared schema, with no check that an AUTO
CDC flow writes to it, and the same is true of the strip on both sides in
`GraphValidations`. Nothing outside `AutoCdcMergeFlow` rejects
reserved-prefixed names -- `requireReservedPrefixAbsentInSourceColumns` lives
on that flow and checks the source DataFrame, not user-declared table schemas.
So for an ordinary table whose user declares a column named
`__spark_autocdc_x`, `stripReservedFields(ss, ...)` drops it and the append
cannot put it back: `DataflowGraph.inferSchemas` passes `userSpecifiedSchema =
None`, so `inferredSchemas` here is purely flow-derived and never contains a
user-only column. Previously `outputSchema` was `ss` verbatim and the column
was created; now it is silently dropped, and on an MV or full refresh
`diffSchemas` turns that into a `deleteColumn` against an existing table. On
the validation side, a declared schema that omits a reserved-prefixed column a
non-CDC flow actually produces now passes where it used to fail.
`materializeTable` already computes `autoCdcAuxTableSpecOpt` and
`GraphValidations` now has `flows` in scope, so gating both on the table
actually having an AUTO CDC flow looks cheap and keeps the relaxation inside
the feature it was written for.
Separately, worth saying in the comment that `inferredSchemas` here is
flow-only, unlike the `inferredSchema` in `GraphValidations`, which merges the
declaration in. The correctness of "append the engine-owned shape" rests on
that difference and the two are hard to tell apart at the call site.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -317,9 +318,23 @@ object DatasetManager extends Logging {
val (catalog, identifier) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
- val outputSchema = table.specifiedSchema.getOrElse(
- inferredSchemas(table.identifier).asNullable
- )
+ val outputSchema = table.specifiedSchema match {
Review Comment:
Before this change, `validateUserSpecifiedSchemas` required `inferredSchema
== ss`, so a declared schema was always exactly the schema the table got
materialized with. Parts of the planning path rely on that; in particular
`VirtualTableInput.load` uses the declaration as-is and never consults the
materialized table:
```scala
val deducedSchema = specifiedSchema match {
// If the user specified a schema, use it directly.
case Some(ss) => ss
```
With `outputSchema` now differing from `ss`, a same-graph downstream dataset
resolves against the declared columns at analysis time but reads the real table
at execution time. `reanalyzeFlow` builds a subgraph whose `tables` is only the
destination, and `dfsInternal(..., stopAtMaterializationPoints = true)` skips
the upstream table, so it is not in `context.allInputs` and
`GraphIdentifierManager` classifies it as an `ExternalDatasetIdentifier` -- a
plain read of the four-column catalog table.
Concretely, with `CREATE STREAMING TABLE target (id INT, name STRING,
version BIGINT)` fed by an AUTO CDC flow, plus `CREATE MATERIALIZED VIEW
enriched AS SELECT * FROM target`: `target` materializes with four columns,
`enriched` is planned and materialized with three, and at execution `SELECT *`
yields four columns written into the three-column `enriched`.
I traced this statically rather than running it, so I can't say whether it
surfaces as an "extra fields" `AnalysisException` or as silent evolution on
`enriched`. Either way I think it needs a test with a downstream `SELECT *`
over an AUTO CDC target that has a data-only declaration. The cheapest fix I
can see is to factor "the schema this table materializes with" into one helper
and have both this method and `VirtualTableInput.load` call it.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -317,9 +318,23 @@ object DatasetManager extends Logging {
val (catalog, identifier) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
- val outputSchema = table.specifiedSchema.getOrElse(
- inferredSchemas(table.identifier).asNullable
- )
+ val outputSchema = table.specifiedSchema match {
+ case Some(ss) =>
+ // The user schema describes the logical table; the engine owns the
reserved AUTO CDC
+ // metadata column(s). Drop whatever reserved column(s) the user
declared and append the
+ // engine-owned shape from the inferred schema, so the created table
always has exactly the
+ // reserved column(s) the AUTO CDC MERGE writes, even if the user
declared one with a
+ // different type or nullability. Matching goes through the flow's
effective resolver (the
+ // same one the rest of AUTO CDC uses, which honors a case-sensitivity
conf set on the flow,
+ // not just the session).
+ val resolver = SchemaInferenceUtils.resolverFor(
+ effectiveCaseSensitivityFor(resolvedDataflowGraph, table.identifier,
context))
+ StructType(
+ AutoCdcReservedNames.stripReservedFields(ss, resolver).fields ++
+
AutoCdcReservedNames.reservedFields(inferredSchemas(table.identifier),
resolver))
+ case None =>
+ inferredSchemas(table.identifier).asNullable
Review Comment:
The `None` branch applies `.asNullable`, but the appended reserved field in
the `Some` branch keeps the engine's `nullable = false`
(`AutoCdcMergeFlow.schema` sets that for both SCD types). So the same pipeline
creates a NOT NULL metadata column when the user declares a schema and a
nullable one when they don't, and adding or removing a declaration between runs
shows up as an `updateColumnNullability` from `diffSchemas`. Applying
`.asNullable` to the appended fields would make the two branches agree.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -317,9 +318,23 @@ object DatasetManager extends Logging {
val (catalog, identifier) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
- val outputSchema = table.specifiedSchema.getOrElse(
- inferredSchemas(table.identifier).asNullable
- )
+ val outputSchema = table.specifiedSchema match {
+ case Some(ss) =>
+ // The user schema describes the logical table; the engine owns the
reserved AUTO CDC
+ // metadata column(s). Drop whatever reserved column(s) the user
declared and append the
+ // engine-owned shape from the inferred schema, so the created table
always has exactly the
+ // reserved column(s) the AUTO CDC MERGE writes, even if the user
declared one with a
+ // different type or nullability. Matching goes through the flow's
effective resolver (the
+ // same one the rest of AUTO CDC uses, which honors a case-sensitivity
conf set on the flow,
+ // not just the session).
+ val resolver = SchemaInferenceUtils.resolverFor(
+ effectiveCaseSensitivityFor(resolvedDataflowGraph, table.identifier,
context))
Review Comment:
`effectiveCaseSensitivityFor` is already computed a few lines below into
`effectiveCaseSensitive` / `effectiveResolver`. Each call walks the flows and
can throw `conflictingFlowConfigurationError`, so this does the work twice and
reads as though they might be two different resolvers. Hoisting the existing
pair above `outputSchema` and reusing it would cover both.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -258,20 +259,30 @@ 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) {
+ // Check the specified schema matches the inferred schema once the
engine-owned reserved
+ // AUTO CDC metadata column(s) are set aside on both sides. The user
may omit them (the
+ // engine appends them at materialization) or declare them; comparing
both schemas 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.
+ if (AutoCdcReservedNames.stripReservedFields(inferredSchema, resolver)
!=
+ AutoCdcReservedNames.stripReservedFields(ss, resolver)) {
val datasetType = GraphElementTypeUtils
.getDatasetTypeForMaterializedViewOrStreamingTable(
flowsTo(t.identifier).map(f => resolvedFlow(f.identifier))
Review Comment:
Nit: this can use the `flows` val hoisted at the top of the loop now.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcReservedColumnMaterializationSuite.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.{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
Review Comment:
Both cases here cover a target created from scratch and inspected directly.
Could we also add one where another dataset in the same graph does `SELECT *`
from the target? See the comment on `DatasetManager.materializeTable` -- that
is the case where the declared schema and the materialized schema differing
becomes observable.
--
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]
