[SPARK-58941][SDP] Sort schema inference flows by identifier parts to avoid dotted-name collisions - #58223
[SPARK-58941][SDP] Sort schema inference flows by identifier parts to avoid dotted-name collisions#58223anew wants to merge 3 commits into
Conversation
…o avoid dotted-name collisions `SchemaInferenceUtils.inferSchemaFromFlows` sorts the flows it merges by `identifier.unquotedString` to make the surviving spelling of a case-only-differing column deterministic. `unquotedString` joins the identifier's name parts with an unescaped `.`, so two distinct identifiers whose parts contain dots (e.g. a back-tick-quoted schema or flow name) can render to the same string. A stable `sortBy` then falls back to the incoming flow order -- the nondeterministic flow-resolution completion order -- so the surviving column casing could flip between runs. Sort on `quotedString` instead: it back-tick-quotes each part (escaping embedded back-ticks), so the ordering stays one-to-one with the identifier and the two names can no longer collide. Follow-up to SPARK-58517.
|
LGTM |
| // The merge order decides which spelling of a case-only-differing column survives, so it has to | ||
| // be deterministic and independent of the incoming flow order (which is the nondeterministic | ||
| // flow-resolution completion order). inferSchemaFromFlows sorts on the identifier's | ||
| // `quotedString`. Sorting on `unquotedString` would collapse these two DISTINCT identifiers to |
There was a problem hiding this comment.
nit: there's a lot of comment about unquoted string in this test, can we remove them as they are a bit repetitive and not that relevant as we now use 'quotedString'
| // the same key ("c.a.b.x"), and a stable sort would then fall back to the incoming order -- | ||
| // flipping the surviving column casing when the flows happen to arrive swapped. | ||
| val destination = TableIdentifier("t", Some("d"), Some("c")) | ||
| // Both identifiers render to "c.a.b.x" under unquotedString, but differ under quotedString. |
There was a problem hiding this comment.
like this one (we can maybe keep one version of them?)
| sqlConf = Map.empty)) | ||
| } | ||
|
|
||
| test("inferSchemaFromFlows merges deterministically for identifiers that collide under " + |
There was a problem hiding this comment.
optional: maybe dotted-name collision describes it better? again unquoted string is a bit specific to be persisted as comment, as its not used anymore.
| import org.apache.spark.sql.types.{IntegerType, StringType, StructType} | ||
|
|
||
| /** Tests for the flow ordering used by [[SchemaInferenceUtils.inferSchemaFromFlows]]. */ | ||
| class InferSchemaFromFlowsSuite extends QueryTest with SharedSparkSession { |
There was a problem hiding this comment.
can we put this in an existing suite rather than a suite just for one test?
There was a problem hiding this comment.
The fix looks right to me: quotedString is injective (back-tick doubling makes the token stream uniquely decodable, and TableIdentifier's assert(catalog.isEmpty || database.isDefined) rules out the one shape both keys would flatten), so the collision is gone. I also checked that the described collision is reachable through the flow registration path, and that the new test does fail on the old key.
One correction to the description, no action needed: the ordering also changes for ordinary dotless identifiers, not only dotted ones. ` (0x60) sorts after every character normally used in names while . (0x2E) sorts before them, so any pair where one name is a prefix of the other flips:
| A | B | unquotedString |
quotedString |
|---|---|---|---|
c.d.events |
c.d.events_us |
A first | B first |
c.d.f1 |
c.d.f10 |
A first | B first |
Roughly 2% of random identifier pairs reorder. The "no user-facing change" answer still holds since the sort is unreleased; it is just the stated reason that is off.
One more nit for existing test: MaterializeTablesSuite:1281 derives its expected surviving spelling by re-sorting the flows itself with .sortBy(_.identifier.unquotedString), so it now mirrors a key the code no longer uses. It passes today because f1/f2 order the same under both keys, but with names like f1/f10 it would assert the opposite of the correct behavior. Switching it to quotedString, or asserting the literal field names the way the sibling test at line 1292 does, would keep it honest.
Ah I missed that, and I don't really like it, it is not that intuitive. Let me sort by (catalog, database, table) instead. |
…into SchemaInferenceUtilsSuite Address review feedback on the schema-inference flow sort: - Sort on the identifier's parts `(catalog, database, table)` instead of `quotedString`. Both are injective (no dotted-name collision), but the parts tuple also preserves the existing order for dotless identifiers, whereas `quotedString` reordered some pairs (back-tick 0x60 sorts after name characters, dot 0x2E before them). Flow identifiers are always fully qualified, so all three parts are present and the ordering never depends on an absent catalog/database. - Move the `inferSchemaFromFlows` test into the existing `SchemaInferenceUtilsSuite` (now `QueryTest with SharedSparkSession`) and delete the single-test suite; reword it around dotted-name folding and trim the comments.
anew
changed the title
…eTablesSuite The multi-flow case-only fold test derived its expected surviving spelling by re-sorting the flows with `.sortBy(_.identifier.unquotedString)` -- a key inferSchemaFromFlows no longer uses. Assert the literal field names instead, matching the sibling test, so the expectation is not coupled to a stale sort key.
3 participants
What changes were proposed in this pull request?
SchemaInferenceUtils.inferSchemaFromFlowsmerges the flows that write to a table in adeterministic 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.unquotedStringjoins the identifier's name parts with an unescaped.. Thatmapping 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 toc.a.b.x. BecausesortByis stable,colliding keys fall back to the incoming
flowsorder, which is the nondeterministicflow-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 partstuple 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), unlikeTableIdentifier.quotedString, whichwould additionally reorder some ordinary pairs because
`(0x60) sorts after name characters.Flow identifiers are always fully qualified (
assertIsFullyQualifiedForCreate), so all three partsare 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.
unquotedStringis a lossykey, 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
diffSchemaskeys column identity on the exact name, so a run-to-run flip surfaces as adeleteColumn+addColumnfor a column that only changed case. Sorting on the identifier parts isinjective 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 theorder is unchanged.
How was this patch tested?
Added a unit test to
SchemaInferenceUtilsSuitethat builds two resolved flows whose identifiers(
`c`.`a.b`.`x`and`c`.`a`.`b.x`) would collide under a dot-joined key but stay distinctwhen 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
unquotedStringkey 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