peter-toth commented on code in PR #59706:
URL: https://github.com/apache/spark/pull/59706#discussion_r4411981968
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -990,7 +991,26 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
}
def pruneColumns(plan: LogicalPlan): LogicalPlan = plan.transform {
+ case op @ SchemaPruning.GeneratorsOverScan(_, _, sHolder:
ScanBuilderHolder) =>
+ // Fields read from the generators' elements above them narrow the scan
below them. If the
+ // plan above cannot be re-resolved, the scan is pruned again on its own
below, which calls
+ // `pruneColumns` on the scan builder a second time.
Review Comment:
**Finding 1.** Iceberg breaks the "last call wins" assumption. Its builder
prunes the projection the previous call left
([`BaseSparkScanBuilder.java:139`](https://github.com/apache/iceberg/blob/29d21c2b73c2fb910cd7fa513dc2229327037846/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkScanBuilder.java#L139)),
and `SparkScanBuilder` in the 3.5 and 4.0 modules does the same. So a second,
wider `pruneColumns` can't widen the scan back, and a javadoc note would not
hold for it.
I measured it with a `ParquetScanBuilder` subclass whose `pruneColumns`
intersects with its previous call, and a local patch that makes the rewrite
check always reject. The query is `select item.a, item.c from t lateral view
explode(items) e as item`, with `items:
array<struct<a:bigint,b:bigint,c:bigint,s:int>>`:
| Builder | Reader | Result |
|---|---|---|
| intersecting | row-based | `[1,4], [10,40]` instead of `[1,3], [10,30]` |
| intersecting | vectorized | `NullPointerException` in the task |
| Spark's V2 Parquet | both | correct |
`item.c` keeps its ordinal in `struct<a,b,c,s>` but reads `struct<a,c,s>`,
so it gets `s`. Without the patch I found no query that makes the rewrite
reject. None of the 1817 tests in the touched suites gets there either. So this
is latent today, but the fallback is the safety net for the cases nobody has
found yet.
The plain case is not the only retry. The builder got 3 `pruneColumns` +
`build()` calls for that single `explode`. After the rejection, `transformDown`
reaches the `Generate`, where `GeneratorsOverScan` matches the shorter chain
and prunes the same builder again. Only then does the plain case run.
The fresh-builder proposal in
[issuecomment-6491505772](https://github.com/apache/spark/pull/59706#issuecomment-6491505772)
works for me, with these points:
- Stop after the first attempt that touched the holder's builder, and build
the plain scan with `rebuildScan` right there. If the case returns `op` and
lets `transformDown` descend, each lower `Generate` prunes the same builder
again.
- Skipping the generator path when `hasBlockingPushdown(sHolder)` is right,
since a fresh builder can't reproduce a pushed limit, offset, sample or sort.
- `rebuildScan` also needs `isSubquery`, which `pruneColumns` doesn't get
from `applyWithContext` yet.
- A test that reaches the rejection with a builder that prunes cumulatively,
as asked in [the earlier
review](https://github.com/apache/spark/pull/59706#pullrequestreview-5633214297),
would pin it.
--
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]
