yyanyy opened a new pull request, #58298:
URL: https://github.com/apache/spark/pull/58298
### What changes were proposed in this pull request?
When `V2TableRefreshUtil` refreshes a `DataSourceV2Relation` in
`ALLOW_NEW_FIELDS` mode, the relation
is now replaced with a `Project` over a relation that exposes the
**current** table schema. The
relation stays aligned with the physical scan, while the projection
recreates the **captured** output
— same names, expression IDs, data types, nullability and metadata — for the
already-analyzed parent
plan.
A new `CapturedSchemaProjection` builds that projection, recursing through
structs, arrays and maps
so a captured nested type can be rebuilt from a wider current one.
When the captured and current outputs already have the same shape, the
relation is returned
untouched, so plans are unchanged whenever the table has not changed. That
check also makes the
rewrite idempotent, since `transformDown` revisits the node it just produced.
#### Alternatives considered
- **Pass the relation's output down to the connector** so it can retain only
the roots Spark still
knows about. Rejected: `relation.output` is a Catalyst planner concept
rather than part of the
connector API, so this introduces a dependency from connectors onto
Spark's logical plan and needs
plumbing through several layers.
- **Widen the scan output instead** — keep every field `Scan.readSchema()`
returns, reusing the
captured `ExprId` where the name is known and a fresh attribute otherwise,
and let the existing
`V2ScanRelationPushDown` projection drop the extras. Smaller diff, but it
changes the shared
`toOutputAttrs`, whose other callers (including the row-level-operation
path) would need auditing
first. Attributes also cannot simply be dropped for unknown names: an
extra physical column in the
middle would shift row ordinals.
- **Interpose the projection at refresh time** (this PR) — done where the
captured output is still
in hand, leaving `toOutputAttrs` and its other callers untouched.
### Why are the changes needed?
Spark refreshes versioned DSv2 tables **between analysis and optimization**,
in
`QueryExecution`. `V2TableRefreshUtil` loads the current table and does:
```scala
r.copy(table = currentTable)
```
which intentionally swaps in the current table/snapshot while preserving the
analyzed output. That
leaves the relation internally inconsistent:
```
relation.output = [id, salary] // captured at analysis
refreshed table = the current version
scan data schema = [id, salary, new_column] // current snapshot
```
Spark then asks the connector to prune to `[id, salary]`. A source that does
not prune all the way
back reports more than that, and `PushDownUtils.pruneColumns` calls:
```scala
scan -> toOutputAttrs(scan.readSchema(), relation)
```
`toOutputAttrs` builds its name map only from the stored `relation.output`,
so:
```scala
a => a.withExprId(nameToAttr(a.name).exprId)
// ^ nameToAttr("new_column") throws
```
```
java.util.NoSuchElementException: key not found: new_column
```
The precise trigger is not "`readSchema` is wider than the query's project".
It is:
```
Scan.readSchema field names are not a subset of relation.output field names.
```
**Reporting more than was requested is permitted.**
`SupportsPushDownRequiredColumns.pruneColumns` states that an implementation
should try its best but
that partial pruning is acceptable, giving as its example a source that
cannot prune nested fields
and only prunes top-level columns. That example alone reproduces the bug,
without the source doing
anything non-compliant:
```scala
sql("CREATE TABLE t (id INT, person STRUCT<name: STRING>) USING ...")
sql("INSERT INTO t VALUES (1, named_struct('name', 'Alice'))")
val df = spark.table("t") // captures id,
person: STRUCT<name>
sql("ALTER TABLE t ADD COLUMN person.age INT FIRST") // the table changes
underneath
df.collect()
```
Spark pushes the captured `person: STRUCT<name>`; a source that prunes
top-level columns only honours
"give me `person`" and reports its current full type, and the query fails
plan validation:
```
[PLAN_VALIDATION_FAILED_RULE_IN_BATCH] Rule ...V2ScanRelationPushDown in
batch Early Filter and
Projection Push-Down generated an invalid plan: The plan output schema has
changed from
STRUCT<id: INT, person: STRUCT<name: STRING>> to
STRUCT<id: INT, person: STRUCT<age: INT, name: STRING>>
```
#### Why this is rarely observed
A one-shot query analyzes and optimizes back to back, so the window between
the captured output and
the refreshed table is normally too small to notice. It becomes
deterministic in two ways, and this
is not specific to Spark Connect — the refresh phase lives in common
`QueryExecution`:
1. Force analysis on a `DataFrame`, change the table, and only then run its
first action:
```scala
val df = spark.table("t").filter("salary < 999")
df.queryExecution.analyzed // analysis only: output = [id, salary]
// external ALTER TABLE ADD COLUMN + write
df.collect() // first optimization: refresh sees
new_column
```
Do not call `df.collect()` before the change in this form — that would
already evaluate and cache
the refresh phase for this `QueryExecution`.
2. Use a temp view created from a `Dataset`, which stores an analyzed plan
rather than SQL text. Each
action builds a new `QueryExecution` from that stored plan, so a baseline
read is still possible
before the change. This is what the tests below use.
#### A second, silent bug
Because `toOutputAttrs` keys by name, a captured metadata column whose name
a new data column has
taken collapses onto a single attribute together with that data column:
```
RelationV2[id#50, index#51, index#51] <- one expression ID, two
read-schema fields
```
`SELECT id, index`, where `index` was the metadata column when the plan was
captured, then returned
`[2, 99]` — the new data column's value — instead of `[2, 0]`, the row index
it asked for. Refresh now
fails with an internal error in that case. Reporting it with a user-facing
message belongs in
`V2TableUtil.validateCapturedMetadataColumns`, which does not detect the
conflict yet because it only
compares the captured metadata columns against the ones the connector still
reports;
`V2TableUtilSuite` pins that gap and a follow-up closes it.
### Does this PR introduce _any_ user-facing change?
Yes, three changes, all relative to `master`:
1. Queries against a source that prunes partially, run after the table
gained a column, previously
failed with `NoSuchElementException` or
`PLAN_VALIDATION_FAILED_RULE_IN_BATCH`. They now succeed
and return the captured schema.
2. A captured metadata column shadowed by a new data column of the same name
previously returned the
data column's values silently; it now raises an error.
3. No change when the table has not changed — the relation is returned
untouched and the plan is
identical.
### How was this patch tested?
New unit tests in `CapturedSchemaProjectionSuite` cover the projection at
the expression level:
structs, arrays of structs, arrays of arrays, map keys and values, three
levels of nested structs, and
a heterogeneous `struct -> map value -> array -> struct` chain — each adding
a field at every level,
with a different ordinal shift per level so a cross-level mixup cannot pass;
null preservation at each
level; case-sensitive and case-insensitive name matching; duplicate field
names; and every rejection
path.
New end-to-end tests in `DataSourceV2DataFrameSuite` use a stored-plan temp
view against a source that
reports every data column from `readSchema()`: flat and nested widening, a
column added first, an
external case-only rename, a metadata column across a wider scan, a metadata
column shadowed by a new
data column, and a self-join.
`V2TableUtilSuite` gains a characterization test pinning the validation
blind spot described above.
**Regression check.** With the rebinding call in `V2TableRefreshUtil`
reverted, 7 of the 8 end-to-end
tests fail:
| Test | Failure without the change |
|---|---|
| reconciles a wider partially-pruned scan with stored temp view output |
`NoSuchElementException: key not found: new_column` |
| recreates a captured nested schema from a partially-pruned scan |
`PLAN_VALIDATION_FAILED_RULE_IN_BATCH`, `STRUCT<id, person: STRUCT<name>>` ->
`STRUCT<id, person: STRUCT<age, name>>` |
| recreates a captured schema nested through a map and an array | same, for
`data: STRUCT<m: MAP<STRING, STRUCT<arr: ARRAY<STRUCT<v>>>>>` |
| preserves a captured metadata column across a wider partially-pruned scan
| `NoSuchElementException: key not found: new_column` |
| restores the captured column order when a column is added first |
`NoSuchElementException: key not found: new_column` |
| restores the captured column name after an external case-only rename |
`Aliases salary#9310 are dangling in the references for plan` |
| rebinds every relation of a self-joined partially-pruned table |
`NoSuchElementException: key not found: new_column` |
The eighth test does not enable partial pruning, so its scan reports only
the captured columns and
`toOutputAttrs` resolves all of them — which is why the failure is specific
to sources that report
names outside the captured output.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code 2.1.246
--
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]
