You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anew (via GitHub)" <gi...@apache.org> on 2026/07/14 07:16:15 UTC
[PR] [SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow [spark]
anew opened a new pull request, #57245:
URL: https://github.com/apache/spark/pull/57245
### What changes were proposed in this pull request?
`GraphValidations.validateUserSpecifiedSchemas` looked up a table by the incoming flow's own identifier (`table.get(f.identifier)`). That only matches when the flow is an implicit/default flow whose identifier equals its destination table's. For a named flow (e.g. `CREATE FLOW <name> AS AUTO CDC INTO <target>`, or any explicitly-named flow) the lookup returned `None`, so the table's declared schema was never validated against the inferred schema.
This changes the lookup to key on `f.destinationIdentifier` (with `.distinct`, since multiple flows can share a destination), matching the sibling `validateFlowStreamingness` validation.
### Why are the changes needed?
An incompatible user-declared schema on a named-flow table went undetected at graph-validation time and surfaced only as a confusing mid-stream runtime failure at materialization.
### Does this introduce _any_ user-facing change?
Yes: an incompatible user-specified schema on a table fed by a named flow now fails with `USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE` at validation time, as it already did for the implicit-flow form.
### How was this patch tested?
New `UserSpecifiedSchemaValidationSuite` asserts the error is raised for both the implicit- and named-flow forms; confirmed the named-flow case fails without the fix. Ran ConnectValid/Invalid, SqlPipeline, and AutoCdc pipeline suites (130 tests) green.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57245:
URL: https://github.com/apache/spark/pull/57245#discussion_r3591790870
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -253,7 +253,12 @@ trait GraphValidations extends Logging {
}
protected def validateUserSpecifiedSchemas(): Unit = {
- flows.flatMap(f => table.get(f.identifier)).foreach { t: TableElement =>
+ // Look up tables by their destination identifier, not by the flow's own identifier. The two
+ // coincide only for an implicit/default flow (whose identifier equals its destination
+ // 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` is
Review Comment:
nit: can we remove this comment 'flowsTo' is grouped by destination... mirroring 'validateFlowStreamingness'
(seems too low level)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57245:
URL: https://github.com/apache/spark/pull/57245#discussion_r3592216800
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -253,7 +253,12 @@ trait GraphValidations extends Logging {
}
protected def validateUserSpecifiedSchemas(): Unit = {
- flows.flatMap(f => table.get(f.identifier)).foreach { t: TableElement =>
+ // Look up tables by their destination identifier, not by the flow's own identifier. The two
+ // coincide only for an implicit/default flow (whose identifier equals its destination
+ // 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` is
Review Comment:
ok
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #57245: [SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow
URL: https://github.com/apache/spark/pull/57245
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57245:
URL: https://github.com/apache/spark/pull/57245#discussion_r3591526526
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -253,7 +253,12 @@ trait GraphValidations extends Logging {
}
protected def validateUserSpecifiedSchemas(): Unit = {
- flows.flatMap(f => table.get(f.identifier)).foreach { t: TableElement =>
+ // Look up the destination table of each flow, not the table matching the flow's own
+ // identifier. The two coincide only for an implicit/default flow (whose identifier equals its
+ // destination 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. `distinct`
+ // collapses the multiple flows that can share a single destination.
+ flows.flatMap(f => table.get(f.destinationIdentifier)).distinct.foreach { t: TableElement =>
Review Comment:
yes good catch
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57245:
URL: https://github.com/apache/spark/pull/57245#discussion_r3591508531
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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 org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{ChangeArgs, ScdType, UnqualifiedColumnName}
+import org.apache.spark.sql.pipelines.utils.{PipelineTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Tests for `GraphValidations.validateUserSpecifiedSchemas`, which requires a table's
+ * user-declared schema to match the schema inferred from its incoming flows.
+ *
+ * The validation must apply regardless of whether the incoming flow's identifier equals the
+ * destination table's identifier (an implicit/default flow) or differs from it (a separately-named
+ * flow). Named flows previously bypassed the check because the table lookup was keyed on the flow
+ * identifier rather than the destination identifier (SPARK-58116). The validation operates on the
+ * resolved dataflow graph, so the flow type is immaterial to that lookup; this suite covers both
+ * plain flows and AUTO CDC flows, in the implicit and named forms, for compatible and incompatible
+ * declared schemas.
+ *
+ * For an AUTO CDC flow the inferred schema is the source's data columns plus an appended reserved
+ * `__spark_autocdc_metadata` column, so a declared schema listing only the data columns is
+ * incompatible (it is missing the metadata column) while one that also includes the metadata
+ * column is compatible.
+ */
+class UserSpecifiedSchemaValidationSuite extends PipelineTest with SharedSparkSession {
+
+ /** Source change feed with data columns `(id, name, version)`. */
+ private def sourceDf = {
+ val session = spark
+ import session.implicits._
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 1L))
+ stream.toDF().toDF("id", "name", "version")
+ }
+
+ /** The data-column schema produced by a plain flow (and the pre-metadata AUTO CDC schema). */
+ private def dataSchema: StructType = sourceDf.schema
+
+ /** A declared schema that omits a data column the flow produces, hence incompatible. */
+ private def dataSchemaMissingColumn: StructType = StructType(dataSchema.dropRight(1))
+
+ private def targetIdentifier = fullyQualifiedIdentifier("target")
+
+ /** Registers a plain flow into `target`; `flowName == "target"` yields an implicit flow. */
+ private def plainGraph(flowName: String, declaredSchema: Option[StructType]): DataflowGraph = {
+ val ctx = new TestGraphRegistrationContext(spark)
+ if (flowName == "target") {
+ ctx.registerTable(
+ "target",
+ query = Some(ctx.dfFlowFunc(sourceDf)),
+ specifiedSchema = declaredSchema)
+ } else {
+ ctx.registerTable("target", specifiedSchema = declaredSchema)
+ ctx.registerFlow(
+ destinationName = "target", name = flowName, query = ctx.dfFlowFunc(sourceDf))
+ }
+ ctx.resolveToDataflowGraph()
+ }
+
+ /** Registers an AUTO CDC flow into `target`; `flowName == "target"` yields an implicit flow. */
+ private def autoCdcGraph(flowName: String, declaredSchema: Option[StructType]): DataflowGraph = {
+ val ctx = new TestGraphRegistrationContext(spark)
+ ctx.registerTable("target", specifiedSchema = declaredSchema)
+ ctx.registerFlow(AutoCdcFlow(
+ identifier = fullyQualifiedIdentifier(flowName),
+ destinationIdentifier = targetIdentifier,
+ func = ctx.dfFlowFunc(sourceDf),
+ queryContext = QueryContext(
+ currentCatalog = catalogInPipelineSpec,
+ currentDatabase = databaseInPipelineSpec),
+ origin = QueryOrigin.empty,
+ changeArgs = ChangeArgs(
+ keys = Seq(UnqualifiedColumnName(Seq("id"))),
+ sequencing = functions.col("version"),
+ columnSelection = None,
+ deleteCondition = None,
+ storedAsScdType = ScdType.Type1)))
+ ctx.resolveToDataflowGraph()
+ }
+
+ /** The full inferred AUTO CDC output schema (data columns plus the reserved metadata column). */
+ private def autoCdcInferredSchema(flowName: String): StructType =
+ autoCdcGraph(flowName, declaredSchema = None).inferredSchema(targetIdentifier)
+
+ private def assertSchemaIncompatible(graph: DataflowGraph): Unit = {
+ val ex = intercept[AnalysisException](graph.validate())
+ assert(ex.getCondition == "USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE")
+ }
Review Comment:
ok that makes sense
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57245:
URL: https://github.com/apache/spark/pull/57245#discussion_r3591478900
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -253,7 +253,12 @@ trait GraphValidations extends Logging {
}
protected def validateUserSpecifiedSchemas(): Unit = {
- flows.flatMap(f => table.get(f.identifier)).foreach { t: TableElement =>
+ // Look up the destination table of each flow, not the table matching the flow's own
+ // identifier. The two coincide only for an implicit/default flow (whose identifier equals its
+ // destination 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. `distinct`
+ // collapses the multiple flows that can share a single destination.
+ flows.flatMap(f => table.get(f.destinationIdentifier)).distinct.foreach { t: TableElement =>
Review Comment:
`flowsTo` is already grouped by `destinationIdentifier` and its keys are distinct, so we can iterate it directly and drop `.distinct`. This also mirrors the sibling `validateFlowStreamingness`, which walks `flowsTo`, keeping the two validations structurally aligned.
```suggestion
flowsTo.keys.flatMap(table.get).foreach { t: TableElement =>
```
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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 org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{ChangeArgs, ScdType, UnqualifiedColumnName}
+import org.apache.spark.sql.pipelines.utils.{PipelineTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Tests for `GraphValidations.validateUserSpecifiedSchemas`, which requires a table's
+ * user-declared schema to match the schema inferred from its incoming flows.
+ *
+ * The validation must apply regardless of whether the incoming flow's identifier equals the
+ * destination table's identifier (an implicit/default flow) or differs from it (a separately-named
+ * flow). Named flows previously bypassed the check because the table lookup was keyed on the flow
+ * identifier rather than the destination identifier (SPARK-58116). The validation operates on the
+ * resolved dataflow graph, so the flow type is immaterial to that lookup; this suite covers both
+ * plain flows and AUTO CDC flows, in the implicit and named forms, for compatible and incompatible
+ * declared schemas.
+ *
+ * For an AUTO CDC flow the inferred schema is the source's data columns plus an appended reserved
+ * `__spark_autocdc_metadata` column, so a declared schema listing only the data columns is
+ * incompatible (it is missing the metadata column) while one that also includes the metadata
+ * column is compatible.
+ */
+class UserSpecifiedSchemaValidationSuite extends PipelineTest with SharedSparkSession {
+
+ /** Source change feed with data columns `(id, name, version)`. */
+ private def sourceDf = {
+ val session = spark
+ import session.implicits._
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 1L))
+ stream.toDF().toDF("id", "name", "version")
+ }
+
+ /** The data-column schema produced by a plain flow (and the pre-metadata AUTO CDC schema). */
+ private def dataSchema: StructType = sourceDf.schema
+
+ /** A declared schema that omits a data column the flow produces, hence incompatible. */
+ private def dataSchemaMissingColumn: StructType = StructType(dataSchema.dropRight(1))
+
+ private def targetIdentifier = fullyQualifiedIdentifier("target")
+
+ /** Registers a plain flow into `target`; `flowName == "target"` yields an implicit flow. */
+ private def plainGraph(flowName: String, declaredSchema: Option[StructType]): DataflowGraph = {
+ val ctx = new TestGraphRegistrationContext(spark)
+ if (flowName == "target") {
+ ctx.registerTable(
+ "target",
+ query = Some(ctx.dfFlowFunc(sourceDf)),
+ specifiedSchema = declaredSchema)
+ } else {
+ ctx.registerTable("target", specifiedSchema = declaredSchema)
+ ctx.registerFlow(
+ destinationName = "target", name = flowName, query = ctx.dfFlowFunc(sourceDf))
+ }
+ ctx.resolveToDataflowGraph()
+ }
+
+ /** Registers an AUTO CDC flow into `target`; `flowName == "target"` yields an implicit flow. */
+ private def autoCdcGraph(flowName: String, declaredSchema: Option[StructType]): DataflowGraph = {
+ val ctx = new TestGraphRegistrationContext(spark)
+ ctx.registerTable("target", specifiedSchema = declaredSchema)
+ ctx.registerFlow(AutoCdcFlow(
+ identifier = fullyQualifiedIdentifier(flowName),
+ destinationIdentifier = targetIdentifier,
+ func = ctx.dfFlowFunc(sourceDf),
+ queryContext = QueryContext(
+ currentCatalog = catalogInPipelineSpec,
+ currentDatabase = databaseInPipelineSpec),
+ origin = QueryOrigin.empty,
+ changeArgs = ChangeArgs(
+ keys = Seq(UnqualifiedColumnName(Seq("id"))),
+ sequencing = functions.col("version"),
+ columnSelection = None,
+ deleteCondition = None,
+ storedAsScdType = ScdType.Type1)))
+ ctx.resolveToDataflowGraph()
+ }
+
+ /** The full inferred AUTO CDC output schema (data columns plus the reserved metadata column). */
+ private def autoCdcInferredSchema(flowName: String): StructType =
+ autoCdcGraph(flowName, declaredSchema = None).inferredSchema(targetIdentifier)
+
+ private def assertSchemaIncompatible(graph: DataflowGraph): Unit = {
+ val ex = intercept[AnalysisException](graph.validate())
+ assert(ex.getCondition == "USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE")
+ }
Review Comment:
Since the fix is that the error is now attributed to the right table for named flows, assert the message actually names the target too, not just the error condition. `targetIdentifier` is already in scope, so no signature change is needed.
```suggestion
private def assertSchemaIncompatible(graph: DataflowGraph): Unit = {
val ex = intercept[AnalysisException](graph.validate())
assert(ex.getCondition == "USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE")
assert(ex.getMessage.contains(targetIdentifier.unquotedString))
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #57245:
URL: https://github.com/apache/spark/pull/57245#issuecomment-4995521308
One more thing worth calling out about the latent bug: keying on `f.identifier` didn't only *skip* validation for named flows, it could validate the *wrong* table. If a named flow's identifier happened to collide with some other table's identifier, the old `table.get(f.identifier)` would pull up that unrelated table and validate it, while the flow's actual destination table went unchecked. Keying on the destination identifier removes that misattribution 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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org