(spark) branch branch-4.x updated: [SPARK-58941][SDP] Sort schema inference flows by identifier parts to avoid dotted-name collisions

[sz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new 9c2e941c4fd8 [SPARK-58941][SDP] Sort schema inference flows by identifier parts to avoid dotted-name collisions
9c2e941c4fd8 is described below

commit 9c2e941c4fd8a201b90460bd53b7cc4801e39ea2
Author: Andreas Neumann <an...@apache.org>
AuthorDate: Mon Aug 24 11:28:34 2026 -0700

    [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 <an...@apache.org>
    Signed-off-by: Szehon Ho <sz...@gmail.com>
    (cherry picked from commit abc55fe2309e21c55b83b8293b7392674831502b)
    Signed-off-by: Szehon Ho <sz...@gmail.com>
---
 .../sql/pipelines/util/SchemaInferenceUtils.scala  |  6 +-
 .../pipelines/graph/MaterializeTablesSuite.scala   | 14 +---
 .../pipelines/util/SchemaInferenceUtilsSuite.scala | 88 +++++++++++++++++++++-
 3 files changed, 94 insertions(+), 14 deletions(-)

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 10b3202a61af..fd6068415074 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 @@ object SchemaInferenceUtils {
    * 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 @@ object SchemaInferenceUtils {
     )
 
     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 a4df9867e2ab..916169c4de2d 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 @@ abstract class MaterializeTablesSuite extends BaseCoreExecutionTest {
 
         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 e6aed97f35dd..92e1c700b99b 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 @@ class SchemaInferenceUtilsSuite extends SparkFunSuite {
       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")
+    }
+  }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org