This is an automated email from the ASF dual-hosted git repository. cloud-fan pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git
commit 897e43402d18030805a3c9cb4f095ab572ef47e2 Author: Wenchen Fan <[email protected]> AuthorDate: Tue Sep 29 07:11:38 2026 +0000 Revert "[SPARK-58623][SQL] Extend ExpandBenchmark with CSE/rewrite matrix" This reverts commit 364fa9b3d576a0d3773053e51ec1b7321d0d085e. --- .../benchmarks/ExpandBenchmark-jdk21-results.txt | 81 +++++------- .../sql/execution/benchmark/ExpandBenchmark.scala | 144 +++++---------------- 2 files changed, 68 insertions(+), 157 deletions(-) diff --git a/sql/core/benchmarks/ExpandBenchmark-jdk21-results.txt b/sql/core/benchmarks/ExpandBenchmark-jdk21-results.txt index 9cc85852fafc..698d15902069 100644 --- a/sql/core/benchmarks/ExpandBenchmark-jdk21-results.txt +++ b/sql/core/benchmarks/ExpandBenchmark-jdk21-results.txt @@ -2,69 +2,56 @@ Expand: varying number of COUNT(DISTINCT) ================================================================================================ -OpenJDK 64-Bit Server VM 21.0.3+9-LTS on Mac OS X 26.5.2 -Apple M3 Pro +OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure +AMD EPYC 7763 64-Core Processor 2 distinct aggregates: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -with sum - baseline (ratio 3) 2518 2821 425 4.2 240.1 1.0X -with sum - optimized (ratio 3) 2363 2429 52 4.4 225.4 1.1X -pure distinct - baseline (ratio 2) 1382 1471 149 7.6 131.8 1.8X -pure distinct - optimized (ratio 2) 400 409 11 26.2 38.2 6.3X +with sum - baseline (ratio 3) 3150 3221 71 3.3 300.4 1.0X +with sum - optimized (ratio 3) 3095 3136 33 3.4 295.2 1.0X +pure distinct - baseline (ratio 2) 2038 2064 26 5.1 194.4 1.5X +pure distinct - optimized (ratio 2) 675 711 31 15.5 64.3 4.7X -OpenJDK 64-Bit Server VM 21.0.3+9-LTS on Mac OS X 26.5.2 -Apple M3 Pro +OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure +AMD EPYC 7763 64-Core Processor 4 distinct aggregates: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -with sum - baseline (ratio 5) 4569 4650 60 2.3 435.8 1.0X -with sum - optimized (ratio 5) 4584 4644 41 2.3 437.2 1.0X -pure distinct - baseline (ratio 4) 3393 3471 64 3.1 323.6 1.3X -pure distinct - optimized (ratio 4) 471 510 44 22.3 44.9 9.7X +with sum - baseline (ratio 5) 6006 6042 30 1.7 572.8 1.0X +with sum - optimized (ratio 5) 5981 6030 33 1.8 570.4 1.0X +pure distinct - baseline (ratio 4) 4540 4602 49 2.3 432.9 1.3X +pure distinct - optimized (ratio 4) 767 800 22 13.7 73.2 7.8X -OpenJDK 64-Bit Server VM 21.0.3+9-LTS on Mac OS X 26.5.2 -Apple M3 Pro +OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure +AMD EPYC 7763 64-Core Processor 6 distinct aggregates: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -with sum - baseline (ratio 7) 6311 6527 288 1.7 601.8 1.0X -with sum - optimized (ratio 7) 5659 5853 206 1.9 539.7 1.1X -pure distinct - baseline (ratio 6) 5145 5184 28 2.0 490.6 1.2X -pure distinct - optimized (ratio 6) 506 518 10 20.7 48.2 12.5X +with sum - baseline (ratio 7) 9223 9268 50 1.1 879.6 1.0X +with sum - optimized (ratio 7) 9253 9346 122 1.1 882.4 1.0X +pure distinct - baseline (ratio 6) 7420 7484 45 1.4 707.6 1.2X +pure distinct - optimized (ratio 6) 918 969 38 11.4 87.6 10.0X -OpenJDK 64-Bit Server VM 21.0.3+9-LTS on Mac OS X 26.5.2 -Apple M3 Pro +OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure +AMD EPYC 7763 64-Core Processor 8 distinct aggregates: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -with sum - baseline (ratio 9) 8226 8286 36 1.3 784.5 1.0X -with sum - optimized (ratio 9) 8218 8272 46 1.3 783.8 1.0X -pure distinct - baseline (ratio 8) 7086 7134 40 1.5 675.8 1.2X -pure distinct - optimized (ratio 8) 955 976 12 11.0 91.1 8.6X +with sum - baseline (ratio 9) 13382 13516 91 0.8 1276.2 1.0X +with sum - optimized (ratio 9) 13465 13538 108 0.8 1284.1 1.0X +pure distinct - baseline (ratio 8) 11329 11434 110 0.9 1080.4 1.2X +pure distinct - optimized (ratio 8) 1869 1920 37 5.6 178.3 7.2X ================================================================================================ Expand: varying data characteristics (pure distinct) ================================================================================================ -OpenJDK 64-Bit Server VM 21.0.3+9-LTS on Mac OS X 26.5.2 -Apple M3 Pro +OpenJDK 64-Bit Server VM 21.0.12+8-LTS on Linux 6.17.0-1020-azure +AMD EPYC 7763 64-Core Processor 6 pure distinct aggs with varying data: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -1K groups, moderate card - baseline 5175 5218 37 2.0 493.6 1.0X -1K groups, moderate card - optimized 497 507 10 21.1 47.4 10.4X -100K groups, moderate card - baseline 9323 9420 130 1.1 889.1 0.6X -100K groups, moderate card - optimized 1993 2069 139 5.3 190.1 2.6X -1K groups, low card (5 vals) - baseline 5277 5315 23 2.0 503.2 1.0X -1K groups, low card (5 vals) - optimized 466 474 8 22.5 44.5 11.1X -no grouping key - baseline 3577 3610 23 2.9 341.1 1.4X -no grouping key - optimized 218 221 5 48.2 20.8 23.8X - -================================================================================================ -Expand: subexpression elimination across branches -================================================================================================ - -OpenJDK 64-Bit Server VM 21.0.3+9-LTS on Mac OS X 26.5.2 -Apple M3 Pro -9 conditional COUNT(DISTINCT) + 9 conditional SUM sharing one subexpression: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------------------------------------------ -rewrite off, CSE off (10x amplify, 18 evals/row) 84450 84850 382 0.1 16107.6 1.0X -rewrite on, CSE off (2x amplify, 18 evals/row) 41354 41544 312 0.1 7887.7 2.0X -rewrite on, CSE on (2x amplify, 1 eval/row) 10712 10843 120 0.5 2043.2 7.9X -rewrite off, CSE on (10x amplify, 1 eval/row) 53281 54021 722 0.1 10162.6 1.6X +1K groups, moderate card - baseline 7437 7471 24 1.4 709.3 1.0X +1K groups, moderate card - optimized 898 922 18 11.7 85.7 8.3X +100K groups, moderate card - baseline 15063 15226 135 0.7 1436.5 0.5X +100K groups, moderate card - optimized 4454 4561 95 2.4 424.8 1.7X +1K groups, low card (5 vals) - baseline 7293 7342 40 1.4 695.5 1.0X +1K groups, low card (5 vals) - optimized 762 783 21 13.8 72.6 9.8X +no grouping key - baseline 4888 4948 39 2.1 466.1 1.5X +no grouping key - optimized 377 379 2 27.8 35.9 19.7X diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/ExpandBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/ExpandBenchmark.scala index 42644708ab5b..586dfec19620 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/ExpandBenchmark.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/ExpandBenchmark.scala @@ -33,22 +33,10 @@ import org.apache.spark.sql.internal.SQLConf * OptimizeExpand rule that pre-aggregates data before the Expand. Controlled by * spark.sql.optimizer.optimizeExpandRatio (default -1 = disabled). * - * It also measures two optimizations that target a shared expensive - * subexpression landing in the branches of the Expand, using a traffic/BI - * "N-day active users" rollup (conditional COUNT(DISTINCT) and conditional SUM - * aggregates whose conditions all share the same datetime subexpression): - * - rewriteCountDistinctConditional collapses N conditional COUNT(DISTINCT) - * on the same base column into one distinct group, shrinking the Expand - * fan-out from Nx to 1x (cuts data amplification, not the per-row - * subexpression evaluation cost); - * - subexpressionElimination (in Expand, via whole-stage codegen) evaluates - * the shared subexpression once per input row instead of once per Expand - * branch occurrence. - * The four cases form a 2x2 matrix over the two optimizations (base, +rewrite, - * +rewrite+CSE, and +CSE alone), so the benefit of each optimization can be - * attributed independently. Controlled by - * spark.sql.optimizer.rewriteCountDistinctConditional.enabled (default true) - * and spark.sql.subexpressionElimination.enabled (default true). + * It also measures subexpression elimination in Expand: conditional-aggregate + * queries whose conditions all share the same expensive subexpression, which + * ends up in every branch of the Expand. Controlled by + * spark.sql.subexpressionElimination.enabled (default true). * * To run this benchmark: * {{{ @@ -57,14 +45,6 @@ import org.apache.spark.sql.internal.SQLConf * SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/Test/runMain <this class>" * Results will be written to "benchmarks/ExpandBenchmark-results.txt". * }}} - * - * Optional arguments select benchmark sections (default: all): - * {{{ - * <this class> ratio # Expand: varying number of COUNT(DISTINCT) - * <this class> char # Expand: varying data characteristics (pure distinct) - * <this class> subexpr # Expand: subexpression elimination across branches - * <this class> smoke # all sections with tiny data and a single iteration - * }}} */ object ExpandBenchmark extends SqlBasedBenchmark { @@ -99,7 +79,7 @@ object ExpandBenchmark extends SqlBasedBenchmark { private def expandRatioBenchmark( title: String, N: Long, numDistinct: Int, - table: String, numIters: Int = 5): Unit = { + table: String): Unit = { val benchmark = new Benchmark(title, N, output = output) val sqlWithSum = countDistinctQuery( @@ -114,28 +94,28 @@ object ExpandBenchmark extends SqlBasedBenchmark { benchmark.addCase( s"with sum - baseline (ratio $ratioWithSum)", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "-1") { spark.sql(sqlWithSum).noop() } } benchmark.addCase( s"with sum - optimized (ratio $ratioWithSum)", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "2") { spark.sql(sqlWithSum).noop() } } benchmark.addCase( s"pure distinct - baseline (ratio $ratioPure)", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "-1") { spark.sql(sqlPure).noop() } } benchmark.addCase( s"pure distinct - optimized (ratio $ratioPure)", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "2") { spark.sql(sqlPure).noop() } @@ -163,9 +143,8 @@ object ExpandBenchmark extends SqlBasedBenchmark { * whose conditions all share the same expensive datetime subexpression * (parse `pt`, format it back, then `datediff` against a fixed date). * `RewriteDistinctAggregates` places those conditions in the branches of an - * Expand, so without subexpression elimination the shared subexpression is - * codegen'd once per occurrence: 9 in the distinct branch(es) and 9 in the - * SUM branch, i.e. 18 evaluations per input row. + * Expand, so the shared subexpression is codegen'd once per branch without + * subexpression elimination. */ private def conditionalAggsQuery(table: String, windows: Seq[Int]): String = { val daysSince = "datediff(" + @@ -182,63 +161,18 @@ object ExpandBenchmark extends SqlBasedBenchmark { |GROUP BY user_id""".stripMargin } - /** - * Measures a traffic/BI "N-day active users" rollup (see [[conditionalAggsQuery]]) - * under all four combinations of the two optimizations that affect the shared - * `daysSince` subexpression in the Expand branches: - * - base: rewriteCountDistinctConditional off, subexpression elimination off - * -> 10 Expand branches (one per conditional COUNT(DISTINCT) plus one for - * the SUMs), daysSince evaluated 18 times per input row; - * - +rewrite: rewriteCountDistinctConditional on, CSE off -> the 9 - * conditional COUNT(DISTINCT) expressions collapse into one distinct group, so the - * Expand has 2 branches (data amplification drops from 10x to 2x), but - * daysSince is still evaluated 18 times per input row; - * - +rewrite+CSE: both on -> the Expand runs the standard whole-stage - * subexpression elimination, evaluating daysSince once per input row - * before the branch loop; - * - +CSE alone: rewriteCountDistinctConditional off, subexpression - * elimination on -> the same 10-branch Expand as the base case, but - * daysSince is evaluated once per input row, isolating the CSE benefit - * without the data-amplification reduction. - */ private def subExprEliminationBenchmark( - title: String, N: Long, table: String, numIters: Int = 3): Unit = { + title: String, N: Long, table: String): Unit = { val benchmark = new Benchmark(title, N, output = output) val sql = conditionalAggsQuery(table, Seq(1, 7, 14, 30, 60, 90, 180, 365, 730)) - benchmark.addCase( - "rewrite off, CSE off (10x amplify, 18 evals/row)", - numIters = numIters) { _ => - withSQLConf( - SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "false", - SQLConf.REWRITE_COUNT_DISTINCT_CONDITIONAL_ENABLED.key -> "false") { + benchmark.addCase("CSE disabled (re-evaluate per branch)", numIters = 3) { _ => + withSQLConf(SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "false") { spark.sql(sql).noop() } } - benchmark.addCase( - "rewrite on, CSE off (2x amplify, 18 evals/row)", - numIters = numIters) { _ => - withSQLConf( - SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "false", - SQLConf.REWRITE_COUNT_DISTINCT_CONDITIONAL_ENABLED.key -> "true") { - spark.sql(sql).noop() - } - } - benchmark.addCase( - "rewrite on, CSE on (2x amplify, 1 eval/row)", - numIters = numIters) { _ => - withSQLConf( - SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "true", - SQLConf.REWRITE_COUNT_DISTINCT_CONDITIONAL_ENABLED.key -> "true") { - spark.sql(sql).noop() - } - } - benchmark.addCase( - "rewrite off, CSE on (10x amplify, 1 eval/row)", - numIters = numIters) { _ => - withSQLConf( - SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "true", - SQLConf.REWRITE_COUNT_DISTINCT_CONDITIONAL_ENABLED.key -> "false") { + benchmark.addCase("CSE enabled (evaluate once per row)", numIters = 3) { _ => + withSQLConf(SQLConf.SUBEXPRESSION_ELIMINATION_ENABLED.key -> "true") { spark.sql(sql).noop() } } @@ -247,33 +181,23 @@ object ExpandBenchmark extends SqlBasedBenchmark { } override def runBenchmarkSuite(mainArgs: Array[String]): Unit = { - val smoke = mainArgs.contains("smoke") - val sectionFilter = mainArgs.filterNot(_ == "smoke").toSet - - def runSection(key: String, name: String)(body: => Unit): Unit = { - if (sectionFilter.isEmpty || sectionFilter.contains(key)) { - runBenchmark(name)(body) - } - } - - val numIters = if (smoke) 1 else 5 - val N = if (smoke) 1000L else 10L << 20 // ~10M rows + val N = 10L << 20 // ~10M rows - runSection("ratio", "Expand: varying number of COUNT(DISTINCT)") { + runBenchmark("Expand: varying number of COUNT(DISTINCT)") { prepareTable("expand_bench", N, keyMod = 1000, colMods = Seq(100, 200, 300, 400, 500, 600, 700, 800)) expandRatioBenchmark( - "2 distinct aggregates", N, 2, "expand_bench", numIters) + "2 distinct aggregates", N, 2, "expand_bench") expandRatioBenchmark( - "4 distinct aggregates", N, 4, "expand_bench", numIters) + "4 distinct aggregates", N, 4, "expand_bench") expandRatioBenchmark( - "6 distinct aggregates", N, 6, "expand_bench", numIters) + "6 distinct aggregates", N, 6, "expand_bench") expandRatioBenchmark( - "8 distinct aggregates", N, 8, "expand_bench", numIters) + "8 distinct aggregates", N, 8, "expand_bench") } - runSection("char", "Expand: varying data characteristics (pure distinct)") { + runBenchmark("Expand: varying data characteristics (pure distinct)") { // Default: 1K groups, moderate distinct cardinality prepareTable("expand_default", N, keyMod = 1000, colMods = Seq(100, 200, 300, 400, 500, 600)) @@ -299,49 +223,49 @@ object ExpandBenchmark extends SqlBasedBenchmark { "6 pure distinct aggs with varying data", N, output = output) benchDataChar.addCase("1K groups, moderate card - baseline", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "-1") { spark.sql(sql6d).noop() } } benchDataChar.addCase("1K groups, moderate card - optimized", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "2") { spark.sql(sql6d).noop() } } benchDataChar.addCase("100K groups, moderate card - baseline", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "-1") { spark.sql(sql6dHighKey).noop() } } benchDataChar.addCase("100K groups, moderate card - optimized", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "2") { spark.sql(sql6dHighKey).noop() } } benchDataChar.addCase("1K groups, low card (5 vals) - baseline", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "-1") { spark.sql(sql6dLowCard).noop() } } benchDataChar.addCase("1K groups, low card (5 vals) - optimized", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "2") { spark.sql(sql6dLowCard).noop() } } benchDataChar.addCase("no grouping key - baseline", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "-1") { spark.sql(sql6dGlobal).noop() } } benchDataChar.addCase("no grouping key - optimized", - numIters = numIters) { _ => + numIters = 5) { _ => withSQLConf(OPTIMIZE_EXPAND_RATIO -> "2") { spark.sql(sql6dGlobal).noop() } @@ -350,12 +274,12 @@ object ExpandBenchmark extends SqlBasedBenchmark { benchDataChar.run() } - runSection("subexpr", "Expand: subexpression elimination across branches") { - val numRows = if (smoke) 1000L else 5L << 20 // ~5M rows + runBenchmark("Expand: subexpression elimination across branches") { + val numRows = 5L << 20 // ~5M rows prepareTrafficTable("expand_traffic", numRows, userMod = 1000000) subExprEliminationBenchmark( "9 conditional COUNT(DISTINCT) + 9 conditional SUM sharing one subexpression", - numRows, "expand_traffic", numIters = if (smoke) 1 else 3) + numRows, "expand_traffic") } } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
