ohad-definity commented on code in PR #59706:
URL: https://github.com/apache/spark/pull/59706#discussion_r4431659717
##########
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:
Thanks, removed in 75983e45: the ORC benchmarks skip the Explode cases while
ORC is excluded, and the six ORC result files are back to master's. The Parquet
results are unchanged and come from this path.
##########
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:
Thanks, added in 0ddaf12b next to the rejection tests, using your probe:
under ANSI, `rand(3) > 0`, then `k or rand(7) < 0.000001` (NULL where `s =
'bad'`), then `cast(s as int) > 0`, below an outer explode, with the rewrite
accepted and forced to reject. It returns `(7,21)` on the current code, and
with only this line set back to `reduceOption(And)` the forced rejection throws
`CAST_INVALID_INPUT` (checked on a mutated copy).
--
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]
