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: [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]
