HyukjinKwon commented on code in PR #59299:
URL: https://github.com/apache/spark/pull/59299#discussion_r4285347256
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSchemaEvolution.scala:
##########
@@ -95,7 +96,12 @@ object ResolveSchemaEvolution extends Rule[LogicalPlan] {
plan: T,
oldOutput: Seq[Attribute],
newOutput: Seq[Attribute]): T = {
- val attrMap = AttributeMap(oldOutput.zip(newOutput))
+ // Match attributes by name rather than position: schema changes may add
fields at
+ // any position, shifting the ordinal of following columns.
+ val resolver = conf.resolver
+ val attrMap = AttributeMap(oldOutput.flatMap { oldAttr =>
Review Comment:
**Blocking (P1):** With `spark.sql.schemaEvolution.preserveColumnOrder` on,
`assignAddPositions` can insert a nested field ahead of existing siblings, but
this rebinding only swaps top-level attributes. `QueryPlan.rewriteAttrs`
replaces `AttributeReference`s only, so a `GetStructField` over a target column
that was resolved before evolution keeps its old ordinal, and `GetStructField`
reads `childSchema(ordinal)`.
Two kinds of expression are affected. One is MERGE `ON`, `WHEN MATCHED AND`
and `WHEN NOT MATCHED BY SOURCE AND` conditions, which are resolved in the same
`ResolveReferences` pass that expands the stars `ResolveSchemaEvolution` waits
for. The other is the `deleteExpr` of `INSERT WITH SCHEMA EVOLUTION ... REPLACE
WHERE`.
For example, take a target `info struct<salary int, status string>` and a
source `info struct<salary int, bonus int, status string>`:
```sql
MERGE WITH SCHEMA EVOLUTION INTO t USING s ON t.pk = s.pk
WHEN MATCHED AND t.info.status = 'active' THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
```
The condition is bound as `GetStructField(info, 1, "status")`. After `ADD
info.bonus AFTER salary`, ordinal 1 is `bonus`, so the condition compares
`bonus` with `'active'` and fails with an unrelated cast error. If the inserted
field has the same type as the shifted one (say a new `region string` ahead of
`status string`), the condition silently reads the new, all-NULL field, and the
MERGE updates or deletes the wrong rows. With the flag off, adds are appended
and no ordinal shifts, so the default path is unaffected. None of the new tests
uses a clause condition or a nested predicate, so nothing catches this.
**Recommended change:** After the attribute rebinding in replaceWriteTarget,
run a bottom-up expression transform over the rewritten write node. It
re-derives each GetStructField ordinal from its recorded field name against the
child's current struct type, and each GetArrayStructFields
field/ordinal/numFields from the element struct, using the session resolver.
Then add flag-on MERGE and INSERT REPLACE WHERE schema-evolution tests in which
a condition or predicate reads an existing nested target field placed after a
positioned nested add, with both a different-typed and a same-typed inserted
field.
**Why this works:** Running child-first means each extraction sees its
child's already-rebound type, so chained paths such as info.addr.zip resolve
level by level. Looking the field up by its recorded name restores the binding
the analyzer chose originally. When the ordinal already matches (flag off, or
appended adds), the transform is a no-op, so the top-level by-name map plus
this nested pass together preserve every pre-evolution binding.
**Scope:** Make post-evolution rebinding preserve nested field bindings, not
just top-level attribute identity, and add regression tests for nested target
references after positioned nested adds.
**Compatibility:** The flag-off path and the top-level by-name attribute
rebinding stay unchanged. Positions continue to be assigned after deduplication
and connector filtering.
**Risks:** A transform that rebinds extractions on the source side, or on
non-target children, must leave them unchanged. Rebinding by recorded name is a
no-op when the ordinal already matches, but it must not raise for extractions
whose child type did not change. Ambiguous case-insensitive names in an evolved
struct must surface the existing ambiguity error rather than picking an
arbitrary field.
**Constraints:** Use the session resolver, consistent with the by-name
top-level rebinding and findNestedField. Leave GetStructField with name = None
unchanged. With spark.sql.schemaEvolution.preserveColumnOrder off, the analyzed
plans and results of existing schema-evolution statements must not change.
**Success:** After a positioned nested add, every target-side field
extraction resolved before evolution reads the same named field it read before
evolution. A flag-on MERGE whose WHEN MATCHED AND condition reads info.status,
with a field inserted ahead of status, updates exactly the rows whose status
matches. A flag-on by-name INSERT ... REPLACE WHERE whose predicate reads an
existing nested field after a positioned nested add deletes and replaces
exactly the matching rows. With the flag off, schema-evolution plans and
results are unchanged.
**Verification:**
- **Regression:** A flag-on MERGE with a WHEN MATCHED AND condition on an
existing nested target field placed after a positioned nested add of a
different type evaluates the intended field and produces the expected rows
instead of a cast or type error.
- **Regression:** The same MERGE where the inserted nested field has the
same type as the referenced field updates exactly the rows whose referenced
field matches, proving the condition does not silently read the inserted field.
- **Regression:** A flag-on by-name INSERT WITH SCHEMA EVOLUTION ... REPLACE
WHERE with a predicate on an existing nested field after a positioned nested
add replaces exactly the rows matching that field.
- **Behavior:** A condition that reads a field of an array-of-struct column
after a positioned add inside the element struct still reads the intended
element field.
- **Compatibility:** Existing flag-off and preserve-order MERGE and INSERT
schema-evolution cases keep their expected schemas and rows.
--
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]
