This is an automated email from the ASF dual-hosted git repository.
gengliangwang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new c79834462b35 [SPARK-57915][SQL][TEST] Add a TPC-DS whole-stage-codegen
size benchmark
c79834462b35 is described below
commit c79834462b35855bb9b1e22829f0181cbb2b2241
Author: Gengliang Wang <[email protected]>
AuthorDate: Sat Jul 4 09:38:58 2026 -0700
[SPARK-57915][SQL][TEST] Add a TPC-DS whole-stage-codegen size benchmark
### What changes were proposed in this pull request?
Add `WholeStageCodegenSizeBenchmark`, a planning-only benchmark that
measures the size of the Java
code generated by whole-stage codegen across the TPC-DS queries. Unlike
`TPCDSQueryBenchmark`, it
does not execute the queries and needs no data: it creates the TPC-DS
tables empty (with SF=100 stats
injected so plan shapes are realistic), plans each query to its
`executedPlan`, walks the
`WholeStageCodegenExec` subtrees via the existing `debug.codegenStringSeq`,
and reports grand totals:
- source code size (chars of generated Java, summed over stages);
- max method bytecode, summed over stages, and the single max -- the metric
behind the HotSpot 8KB
JIT-compilation threshold and the 64KB method limit;
- number of generated inner classes;
- constant-pool entries (sum and max);
- total Janino compile time (ms), measured cold via
`CodeGenerator.resetCompileTime()`;
- codegen fallbacks (stages whose generated code failed to compile);
- whole-stage-codegen stage count and number of queries measured.
Results are written to
`sql/core/benchmarks/WholeStageCodegenSizeBenchmark-results.txt`.
### Why are the changes needed?
The umbrella
[SPARK-56908](https://issues.apache.org/jira/browse/SPARK-56908) tracks changes
that
reduce the size of generated whole-stage-codegen Java (smaller generated
methods reduce Janino
compilation work, JIT work, and pressure against the JVM 64KB method limit
/ HotSpot 8KB
JIT-compilation threshold). Those changes so far have been justified with
ad-hoc, out-of-tree
measurements. This adds a standard, reproducible, CI-runnable instrument so
any such change has a
consistent before/after size measurement over a realistic query set.
### Does this PR introduce _any_ user-facing change?
No. This is a test-only benchmark.
### How was this patch tested?
Ran `build/sbt "sql/Test/runMain
org.apache.spark.sql.execution.benchmark.WholeStageCodegenSizeBenchmark"`
locally: it plans all TPC-DS v1.4 + v2.7 queries with no data location and
prints the summary; the
committed results file was produced with
`SPARK_GENERATE_BENCHMARK_FILES=1`. Codegen fallbacks are 0
on master.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #56981 from gengliangwang/SPARK-57915-codegen-size-benchmark.
Authored-by: Gengliang Wang <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>
(cherry picked from commit bcad90da12663e31fc5d13e8469482af20d8ac81)
Signed-off-by: Gengliang Wang <[email protected]>
---
.../WholeStageCodegenSizeBenchmark-results.txt | 16 ++
.../benchmark/WholeStageCodegenSizeBenchmark.scala | 236 +++++++++++++++++++++
2 files changed, 252 insertions(+)
diff --git a/sql/core/benchmarks/WholeStageCodegenSizeBenchmark-results.txt
b/sql/core/benchmarks/WholeStageCodegenSizeBenchmark-results.txt
new file mode 100644
index 000000000000..7dc8f3ae5e46
--- /dev/null
+++ b/sql/core/benchmarks/WholeStageCodegenSizeBenchmark-results.txt
@@ -0,0 +1,16 @@
+================================================================================================
+TPCDS codegen size
+================================================================================================
+
+queries measured 135
+codegen stages 2247
+source code size (chars) 23794382
+max method bytecode (sum) 925512
+max method bytecode (max) 5843
+inner classes 516
+constant pool (sum) 440540
+constant pool (max) 770
+codegen fallbacks 0
+compile time (ms) 10618.3
+
+
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WholeStageCodegenSizeBenchmark.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WholeStageCodegenSizeBenchmark.scala
new file mode 100644
index 000000000000..0a9fa43ea397
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WholeStageCodegenSizeBenchmark.scala
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.benchmark
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.SparkConf
+import org.apache.spark.benchmark.Benchmark
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{SparkSession, TPCDSSchema, TPCDSTableStats}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.expressions.codegen.{ByteCodeStats,
CodeGenerator}
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.execution.debug._
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Benchmark to measure the size of the Java code generated by whole-stage
codegen for the TPC-DS
+ * queries. Unlike [[TPCDSQueryBenchmark]], this does not execute the queries
and needs no data: it
+ * plans each query over empty TPC-DS tables, collects the whole-stage-codegen
subtrees, and reports
+ * how large the generated (and compiled) code is. It is the standard
instrument for changes that
+ * aim to reduce generated Java size (see the umbrella SPARK-56908).
+ *
+ * The reported grand totals (lower is better, except the query/stage counts)
are:
+ * - source code size: characters of generated Java summed over all stages;
+ * - max method bytecode (sum): the per-stage largest compiled method size,
summed over stages --
+ * the metric behind the HotSpot 8KB JIT-compilation threshold and the
64KB method limit;
+ * - max method bytecode (max): the single largest compiled method across
all stages;
+ * - inner classes: number of generated inner classes summed over all stages;
+ * - constant pool (sum / max): compiled constant-pool entries, summed and
the per-stage max;
+ * - compile time: total Janino compilation time (ms), measured cold via
+ * [[CodeGenerator.resetCompileTime()]];
+ * - codegen fallbacks: stages whose generated code failed to compile (fell
back to interpreted);
+ * - stages / queries: the whole-stage-codegen stage count and the number of
queries measured.
+ *
+ * To run this benchmark:
+ * {{{
+ * 1. without sbt:
+ * bin/spark-submit --jars <spark core test jar>,<spark catalyst test
jar>
+ * --class <this class> <spark sql test jar>
+ * 2. build/sbt "sql/Test/runMain <this class>"
+ * 3. generate result:
+ * SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/Test/runMain <this
class>"
+ * Results will be written to
+ * "sql/core/benchmarks/WholeStageCodegenSizeBenchmark-results.txt".
+ * }}}
+ */
+object WholeStageCodegenSizeBenchmark extends SqlBasedBenchmark with Logging {
+
+ // Inject SF=100 table stats so join/aggregate strategy selection (and
therefore the shape of the
+ // generated code) resembles planning on real data, matching
PlanStabilitySuite's sf100 plans.
+ private val injectStats = true
+
+ override def getSparkSession: SparkSession = {
+ val conf = new SparkConf()
+ .setMaster(System.getProperty("spark.sql.test.master", "local[1]"))
+ .setAppName(this.getClass.getCanonicalName)
+ .set(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key, "true")
+ // Keep the codegen fallback path so we can count stages that fail to
compile, instead of
+ // failing the benchmark.
+ .set(SQLConf.CODEGEN_FALLBACK.key, "true")
+ // Disable AQE so the physical plan (and its whole-stage-codegen stages)
is fully materialized
+ // at planning time, without needing to execute the query to resolve
query stages.
+ .set(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key, "false")
+ .set(SQLConf.SHUFFLE_PARTITIONS.key, "4")
+ .set("spark.sql.crossJoin.enabled", "true")
+ .set("spark.sql.sources.default", "parquet")
+
+ val spark = SparkSession.builder().config(conf).getOrCreate()
+ if (injectStats) {
+ // Stats-based strategies (broadcast vs shuffle join, etc.) need the CBO
on to be applied.
+ spark.conf.set(SQLConf.CBO_ENABLED.key, "true")
+ spark.conf.set(SQLConf.PLAN_STATS_ENABLED.key, "true")
+ spark.conf.set(SQLConf.JOIN_REORDER_ENABLED.key, "true")
+ }
+ spark
+ }
+
+ /** Grand-total accumulators across all measured stages of all queries. */
+ private case class Totals(
+ var queries: Int = 0,
+ var stages: Int = 0,
+ var sourceChars: Long = 0L,
+ var maxMethodBytecodeSum: Long = 0L,
+ var maxMethodBytecodeMax: Long = 0L,
+ var innerClasses: Long = 0L,
+ var constPoolSum: Long = 0L,
+ var constPoolMax: Long = 0L,
+ var fallbacks: Int = 0)
+
+ private def createTables(): Unit = {
+ CodegenSizeTpcdsSchema.tableNames.foreach { tableName =>
+ spark.sql(
+ s"""
+ |CREATE TABLE `$tableName`
(${CodegenSizeTpcdsSchema.columnsDDL(tableName)})
+ |USING parquet
+ |${CodegenSizeTpcdsSchema.partitionedByClause(tableName)}
+ """.stripMargin)
+ if (injectStats) {
+ spark.sessionState.catalog.alterTableStats(
+ TableIdentifier(tableName),
Some(TPCDSTableStats.sf100TableStats(tableName)))
+ }
+ }
+ }
+
+ private def dropTables(): Unit = {
+ CodegenSizeTpcdsSchema.tableNames.foreach { tableName =>
+ spark.sessionState.catalog.dropTable(TableIdentifier(tableName),
ignoreIfNotExists = true,
+ purge = true)
+ }
+ }
+
+ /** Plans one query and folds its whole-stage-codegen size metrics into
`totals`. */
+ private def measureQuery(queryLocation: String, name: String, totals:
Totals): Unit = {
+ val queryString = resourceToString(s"$queryLocation/$name.sql",
+ classLoader = Thread.currentThread().getContextClassLoader)
+ try {
+ val plan = spark.sql(queryString).queryExecution.executedPlan
+ // codegenStringSeq walks all WholeStageCodegenExec subtrees (recursing
subqueries), calls
+ // doCodeGen() for the source and CodeGenerator.compile() for the
compiled ByteCodeStats.
+ val stages = codegenStringSeq(plan)
+ totals.queries += 1
+ stages.foreach { case (_, source, stats) =>
+ totals.stages += 1
+ totals.sourceChars += source.length
+ if (stats == ByteCodeStats.UNAVAILABLE) {
+ totals.fallbacks += 1
+ } else {
+ totals.maxMethodBytecodeSum += stats.maxMethodCodeSize
+ totals.maxMethodBytecodeMax =
+ math.max(totals.maxMethodBytecodeMax, stats.maxMethodCodeSize)
+ totals.innerClasses += stats.numInnerClasses
+ totals.constPoolSum += stats.maxConstPoolSize
+ totals.constPoolMax = math.max(totals.constPoolMax,
stats.maxConstPoolSize)
+ }
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(s"Skipping query $name: failed to plan (${e.getMessage})")
+ }
+ }
+
+ override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+ createTables()
+ try {
+ val queriesV1_4 = CodegenSizeTpcdsSchema.tpcdsQueries
+ val queriesV2_7 = CodegenSizeTpcdsSchema.tpcdsQueriesV2_7_0
+
+ // Measure compile time cold: reset the accumulator, let
codegenStringSeq trigger the Janino
+ // compiles (the compile cache is keyed on the source string, so
distinct queries miss), then
+ // read the total below.
+ CodeGenerator.resetCompileTime()
+ val totals = Totals()
+ queriesV1_4.foreach(measureQuery("tpcds", _, totals))
+ queriesV2_7.foreach(measureQuery("tpcds-v2.7.0", _, totals))
+ val compileTimeMs = CodeGenerator.compileTime / 1000000.0
+
+ runBenchmark("TPCDS codegen size") {
+ // Use a Benchmark instance only for its `out` PrintStream, which tees
to stdout and the
+ // results file; the metrics here are static sizes, not wall-clock
timings.
+ val benchmark = new Benchmark("TPCDS codegen size (grand totals, lower
is better)",
+ totals.queries.toLong, output = output)
+ val out = benchmark.out
+ // scalastyle:off println
+ def row(label: String, value: Any): Unit =
+ out.println("%-32s %s".format(label, value))
+ row("queries measured", totals.queries)
+ row("codegen stages", totals.stages)
+ row("source code size (chars)", totals.sourceChars)
+ row("max method bytecode (sum)", totals.maxMethodBytecodeSum)
+ row("max method bytecode (max)", totals.maxMethodBytecodeMax)
+ row("inner classes", totals.innerClasses)
+ row("constant pool (sum)", totals.constPoolSum)
+ row("constant pool (max)", totals.constPoolMax)
+ row("codegen fallbacks", totals.fallbacks)
+ row("compile time (ms)", "%.1f".format(compileTimeMs))
+ out.println()
+ // scalastyle:on println
+ }
+ } finally {
+ dropTables()
+ }
+ }
+}
+
+/**
+ * Exposes the protected TPC-DS schema/query metadata (from [[TPCDSSchema]])
that the benchmark
+ * object needs. The benchmark cannot mix in `TPCDSBase` because that trait
extends the
+ * ScalaTest-based `SharedSparkSession`.
+ */
+private object CodegenSizeTpcdsSchema extends TPCDSSchema {
+ val tableNames: Iterable[String] = tableColumns.keys
+
+ def columnsDDL(tableName: String): String = tableColumns(tableName)
+
+ def partitionedByClause(tableName: String): String =
+ tablePartitionColumns.get(tableName) match {
+ case Some(cols) if cols.nonEmpty => s"PARTITIONED BY (${cols.mkString(",
")})"
+ case _ => ""
+ }
+
+ // TPC-DS v1.4 queries (the same list as TPCDSBase, without the test-only
exclusions).
+ val tpcdsQueries: Seq[String] = Seq(
+ "q1", "q2", "q3", "q4", "q5", "q6", "q7", "q8", "q9", "q10", "q11",
+ "q12", "q13", "q14a", "q14b", "q15", "q16", "q17", "q18", "q19", "q20",
+ "q21", "q22", "q23a", "q23b", "q24a", "q24b", "q25", "q26", "q27", "q28",
"q29", "q30",
+ "q31", "q32", "q33", "q34", "q35", "q36", "q37", "q38", "q39a", "q39b",
"q40",
+ "q41", "q42", "q43", "q44", "q45", "q46", "q47", "q48", "q49", "q50",
+ "q51", "q52", "q53", "q54", "q55", "q56", "q57", "q58", "q59", "q60",
+ "q61", "q62", "q63", "q64", "q65", "q66", "q67", "q68", "q69", "q70",
+ "q71", "q72", "q73", "q74", "q75", "q76", "q77", "q78", "q79", "q80",
+ "q81", "q82", "q83", "q84", "q85", "q86", "q87", "q88", "q89", "q90",
+ "q91", "q92", "q93", "q94", "q95", "q96", "q97", "q98", "q99")
+
+ // TPC-DS v2.7 queries that differ from v1.4.
+ val tpcdsQueriesV2_7_0: Seq[String] = Seq(
+ "q5a", "q6", "q10a", "q11", "q12", "q14", "q14a", "q18a",
+ "q20", "q22", "q22a", "q24", "q27a", "q34", "q35", "q35a", "q36a", "q47",
"q49",
+ "q51a", "q57", "q64", "q67a", "q70a", "q72", "q74", "q75", "q77a", "q78",
+ "q80a", "q86a", "q98")
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
