[SPARK-58941][SDP] Sort schema inference flows by identifier parts to avoid dotted-name collisions ### What changes were proposed in this pull request? `SchemaInferenceUtils.inferSchemaFromFlows` merges the flows that write to a table in a deterministic order so that, when two flows emit a column whose names differ only in case, the surviving spelling is well-defined and every caller agrees on it. It established that order by sorting on `flow.identifier.unquotedString`. `TableIdentifier.unquotedString` joins the identifier's name parts with an **unescaped** `.`. That mapping is not one-to-one: two structurally distinct identifiers whose parts contain dots (a dot is legal inside a back-tick-quoted schema or flow name) can render to the same string -- e.g. `` `c`.`a.b`.`x` `` and `` `c`.`a`.`b.x` `` both render to `c.a.b.x`. Because `sortBy` is stable, colliding keys fall back to the incoming `flows` order, which is the nondeterministic flow-resolution completion order the sort was meant to remove, so the surviving column casing could flip between runs. This PR sorts on the identifier's parts `(catalog, database, table)` instead. Comparing the parts tuple is injective -- two distinct identifiers are two distinct tuples, so they can never collide. It also preserves the existing order for dotless identifiers (a shorter part sorts first, exactly as the `.` separator, 0x2E, ordered a joined string), unlike `TableIdentifier.quotedString`, which would additionally reorder some ordinary pairs because `` ` `` (0x60) sorts after name characters. Flow identifiers are always fully qualified (`assertIsFullyQualifiedForCreate`), so all three parts are present and the ordering never depends on an absent catalog/database. This is a follow-up to SPARK-58517, which added the sort. ### Why are the changes needed? The sort exists to guarantee a deterministic surviving column spelling. `unquotedString` is a lossy key, so under identifiers that contain dots the guarantee silently breaks: the merge order reverts to the nondeterministic completion order of concurrent flow resolution, and a case-only column spelling can flip between runs of an unchanged pipeline. On the non-merging evolution paths `diffSchemas` keys column identity on the exact name, so a run-to-run flip surfaces as a `deleteColumn` + `addColumn` for a column that only changed case. Sorting on the identifier parts is injective and removes the collision. ### Does this PR introduce _any_ user-facing change? No. The merge order changes only for identifiers with dotted (back-tick-quoted) name parts, and the change is confined to the unreleased `master` / `branch-4.x`. For ordinary (dotless) identifiers the order is unchanged. ### How was this patch tested? Added a unit test to `SchemaInferenceUtilsSuite` that builds two resolved flows whose identifiers (`` `c`.`a.b`.`x` `` and `` `c`.`a`.`b.x` ``) would collide under a dot-joined key but stay distinct when sorted on their parts, each carrying a case-only-differing column, and asserts the inferred schema is identical regardless of the order the flows are passed in. The test fails with the previous `unquotedString` key and passes with the parts key. Ran `build/sbt 'pipelines/testOnly *SchemaInferenceUtilsSuite'` (13 tests, all passed). ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 4.8 Closes #58223 from anew/sdp-sort-flows-quoted-identifier. Authored-by: Andreas Neumann <anew@apache.org> Signed-off-by: Szehon Ho <szehon.apache@gmail.com>
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala index 10b3202..fd60684 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala
@@ -122,7 +122,9 @@ * with how the rest of the engine resolves those names. * * When flows differ only in column casing, the surviving spelling is the one from the flow with - * the lowest identifier: `flows` is merged in sorted identifier order, not in the order given. + * the lowest identifier: `flows` is merged in sorted identifier order, not in the order given. We + * sort on the identifier's parts (catalog, database, table) to avoid collisions for identifiers + * whose parts contain dots. * Sorting here rather than at the call sites keeps every caller agreeing on the result, since the * schemas they derive are compared against each other -- the graph's inferred schema materializes * the table, while [[org.apache.spark.sql.pipelines.graph.VirtualTableInput]] produces the schema @@ -151,7 +153,7 @@ ) val inferredSchema = flows - .sortBy(_.identifier.unquotedString) + .sortBy(f => (f.identifier.catalog, f.identifier.database, f.identifier.table)) .map(_.schema) .fold(new StructType()) { (schemaSoFar, schema) => try {
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala index a4df986..916169c 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala
@@ -1275,16 +1275,10 @@ val graph = ctx.resolveToDataflowGraph() val inferredSchemas = graph.inferSchemas(spark.sessionState.conf.caseSensitiveAnalysis) - val (targetIdentifier, inferred) = inferredSchemas.head - val lowestIdentifierFlowValueField = - graph.resolvedFlowsTo(targetIdentifier) - .sortBy(_.identifier.unquotedString) - .head - .schema - .fieldNames(1) - // The two spellings must fold into a single column, and the lowest flow identifier - // supplies the surviving spelling. - assert(inferred.fieldNames.toSeq === Seq("id", lowestIdentifierFlowValueField)) + val (_, inferred) = inferredSchemas.head + // The two spellings fold into a single column, and the lowest flow identifier supplies the + // surviving spelling: `f1` sorts before `f2`, so `value` wins. + assert(inferred.fieldNames.toSeq === Seq("id", "value")) } } }
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala index e6aed97..92e1c70 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala
@@ -17,11 +17,66 @@ package org.apache.spark.sql.pipelines.util -import org.apache.spark.{SparkException, SparkFunSuite} +import scala.util.Success + +import org.apache.spark.SparkException +import org.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.connector.catalog.TableChange +import org.apache.spark.sql.pipelines.graph.{ + FlowFunction, + FlowFunctionResult, + Input, + QueryContext, + QueryOrigin, + ResolvedFlow, + StreamingFlow, + UntypedFlow +} +import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types._ -class SchemaInferenceUtilsSuite extends SparkFunSuite { +class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { + + /** A [[FlowFunction]] that throws if invoked; the inferSchemaFromFlows test builds resolved + * flows directly. */ + private val noOpFlowFunction: FlowFunction = new FlowFunction { + override def call( + allInputs: Set[TableIdentifier], + availableInputs: Seq[Input], + configuration: Map[String, String], + queryContext: QueryContext, + queryOrigin: QueryOrigin): FlowFunctionResult = + throw new UnsupportedOperationException( + "noOpFlowFunction.call should not be invoked from SchemaInferenceUtilsSuite tests") + } + + private val queryContext = QueryContext(currentCatalog = Some("c"), currentDatabase = Some("d")) + + /** A resolved flow with the given identifier and output schema, writing to `destination`. */ + private def resolvedFlow( + identifier: TableIdentifier, + destination: TableIdentifier, + schema: StructType): ResolvedFlow = { + val df = spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema) + val flow = UntypedFlow( + identifier = identifier, + destinationIdentifier = destination, + func = noOpFlowFunction, + queryContext = queryContext, + sqlConf = Map.empty, + once = false, + origin = QueryOrigin.empty) + new StreamingFlow( + flow, + FlowFunctionResult( + requestedInputs = Set.empty, + batchInputs = Set.empty, + streamingInputs = Set.empty, + usedExternalInputs = Set.empty, + dataFrame = Success(df), + sqlConf = Map.empty)) + } test("determineColumnChanges - adding new columns") { val currentSchema = new StructType() @@ -398,4 +453,33 @@ assert(!changes.exists(_.isInstanceOf[TableChange.DeleteColumn])) } } + + test("inferSchemaFromFlows folds a case-only column to the same spelling regardless of flow " + + "order, even when identifier names contain dots") { + // The merge order decides which spelling of a case-only-differing column survives, so it must + // not depend on the incoming flow order (the nondeterministic flow-resolution completion + // order). The two identifiers below differ only in where the dot falls, so a dot-joined sort + // key would render them identical; sorting on the identifier parts keeps them distinct. + val destination = TableIdentifier("t", Some("d"), Some("c")) + val flowA = resolvedFlow( + identifier = TableIdentifier("x", Some("a.b"), Some("c")), + destination = destination, + schema = new StructType().add("id", IntegerType).add("value", StringType)) + val flowB = resolvedFlow( + identifier = TableIdentifier("b.x", Some("a"), Some("c")), + destination = destination, + schema = new StructType().add("id", IntegerType).add("Value", StringType)) + + // The lower identifier (flowB: database "a" precedes "a.b") supplies the surviving spelling, in + // either input order. + val expected = new StructType().add("id", IntegerType).add("Value", StringType) + Seq(Seq(flowA, flowB), Seq(flowB, flowA)).foreach { flows => + val inferred = SchemaInferenceUtils.inferSchemaFromFlows( + tableIdentifier = destination, + flows = flows, + userSpecifiedSchema = None, + sessionCaseSensitive = false) + assert(inferred === expected, s"unexpected schema for input order $flows") + } + } }