Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-6087653115 Good catch @szehon-ho, thanks for tracing it. Fixed both: - `appendEngineOwnedReservedFields` now reconciles the metadata struct recursively, so a declared nested name and order survive with the engine's types/nullability (the retained case). - `reservedFieldCasingDrifts` recurses too and reports nested drift with a qualified name (`__spark_autocdc_metadata.DELETESEQUENCE`), reusing the same error, so an omitted declaration that resolves the nested names to canonical against an upper-cased target is rejected up front (the omitted case). Two regression tests on your scenario: the retained declaration with a case-sensitive downstream `SELECT *` (now resolves), and the omitted declaration (now rejected with the drift error). Both fail without the change and pass with it. -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
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 (
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5864227694 @szehon-ho does the early rejection work for you, or would you rather I handle it differently? @anew anything else needed before this can go in? -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5675316420 @szehon-ho Went with the early rejection. Reconciling at analysis time isn't feasible: an incremental run keeps the existing spelling but a full refresh rewrites it to canonical, and the read path can't tell which run this is. So materializeTable now compares the existing reserved column against what this run resolves, and throws AUTOCDC_INVALID_STATE.RESERVED_METADATA_COLUMN_CASING_DRIFT before the merge, pointing to declare it with the existing casing or full refresh. Added a test for the upper-cased omit case. -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
szehon-ho commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5671578591 Thanks — preserving the declared field’s spelling and position fixes the case where the user continues to declare it. The remaining concern is when an existing user adopts this PR’s new behavior and removes that declaration. For example, suppose a case-insensitive AUTO CDC target already contains: ```text id, version, __SPARK_AUTOCDC_METADATA ``` The user then changes its declaration to the newly supported data-only schema: ```sql CREATE STREAMING TABLE target ( id INT, version BIGINT ); ``` and has a case-sensitive downstream dataset: ```sql CREATE MATERIALIZED VIEW copy AS SELECT * FROM target; ``` `VirtualTableInput` plans `copy` using the canonical lowercase `__spark_autocdc_metadata`, while incremental `evolveTable` preserves the existing uppercase catalog field. During execution, the case-sensitive downstream by-name write cannot match those names, even though the AUTO CDC MERGE itself succeeds. The fact that the analysis-time read path cannot currently see the catalog spelling explains the mismatch, but it does not resolve this upgrade path. Could we cover the omitted-declaration case in this PR? If reconciling the catalog schema during analysis is not feasible, detecting and rejecting this legacy mismatch early would still be preferable to a later downstream failure. -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5656747600 @szehon-ho Gentle ping. All comments addressed and CI is green. Anything else needed before this can go in? -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5489082422 @szehon-ho good catch, fixed. appendEngineOwnedReservedFields now keeps the declared reserved column's spelling and position and swaps in only the engine type, so the read path matches an existing target that has it in a different casing. Added the regression test you described and confirmed it fails without the change. Honest caveat: this covers declaring the column, not omitting it on an existing upper-cased target (the read path is analysis-time, so it can't see the catalog spelling). Can file a follow-up for that if you'd like. -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
szehon-ho commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5483130351 One upgrade-path concern: users were previously required to declare `__spark_autocdc_metadata`, and under case-insensitive AUTO CDC they could declare it with different casing or positioning. The new helper canonicalizes that field in `VirtualTableInput`, but incremental `evolveTable` merges the existing catalog schema first, so the existing field spelling and position win. For example, suppose an existing target has: ```text id, name, version, __SPARK_AUTOCDC_METADATA ``` Its AUTO CDC flow is case-insensitive, so the MERGE can write to that uppercase column. The same graph also contains a case-sensitive downstream materialized view: ```sql CREATE MATERIALIZED VIEW copy AS SELECT * FROM target; ``` After this change: ```text Schema used to plan/materialize copy: id, name, version, __spark_autocdc_metadata Schema read from the existing target during execution: id, name, version, __SPARK_AUTOCDC_METADATA ``` The AUTO CDC MERGE itself succeeds, but the downstream by-name write can fail because its case-sensitive resolver cannot match the two metadata names. Could `appendEngineOwnedReservedFields` instead: - Preserve the spelling and position of one resolver-matching declared reserved field. - Replace its type/nullability with the engine-owned definition. - Append the canonical field only when the declaration omits it. - Reject ambiguous multiple matches. A regression test could pre-create the uppercase legacy target, run AUTO CDC case-insensitively, and have a case-sensitive downstream materialized view read it in the same graph. cc @anew if you have any thoughts on this as well -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5320964107 @szehon-ho thanks for the detailed review, this was really helpful. I went with aligning the read path like you suggested instead of hiding the column. There's one shared helper now that both materializeTable and VirtualTableInput.load use to build the target schema, so a downstream SELECT * in the same graph plans against the columns the target actually gets created with. On the incremental question, I added a test that creates the target and then does a second run through evolveTable and checks the schema doesn't change between runs. The existing schema-evolution suites already cover the evolve path more broadly. -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799242412
##
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:
Fixed. The appended reserved fields go through .asNullable in the shared
helper now, so a target built from a declared schema matches one built from an
omitted schema. No more nullability diff when someone adds or removes a
declaration. Added a test that checks the metadata column is nullable in the
declared case.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799232860
##
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:
Added. AutoCdcReservedColumnMaterializationSuite has a test where a
downstream MV in the same graph does SELECT * from the target and asserts the
consumer's schema matches the materialized target. It fails without the
read-path fix.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799195384
##
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:
Done. Hoisted effectiveCaseSensitive and effectiveResolver above
outputSchema and reused them, so the flows only get walked once.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799221432
##
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 AS AUTO CDC INTO
`) 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:
Done, using that flows val now.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799202886
##
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:
Good catch. Both the materialization strip/append and the validation
comparison are gated on the table actually being an AUTO CDC target now.
Materialization checks autoCdcAuxTableSpecOpt, and validation and the read path
check GraphElementTypeUtils.isAutoCdcFlow (I promoted your private helper here
so all three share it). Both come down to the same thing, whether an
AutoCdcMergeFlow writes to the table. So a plain table gets compared exactly
again and a reserved-prefixed column a non-CDC flow produces can't get dropped.
Added a validation test for that, and I put in the comment about
inferredSchemas here being flow-only.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799195384
##
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:
Good catch. Both the materialization strip/append and the validation
comparison are gated on the table actually being an AUTO CDC target now.
Materialization checks autoCdcAuxTableSpecOpt, and validation and the read path
check GraphElementTypeUtils.isAutoCdcFlow (I promoted your private helper here
so all three share it). Both come down to the same thing, whether an
AutoCdcMergeFlow writes to the table. So a plain table gets compared exactly
again and a reserved-prefixed column a non-CDC flow produces can't get dropped.
Added a validation test for that, and I put in the comment about
inferredSchemas here being flow-only.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799193427
##
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:
Fixed. VirtualTableInput.load now builds the same schema materializeTable
does for a declared AUTO CDC target, through the shared helper, instead of
handing back the declaration as-is. So the plan and the created table agree.
Added a same-graph downstream SELECT * test that fails without this.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5320858858 > Thanks for the iterations here -- accepting the declaration and then materializing the engine-owned shape is the right split, and using the flow-effective resolver in both paths keeps this consistent with the rest of AUTO CDC. > > Inline comments below. Two things that seemed better here than inline: > > * **What should a downstream consumer see?** I'd recommend giving `VirtualTableInput.load` the same schema `materializeTable` produces (strip the declared reserved fields, append the flow-inferred ones) through one shared helper, rather than returning the declaration verbatim. Nothing projects the reserved column away today -- the external reads are bare `reader.table(...)` calls, and the cross-pipeline test in `AutoCdcScd1MultiPipelineSuite` drops it by hand with an explicit `.select(...)` -- so the contract is "visible on the target, project it away yourself". This PR makes it hidden at analysis time yet present at execution time and in the table, which is neither. > Step by step, with a data-only declaration on the target and an ordinary downstream consumer: > ```sql > CREATE STREAMING TABLE target (id INT, name STRING, version BIGINT); > CREATE FLOW writer AS AUTO CDC INTO target FROM STREAM(source) KEYS (id) SEQUENCE BY version; > CREATE MATERIALIZED VIEW enriched AS SELECT * FROM target; > ``` > > > > > > > > > > > > > 1. **Planning.** `enriched` resolves `target` through `VirtualTableInput`, which hands back the declaration as-is: 3 columns. > 2. **Materialization.** `target` is created with 4 columns, since this PR appends `__spark_autocdc_metadata`. `enriched` is created with the 3 columns from step 1. > 3. **Execution.** `enriched` is re-analyzed; `target` falls outside the subgraph and is read from the catalog instead, so `SELECT *` now expands to 4 columns. > 4. **Result.** 4 columns written into the 3-column `enriched`. > > Actually hiding the column from consumers instead would need a projection on the read path -- a larger change I wouldn't fold in here. > * **Incremental coverage.** Both new materialization tests create the target from scratch. Is a case where it already exists and goes through `evolveTable` with `mergeWithExistingSchema = true` worth adding, or do you consider that covered by the merge semantics? / -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 closed pull request #57644: [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column URL: https://github.com/apache/spark/pull/57644 -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799147033
##
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
+extends ExecutionTest
+with SharedSparkSession
+with AutoCdcGraphExecutionTestMixin {
+
+ test("materialization appends the engine-owned reserved metadata column when
the user " +
+"schema omits it") {
+val session = spark
+import session.implicits._
+
+// 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
Review Comment:
Added it. Session case-insensitive, flow case-sensitive. The user's
__SPARK_AUTOCDC_METADATA stays distinct from the engine's
__spark_autocdc_metadata, so both survive and the engine only writes the
lower-case one. The user's column just stays null. Also checked it fails
without the fix.
One thing worth flagging: the table ends up with two columns that differ
only in case, so you have to read it back under case-sensitive analysis.
Reading it case-insensitively throws AMBIGUOUS_REFERENCE, which is expected, so
the test does the read that way.
--
This is an automated message from the Apache Git Service.
To respond to the
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799147033
##
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
+extends ExecutionTest
+with SharedSparkSession
+with AutoCdcGraphExecutionTestMixin {
+
+ test("materialization appends the engine-owned reserved metadata column when
the user " +
+"schema omits it") {
+val session = spark
+import session.implicits._
+
+// 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
Review Comment:
Added it. Session case-insensitive, flow case-sensitive. The user's
__SPARK_AUTOCDC_METADATA stays distinct from the engine's
__spark_autocdc_metadata, so both survive and the engine only writes the
lower-case one. The user's column just stays null. Also checked it fails
without the fix.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799142580
##
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
+extends ExecutionTest
+with SharedSparkSession
+with AutoCdcGraphExecutionTestMixin {
+
+ test("materialization appends the engine-owned reserved metadata column when
the user " +
+"schema omits it") {
+val session = spark
+import session.implicits._
Review Comment:
Good call, done. Pulled it up to one import testImplicits._ on the suite and
dropped the per-test val session = spark; import session.implicits._
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
szehon-ho commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3798212958
##
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:
`VirtualTableInput.load` returns the declared schema 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
```
So once `outputSchema` differs from `ss`, a same-graph downstream dataset is
planned against the declared columns but reads the real table at execution
(`reanalyzeFlow` leaves the upstream table out of the subgraph, so it resolves
as an external catalog read).
See the first point in my review comment for the step-by-step and the fix
I'd suggest.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
szehon-ho commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3798212958
##
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:
Relaxing the validation breaks an invariant the planning path depends on.
`validateUserSpecifiedSchemas` used to require `inferredSchema == ss`, so a
declared schema was always exactly what the table materialized with, and
`VirtualTableInput.load` relies on that -- it 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
is planned against the declared columns but reads the real table at execution:
`reanalyzeFlow` builds a subgraph whose `tables` is only the destination,
`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.
The worked example and the fix I'd suggest are in the review body. 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 downstream; either
way I think it needs a test with a downstream `SELECT *` over an AUTO CDC
target that has a data-only declaration.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
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
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
anew commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3794099031
##
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
+extends ExecutionTest
+with SharedSparkSession
+with AutoCdcGraphExecutionTestMixin {
+
+ test("materialization appends the engine-owned reserved metadata column when
the user " +
+"schema omits it") {
+val session = spark
+import session.implicits._
Review Comment:
minor: can this be imported once per suite, instead of for every test?
##
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
+extends ExecutionTest
+with SharedSparkSession
+with AutoCdcGraphExecutionTestMixin {
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3772896238
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala:
##
@@ -494,3 +494,22 @@ class AutoCdcMergeFlow(
}
}
+
+object AutoCdcMergeFlow {
Review Comment:
Done. Moved isReservedFieldName / reservedFields / stripReservedFields into
AutoCdcReservedNames next to the prefix, and removed the now-empty
AutoCdcMergeFlow companion.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5276703735 > Test coverage gap: the new tests cover data-only schemas, named vs implicit flows, wrong data columns, full metadata schemas, and SCD2 omission, but they do not cover the case where the AutoCDC flow’s effective case sensitivity differs from the Spark session. That is the important edge case for this change because the surrounding AutoCDC code derives resolver behavior from pipeline/flow SQL conf, not just the session. I would add coverage with session `spark.sql.caseSensitive=false` and pipeline/flow `spark.sql.caseSensitive=true` (and/or the reverse), plus a materialization-level test to verify the metadata column is appended or not appended under the effective resolver. I would also add a negative test where the user declares `__spark_autocdc_metadata` with the wrong type/nullability so validation catches it before MERGE time. Added coverage in a new AutoCdcReservedColumnMaterializationSuite. There's a materialization-level test with the session case-sensitive and the flow case-insensitive: the user declares the reserved column as __SPARK_AUTOCDC_METADATA and the created table ends up with exactly one engine-owned __spark_autocdc_metadata. I checked it has teeth, with the old session resolver it fails with AMBIGUOUS_REFERENCE on a duplicated metadata column, which is the case you flagged. Also added the wrong-type negative test, and the omit-and-append case lives there too -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3772872662
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##
@@ -269,9 +271,13 @@ trait GraphValidations extends Logging {
)
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 (AutoCdcMergeFlow.stripReservedFields(inferredSchema, resolver) !=
Review Comment:
Went with always materializing the engine-owned shape. DatasetManager strips
whatever reserved column the user declared and appends the inferred
engine-owned one, so the created table always matches what the MERGE writes.
While adding tests I also confirmed a conflicting-type declaration never
reaches materialization anyway, mergeSchemas can't merge a user-declared
__spark_autocdc_metadata of a different type against the inferred struct, so
it's rejected at inference with
USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE before MERGE time. Added a
negative test for that.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3772866357
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##
@@ -317,9 +317,21 @@ 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). Append any that the incoming flows produce but
the user omitted,
+// so the created table matches what the AUTO CDC MERGE writes at
runtime. Matching goes
+// through the session resolver so a reserved column the user declared
in a different case
+// is treated as present (not re-appended as a duplicate) under
case-insensitive analysis.
+val resolver = context.spark.sessionState.conf.resolver
Review Comment:
Fixed. Both the materialization path (DatasetManager) and the
schema-validation path (GraphValidations) now take the resolver from the flow's
effective case sensitivity via SchemaInferenceUtils.effectiveCaseSensitivity
instead of the session resolver, so a pipeline-level spark.sql.caseSensitive
that differs from the session is honored the same way the rest of AUTO CDC does
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
anew commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3770192916
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala:
##
@@ -494,3 +494,22 @@ class AutoCdcMergeFlow(
}
}
+
+object AutoCdcMergeFlow {
Review Comment:
Minor placement thought (non-blocking): `AutoCdcReservedNames` already owns
the reserved `prefix` and `cdcMetadataColName`, and its docs are the canonical
statement of what AUTO CDC reserves. Since `isReservedFieldName` /
`reservedFields` / `stripReservedFields` are all just "is this name/schema
reserved?" predicates keyed off that same prefix, they'd read more naturally in
`AutoCdcReservedNames` (co-located with the definition of what "reserved"
means) than in the `AutoCdcMergeFlow` companion. That keeps the single source
of truth for reserved-column semantics in one object, and avoids growing an
`AutoCdcMergeFlow` companion purely for schema predicates. Feel free to keep it
here if you'd rather have it next to the flow that actually produces the
columns.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
anew commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5271488939 Test coverage gap: the new tests cover data-only schemas, named vs implicit flows, wrong data columns, full metadata schemas, and SCD2 omission, but they do not cover the case where the AutoCDC flow’s effective case sensitivity differs from the Spark session. That is the important edge case for this change because the surrounding AutoCDC code derives resolver behavior from pipeline/flow SQL conf, not just the session. I would add coverage with session `spark.sql.caseSensitive=false` and pipeline/flow `spark.sql.caseSensitive=true` (and/or the reverse), plus a materialization-level test to verify the metadata column is appended or not appended under the effective resolver. I would also add a negative test where the user declares `__spark_autocdc_metadata` with the wrong type/nullability so validation catches it before MERGE time. -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
anew commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3769497937
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##
@@ -317,9 +317,21 @@ 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). Append any that the incoming flows produce but
the user omitted,
+// so the created table matches what the AUTO CDC MERGE writes at
runtime. Matching goes
+// through the session resolver so a reserved column the user declared
in a different case
+// is treated as present (not re-appended as a duplicate) under
case-insensitive analysis.
+val resolver = context.spark.sessionState.conf.resolver
Review Comment:
This should use the target AutoCDC flow’s effective resolver rather than the
Spark session resolver. Other AutoCDC paths derive case sensitivity from the
flows writing to the target because pipeline-level `SET
spark.sql.caseSensitive` can differ from the session
(`SchemaInferenceUtils.effectiveCaseSensitivity`,
`AutoCdcMergeFlow.effectiveResolver`, and the existing materialization resolver
below). With session case-insensitive but pipeline case-sensitive, a
user-declared `__SPARK_AUTOCDC_METADATA` can be treated as matching the
engine-owned lower-case metadata column, so materialization may skip appending
the actual column that the case-sensitive AutoCDC MERGE writes. The reverse
direction can append a duplicate under effective case-insensitive analysis. The
schema-validation path should use the same effective resolver too.
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##
@@ -269,9 +271,13 @@ trait GraphValidations extends Logging {
)
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 (AutoCdcMergeFlow.stripReservedFields(inferredSchema, resolver) !=
Review Comment:
This also drops the declared reserved field before checking compatibility,
which means a user schema that declares `__spark_autocdc_metadata` with the
wrong type/nullability is accepted. Since `DatasetManager` then treats a
same-name reserved field as present and does not append the inferred
engine-owned field, the target can be created with a malformed metadata column
and fail later in the AutoCDC MERGE. If declaring the internal column remains
supported, I think we should either validate the declared reserved field
against the inferred field or always materialize the inferred engine-owned
shape.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5269329590 The prefix handling already covers SCD-2. __START_AT and __END_AT don't carry the __spark_autocdc_ prefix so they stay user-visible, and only the metadata column is stripped. I added a test that declares an SCD-2 schema omitting just the metadata column to lock it in, so no code change was needed. Small note, the metadata column is __spark_autocdc_metadata, not __spark_cdc_metadata. -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3768070732
##
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala:
##
@@ -182,14 +183,24 @@ class UserSpecifiedSchemaValidationSuite extends
PipelineTest with SharedSparkSe
// AUTO CDC flows: the inferred schema appends a reserved metadata column to
the data columns.
- test("data-only user-specified schema is rejected for an implicit AUTO CDC
flow") {
-// Schema lists the data columns but omits the appended metadata column,
so it is incompatible.
-assertSchemaIncompatible(autoCdcGraph(flowName = "target", declaredSchema
= Some(dataSchema)))
+ test("SPARK-58118: data-only user-specified schema is accepted for an
implicit AUTO CDC flow") {
Review Comment:
Removed
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3768066953
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##
@@ -270,8 +270,12 @@ trait GraphValidations extends Logging {
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) {
+// inferred user-facing schema has columns that are not in the
specified one. The reserved
+// AUTO CDC metadata column(s) are engine-owned, so a declared schema
that differs from the
+// inferred schema only by omitting them is accepted; the engine
appends them to the
+// effective table schema at materialization.
+if (inferredSchema != ss &&
Review Comment:
Done. Comparing both schemas with the reserved fields removed now, so
declaring one of several reserved fields won't wrongly fail.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3767986448
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala:
##
@@ -494,3 +494,14 @@ class AutoCdcMergeFlow(
}
}
+
+object AutoCdcMergeFlow {
+
+ /** The engine-owned reserved AUTO CDC column(s) present in `schema`, if
any. */
Review Comment:
They're called on the user-declared schema too (stripReservedFields runs on
the declared schema in the validation check), so I threaded the resolver into
both, same as requireReservedPrefixAbsentInSourceColumns.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3767981372
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##
@@ -317,9 +317,19 @@ 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). Append any that the incoming flows produce but
the user omitted,
+// so the created table matches what the AUTO CDC MERGE writes at
runtime.
+val specifiedNames = ss.fieldNames.toSet
+val omittedReservedFields = AutoCdcMergeFlow
+ .reservedFields(inferredSchemas(table.identifier))
+ .filterNot(f => specifiedNames.contains(f.name))
Review Comment:
Good catch. Switched this to match through the session resolver, so a
reserved column declared in a different case is recognized and not re-appended
as a duplicate. One note, I didn't add a unit test for the case-insensitive
path. I tried, but schema inference reflects the declared column's case, which
made a validation-level test unreliable. The duplicate-append case here is the
one that really needs guarding, and that needs an integration test. Happy to
add one if you'd prefer.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
anew commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5259540842 Since SDC-2 is now fully implemented, can you extend this to SCD-2? In SCD-2, we want to consider `__START_AT` and `__END_AT` as not reserved, because they are part of the SCD-2 contract and expected in the output schema of the flow. But `__spark_cdc_metadata` is still reserved, same as in SCD-1. -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
anew commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3762119317
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##
@@ -270,8 +270,12 @@ trait GraphValidations extends Logging {
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) {
+// inferred user-facing schema has columns that are not in the
specified one. The reserved
+// AUTO CDC metadata column(s) are engine-owned, so a declared schema
that differs from the
+// inferred schema only by omitting them is accepted; the engine
appends them to the
+// effective table schema at materialization.
+if (inferredSchema != ss &&
Review Comment:
this works right now because there is exactly one reserved field. If, in the
future, we add another one, there could be a situation where the user declares
one out of two, and then both conditions won't be met.
it would be better to compare the version of both schemas that has the
reserved fields removed.
##
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala:
##
@@ -182,14 +183,24 @@ class UserSpecifiedSchemaValidationSuite extends
PipelineTest with SharedSparkSe
// AUTO CDC flows: the inferred schema appends a reserved metadata column to
the data columns.
- test("data-only user-specified schema is rejected for an implicit AUTO CDC
flow") {
-// Schema lists the data columns but omits the appended metadata column,
so it is incompatible.
-assertSchemaIncompatible(autoCdcGraph(flowName = "target", declaredSchema
= Some(dataSchema)))
+ test("SPARK-58118: data-only user-specified schema is accepted for an
implicit AUTO CDC flow") {
Review Comment:
minor: I don't think we need the Jira IDs in all these test names.
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala:
##
@@ -494,3 +494,14 @@ class AutoCdcMergeFlow(
}
}
+
+object AutoCdcMergeFlow {
+
+ /** The engine-owned reserved AUTO CDC column(s) present in `schema`, if
any. */
Review Comment:
are these two methods only called on the system-inferred schema? Or also on
the user-declared schema? If so, they need to honor the case sensitivity.
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##
@@ -317,9 +317,19 @@ 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). Append any that the incoming flows produce but
the user omitted,
+// so the created table matches what the AUTO CDC MERGE writes at
runtime.
+val specifiedNames = ss.fieldNames.toSet
+val omittedReservedFields = AutoCdcMergeFlow
+ .reservedFields(inferredSchemas(table.identifier))
+ .filterNot(f => specifiedNames.contains(f.name))
Review Comment:
does this need to account for the case sensitivity conf? If the user
specified the meta fields in different case?
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3726190119
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##
@@ -284,9 +284,18 @@ object DatasetManager extends Logging {
val (catalog, identifier) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
-val outputSchema = table.specifiedSchema.getOrElse(
- resolvedDataflowGraph.inferredSchema(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). Append any that the incoming flows produce but
the user omitted,
+// so the created table matches what the AUTO CDC MERGE writes at
runtime.
+val omittedReservedFields = AutoCdcMergeFlow
+
.reservedFields(resolvedDataflowGraph.inferredSchema(table.identifier))
+ .filterNot(f => ss.fieldNames.contains(f.name))
Review Comment:
Good call @HyukjinKwon , done. Switched ss.fieldNames to a Set before the
filterNot.
--
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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
naveenp2708 commented on PR #57644: URL: https://github.com/apache/spark/pull/57644#issuecomment-5186881313 Good call @HyukjinKwon , done. Switched ss.fieldNames to a Set before the filterNot. -- 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]
Re: [PR] [SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column [spark]
HyukjinKwon commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3717403120
##
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##
@@ -284,9 +284,18 @@ object DatasetManager extends Logging {
val (catalog, identifier) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
-val outputSchema = table.specifiedSchema.getOrElse(
- resolvedDataflowGraph.inferredSchema(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). Append any that the incoming flows produce but
the user omitted,
+// so the created table matches what the AUTO CDC MERGE writes at
runtime.
+val omittedReservedFields = AutoCdcMergeFlow
+
.reservedFields(resolvedDataflowGraph.inferredSchema(table.identifier))
+ .filterNot(f => ss.fieldNames.contains(f.name))
Review Comment:
Optional micro-nit: `ss.fieldNames` is recomputed (fresh Array) with a
linear `contains` per reserved field. `val specifiedNames =
ss.fieldNames.toSet` before the `filterNot` makes membership O(1).
Materialization-time only, so take-it-or-leave-it.
--
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]
