sunchao commented on code in PR #59706:
URL: https://github.com/apache/spark/pull/59706#discussion_r4426192762
##########
sql/core/benchmarks/OrcNestedSchemaPruningBenchmark-jdk21-results.txt:
##########
@@ -2,52 +2,61 @@
Nested Schema Pruning Benchmark For ORC v1
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.13.0.3+1-LTS on Linux 7.0.0-1017-azure
+AMD EPYC 9V45 96-Core Processor
Selection: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Top-level column 62 76
11 16.0 62.4 1.0X
-Nested column 61 72
7 16.4 60.8 1.0X
-Nested column in array 180 189
8 5.6 180.0 0.3X
+Top-level column 35 45
4 28.4 35.2 1.0X
+Nested column 36 42
8 27.5 36.3 1.0X
+Nested column in array 90 99
12 11.1 90.5 0.4X
-OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.13.0.3+1-LTS on Linux 7.0.0-1017-azure
+AMD EPYC 9V45 96-Core Processor
Limiting: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Top-level column 259 271
10 3.9 258.6 1.0X
-Nested column 254 277
18 3.9 254.5 1.0X
-Nested column in array 554 577
15 1.8 553.9 0.5X
+Top-level column 168 183
19 5.9 168.1 1.0X
+Nested column 175 192
17 5.7 175.0 1.0X
+Nested column in array 377 392
13 2.7 377.0 0.4X
-OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.13.0.3+1-LTS on Linux 7.0.0-1017-azure
+AMD EPYC 9V45 96-Core Processor
Repartitioning: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Top-level column 222 230
7 4.5 222.4 1.0X
-Nested column 226 237
8 4.4 226.4 1.0X
-Nested column in array 504 509
3 2.0 503.6 0.4X
+Top-level column 156 162
5 6.4 155.7 1.0X
+Nested column 155 165
5 6.4 155.5 1.0X
+Nested column in array 321 325
3 3.1 320.6 0.5X
-OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.13.0.3+1-LTS on Linux 7.0.0-1017-azure
+AMD EPYC 9V45 96-Core Processor
Repartitioning by exprs: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Top-level column 219 228
8 4.6 219.3 1.0X
-Nested column 246 255
7 4.1 246.2 0.9X
-Nested column in array 532 541
4 1.9 531.9 0.4X
+Top-level column 150 155
5 6.7 149.8 1.0X
+Nested column 164 168
3 6.1 164.3 0.9X
+Nested column in array 317 324
11 3.2 316.9 0.5X
-OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.13.0.3+1-LTS on Linux 7.0.0-1017-azure
+AMD EPYC 9V45 96-Core Processor
Sample: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Top-level column 75 93
22 13.3 75.1 1.0X
-Nested column 79 89
8 12.6 79.5 0.9X
-Nested column in array 237 265
21 4.2 236.6 0.3X
+Top-level column 49 60
9 20.4 49.0 1.0X
+Nested column 55 65
6 18.1 55.4 0.9X
+Nested column in array 122 136
8 8.2 121.5 0.4X
-OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure
-AMD EPYC 7763 64-Core Processor
+OpenJDK 64-Bit Server VM 21.0.13.0.3+1-LTS on Linux 7.0.0-1017-azure
+AMD EPYC 9V45 96-Core Processor
Sorting: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Top-level column 304 320
19 3.3 304.1 1.0X
-Nested column 349 362
13 2.9 348.6 0.9X
-Nested column in array 753 791
37 1.3 753.1 0.4X
+Top-level column 206 217
16 4.8 206.4 1.0X
+Nested column 243 253
13 4.1 242.6 0.9X
+Nested column in array 445 462
17 2.2 445.4 0.5X
+
+OpenJDK 64-Bit Server VM 21.0.13.0.3+1-LTS on Linux 7.0.0-1017-azure
+AMD EPYC 9V45 96-Core Processor
+Explode: Best
Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+---------------------------------------------------------------------------------------------------------------------------------------------------
+Two nested fields of an exploded array, through generate false
544 579 25 9.2 108.8 1.0X
+Two nested fields of an exploded array, through generate true
173 180 7 28.9 34.6 3.1X
Review Comment:
[P3] Refresh the ORC results after excluding ORC from this optimization
These rows still report 544 ms -> 173 ms when enabling through-generator
pruning, but this revision excludes `OrcFileFormat` and `OrcTable` from that
path. The benchmark uses those readers and toggles only this flag, so the saved
improvement comes from the superseded implementation. The same mismatch affects
all six ORC V1/V2 result files across JDK 17/21/25. Please regenerate those
sections, remove them, or label them with their historical revision so they do
not imply a benefit available in the code being merged. This is a
benchmark-evidence follow-up; the PR description already correctly states that
ORC is excluded.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala:
##########
@@ -1015,8 +1017,60 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan]
with PredicateHelper {
Project(rewrittenProjectList, withFilter)
}
- def pruneColumns(plan: LogicalPlan): LogicalPlan = plan.transform {
+ def pruneColumns(plan: LogicalPlan, isSubquery: Boolean): LogicalPlan =
plan.transform {
+ case op @ SchemaPruning.GeneratorsOverScan(segments, generators, sHolder:
ScanBuilderHolder)
+ if !hasBlockingPushdown(sHolder) &&
!sHolder.relation.table.isInstanceOf[OrcTable] =>
+ // ORC is not pruned through generators yet, see
`SchemaPruning.GeneratorsOverScan`.
+ // Fields read from the generators' elements above them narrow the scan
below them. This
+ // prunes and builds the holder's builder before the plan above is
re-resolved. If that is
+ // then rejected, the builder must not be pruned again with a wider
schema (a builder may
+ // only narrow, e.g. Iceberg's), so the plain scan is built from a fresh
builder instead, as
+ // scan merging does, and the lower generators of the chain do not try
again.
+ var builderUsed = false
+ SchemaPruning.pruneThroughGenerators(op, sHolder) { (scan,
otherProjects, inputRootFields) =>
+ builderUsed = true
+ Some(pruneScan(scan.projects, scan.filtersStayUp,
scan.filtersPushDown, sHolder,
+ Some(otherProjects), inputRootFields))
+ }.getOrElse {
+ if (builderUsed) {
+ val bottom = generators.last.child
+ val rebuilt = rebuildWithoutGenerators(segments.last, sHolder,
isSubquery)
+ op.transformDown { case p if p eq bottom => rebuilt }
+ } else {
+ op
+ }
+ }
+
case ScanOperation(project, filtersStayUp, filtersPushDown, sHolder:
ScanBuilderHolder) =>
+ pruneScan(project, filtersStayUp, filtersPushDown, sHolder, None, Nil)
+ }
+
+ /**
+ * Plans `scan`, the projects and filters over `sHolder` below a chain of
generators, from a
+ * fresh `ScanBuilder` of the same relation, like [[rebuildScan]]: the
filters pushed to
+ * `sHolder` and the pushable ones of `scan` are offered to the new builder
together, directly
+ * over the relation. Only valid when `sHolder` has no pushdown such a
rebuild cannot reproduce.
+ */
+ private def rebuildWithoutGenerators(
+ scan: SchemaPruning.Segment,
+ sHolder: ScanBuilderHolder,
+ isSubquery: Boolean): LogicalPlan = {
+ val relation = sHolder.relation
+ val pushable = (sHolder.pushedFilterExpressions ++
scan.filtersPushDown).reduceOption(And)
+ .map(Filter(_, relation)).getOrElse(relation)
+ // one Filter per condition, in order: combining them could evaluate a
condition that an
+ // earlier one would have skipped (e.g. a cast that fails under ANSI)
+ val filtered = scan.filtersStayUp.foldLeft(pushable)((plan, cond) =>
Filter(cond, plan))
Review Comment:
[P3] Add regression coverage for the separate fallback filters
The fix passes the guarded ANSI-cast probe, but neither committed
builder-suite test exercises this evaluation-order invariant. A lower
nondeterministic filter that rejects a NULL condition must prevent an upper
`cast(s AS INT)` from seeing `s = 'bad'`; combining the filters with `And` can
evaluate that cast instead. With current-source production classes,
feature-disabled, accepted, and forced-rejected paths all return `(7,21)`. In
an isolated copy restoring only this line to `reduceOption(And)`, forced
rejection again throws `CAST_INVALID_INPUT` while the other paths pass. Please
add this case alongside the rejection tests. The strengthened partitioned
predicate fixture now catches lost pushdown, but its simple predicates cannot
protect this distinct fix.
--
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]
