Skip to content

Commit 4b60dca

Browse files
andygroveandygrove
authored
test: make expression benchmark harness fair and reproducible (#5371)
* test: make expression benchmark harness fair and reproducible `runExpressionBenchmark` backs 27 benchmark suites. Four defects made its output untrustworthy. Apply `spark.sql.optimizer.excludedRules` to both arms rather than the Comet arm only, and append to the existing value instead of overwriting it. Excluding `ConstantFolding` for Comet alone let Spark fold expressions over literal arguments away entirely: for `select space(2) from parquetV1Table` the Spark plan was `Project [ AS space(2)]`, doing no per-row work, so that row reported a Comet regression that does not exist. Warn when the Spark baseline plan does no work beyond scanning and projecting bare attributes or literals, which means the optimizer removed the benchmarked expression. Any other node counts as work, so the join and aggregate suites are unaffected. Route both that warning and the existing not-fully-native warning to `output`, so they land in the results file above the affected table rather than scrolling past on the console. Running the suites with this in place shows `translate` in `CometStringExpressionBenchmark` and eight `c_short` cases in `CometCastNumericToNumericBenchmark` falling back to a JVM `Project`, meaning those rows labelled "Comet" were measuring Spark. Generate the base table from a pure function of the row id instead of an unseeded `scala.util.Random`, so runs are comparable. Seeding a driver-side `Random` would not have worked: the closure runs per row on the executor. Part of #5363. * refactor: simplify benchmark harness plan checks and warning output Write warnings through `Benchmark.out`, which already tees the console and the results file, instead of hand-rolling the tee. This also keeps the warning ordered against the results table, since both now go through the same stream. Collapse `isTrivialPlan` to a single negated `exists` over the plan, dropping the mid-method `return`, the second traversal and the one-use `unwrapAlias`. Drop the untimed `noop()` from the Spark baseline check. Before execution `stripAQEPlan` already yields the initial physical plan, which carries the projections the check inspects; the nodes missing pre-execution are the codegen and columnar-transition wrappers the check classifies as no-work anyway. The Comet check keeps its execution, because AQE can re-plan a join after the first stage completes. Use `ConstantFolding.ruleName` rather than a hardcoded class name, and `Utils.stringToSeq` rather than an open-coded comma-list parser. * test: benchmark space over a column, and flag row-invariant projections Excluding `ConstantFolding` on both arms made the `space(2)` plans symmetric but not the work. Given a literal, Comet's native `space` receives a `ColumnarValue::Scalar`, builds one string per batch and lets DataFusion broadcast it, while Spark's `StringSpace` calls `UTF8String.blankString` once per row. Benchmark `space(c2)` over a new non-negative integer column instead, so both engines evaluate the expression on every row. The plan check could not catch this, because the retained `ProjectExec` looks like real work either way. Add a second check for projections that reference no input column: their value is the same for every row, so an engine is free to evaluate them once per batch, and engines differ on whether they do. Nondeterministic expressions are excluded, since `rand()` also references no column but genuinely evaluates per row. Verified that the old query trips the new warning, that the new one does not, and that `floor(rand() * 100)` does not. * fix: drop unused string interpolator flagged by scalafix
1 parent a74839c commit 4b60dca

2 files changed

Lines changed: 145 additions & 32 deletions

File tree

‎spark/src/test/scala/org/apache/spark/sql/benchmark/CometBenchmarkBase.scala‎

Lines changed: 134 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -23,15 +23,17 @@ import java.io.File
2323
import java.nio.charset.StandardCharsets
2424
import java.util.Base64
2525

26-
import scala.util.Random
27-
2826
import org.apache.parquet.crypto.DecryptionPropertiesFactory
2927
import org.apache.parquet.crypto.keytools.{KeyToolkit, PropertiesDrivenCryptoFactory}
3028
import org.apache.parquet.crypto.keytools.mocks.InMemoryKMS
3129
import org.apache.spark.SparkConf
3230
import org.apache.spark.benchmark.Benchmark
3331
import org.apache.spark.sql.{DataFrame, DataFrameWriter, Row, SparkSession}
32+
import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, Expression, Literal}
33+
import org.apache.spark.sql.catalyst.optimizer.ConstantFolding
3434
import org.apache.spark.sql.comet.CometPlanChecker
35+
import org.apache.spark.sql.comet.util.Utils
36+
import org.apache.spark.sql.execution.{ColumnarToRowExec, InputAdapter, LeafExecNode, ProjectExec, SparkPlan, WholeStageCodegenExec}
3537
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
3638
import org.apache.spark.sql.execution.benchmark.SqlBasedBenchmark
3739
import org.apache.spark.sql.internal.SQLConf
@@ -97,9 +99,13 @@ trait CometBenchmarkBase
9799
useDictionary: Boolean = false)(f: Int => Any): Unit = {
98100
withTempTable(tbl) {
99101
import spark.implicits._
102+
// Derive the values from the row id with a pure function rather than a random number
103+
// generator, so that results are comparable across runs. Seeding a driver-side `Random`
104+
// would not work here: the closure runs per row on the executor.
100105
spark
101106
.range(values)
102-
.map(_ => if (useDictionary) Random.nextLong() % 5 else Random.nextLong())
107+
.map(i =>
108+
if (useDictionary) CometBenchmarkBase.mix64(i) % 5 else CometBenchmarkBase.mix64(i))
103109
.createOrReplaceTempView(tbl)
104110
runBenchmark(benchmarkName)(f(values))
105111
}
@@ -125,50 +131,132 @@ trait CometBenchmarkBase
125131
extraCometConfigs: Map[String, String] = Map.empty): Unit = {
126132
val benchmark = new Benchmark(name, cardinality, output = output)
127133

128-
benchmark.addCase("Spark") { _ =>
129-
withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
130-
spark.sql(query).noop()
131-
}
132-
}
134+
// Constant folding is excluded so that expressions over literal arguments are still evaluated
135+
// per row. It must be excluded for both arms: if only Comet excludes it, Spark folds the
136+
// expression away and does no per-row work, and the comparison is meaningless.
137+
val noConstantFolding =
138+
SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> excludedRulesWith(ConstantFolding.ruleName)
133139

134-
val cometExecConfigs = Map(
140+
val sparkConfigs = Seq(noConstantFolding, CometConf.COMET_ENABLED.key -> "false")
141+
142+
val cometConfigs = Seq(
143+
noConstantFolding,
135144
CometConf.COMET_ENABLED.key -> "true",
136-
CometConf.COMET_EXEC_ENABLED.key -> "true",
137-
"spark.sql.optimizer.excludedRules" ->
138-
"org.apache.spark.sql.catalyst.optimizer.ConstantFolding") ++ extraCometConfigs
145+
CometConf.COMET_EXEC_ENABLED.key -> "true") ++ extraCometConfigs
146+
147+
// Check that the benchmarked expression survives optimization into the Spark baseline plan.
148+
// The query is not executed: before execution `stripAQEPlan` yields the initial physical
149+
// plan, which already carries the projections this check inspects.
150+
withSQLConf(sparkConfigs: _*) {
151+
val plan = stripAQEPlan(spark.sql(query).queryExecution.executedPlan)
152+
if (isTrivialPlan(plan)) {
153+
warn(
154+
benchmark,
155+
"""WARNING: the Spark baseline plan does no work beyond scanning and projecting
156+
|columns, so this case is not measuring an expression. The expression was most
157+
|likely removed by the optimizer.
158+
|Spark plan:""".stripMargin + "\n" + plan.treeString)
159+
}
160+
val rowInvariant = rowInvariantProjections(plan)
161+
if (rowInvariant.nonEmpty) {
162+
warn(
163+
benchmark,
164+
s"""WARNING: these projections reference no input column, so their value is the same
165+
|for every row: ${rowInvariant.mkString(", ")}
166+
|An engine may evaluate them once per batch and broadcast the result rather than
167+
|per row, so the two arms are not necessarily doing comparable work. Benchmark the
168+
|expression over a column instead.""".stripMargin)
169+
}
170+
}
139171

140-
// Check that the plan is fully Comet native before running the benchmark
141-
withSQLConf(cometExecConfigs.toSeq: _*) {
172+
// Check that the plan is fully Comet native before running the benchmark. Unlike the check
173+
// above this one does execute the query, because AQE can re-plan a join after the first
174+
// stage completes and the Comet operators are what we are checking for.
175+
withSQLConf(cometConfigs: _*) {
142176
val df = spark.sql(query)
143177
df.noop()
144178
val plan = stripAQEPlan(df.queryExecution.executedPlan)
145-
findFirstNonCometOperator(plan) match {
146-
case Some(op) =>
147-
// scalastyle:off println
148-
println()
149-
println("=" * 80)
150-
println("WARNING: Benchmark plan is NOT fully Comet native!")
151-
println(s"First non-Comet operator: ${op.nodeName}")
152-
println("=" * 80)
153-
println("Query plan:")
154-
println(plan.treeString)
155-
println("=" * 80)
156-
println()
157-
// scalastyle:on println
158-
case None =>
159-
// All operators are Comet native, no warning needed
179+
findFirstNonCometOperator(plan).foreach { op =>
180+
warn(
181+
benchmark,
182+
s"""WARNING: the Comet plan is NOT fully Comet native, so the Comet case below is
183+
|partly or wholly measuring Spark.
184+
|First non-Comet operator: ${op.nodeName}
185+
|Comet plan:""".stripMargin + "\n" + plan.treeString)
186+
}
187+
}
188+
189+
benchmark.addCase("Spark") { _ =>
190+
withSQLConf(sparkConfigs: _*) {
191+
spark.sql(query).noop()
160192
}
161193
}
162194

163195
benchmark.addCase("Comet") { _ =>
164-
withSQLConf(cometExecConfigs.toSeq: _*) {
196+
withSQLConf(cometConfigs: _*) {
165197
spark.sql(query).noop()
166198
}
167199
}
168200

169201
benchmark.run()
170202
}
171203

204+
/**
205+
* Returns the value of `spark.sql.optimizer.excludedRules` with `rule` appended, so that
206+
* benchmark-specific exclusions do not clobber exclusions already configured by the caller.
207+
*/
208+
private def excludedRulesWith(rule: String): String =
209+
(Utils.stringToSeq(spark.conf.get(SQLConf.OPTIMIZER_EXCLUDED_RULES.key, "")) :+ rule).distinct
210+
.mkString(",")
211+
212+
/**
213+
* Returns true if the plan does nothing but scan and emit columns, which means the benchmarked
214+
* expression was removed by the optimizer and the case measures nothing.
215+
*
216+
* Any node other than a scan, projection or row conversion counts as real work, so joins,
217+
* aggregates and filters are never reported as trivial even when they project bare attributes.
218+
*/
219+
private def isTrivialPlan(plan: SparkPlan): Boolean = !plan.exists {
220+
case p: ProjectExec => p.projectList.exists(expr => !isTrivialExpr(expr))
221+
case _: ColumnarToRowExec | _: WholeStageCodegenExec | _: InputAdapter | _: LeafExecNode =>
222+
false
223+
case _ => true
224+
}
225+
226+
/** A projection that copies a column or emits a constant, doing no per-row work. */
227+
private def isTrivialExpr(expr: Expression): Boolean = expr match {
228+
case _: AttributeReference | _: Literal => true
229+
case Alias(_: AttributeReference | _: Literal, _) => true
230+
case _ => false
231+
}
232+
233+
/**
234+
* Projections whose value is the same for every row, because they reference no input column.
235+
*
236+
* An engine is free to evaluate these once per batch and broadcast the result, and engines
237+
* differ on whether they do. Comet's native `space` takes a `ColumnarValue::Scalar` and builds
238+
* one string per batch, while Spark's `StringSpace` calls `UTF8String.blankString` per row. The
239+
* two plans look identical, so nothing else here can tell them apart.
240+
*
241+
* Nondeterministic expressions are excluded: `rand()` also references no column but genuinely
242+
* evaluates per row.
243+
*/
244+
private def rowInvariantProjections(plan: SparkPlan): Seq[Expression] =
245+
plan
246+
.collect { case p: ProjectExec => p }
247+
.flatMap(_.projectList)
248+
.filter(expr => expr.references.isEmpty && expr.deterministic && !isTrivialExpr(expr))
249+
250+
/**
251+
* Writes a warning to the benchmark results file as well as the console. `Benchmark.out` tees
252+
* the two, and writing through it keeps the warning ordered against the results table that
253+
* `Benchmark.run` writes to the same stream.
254+
*/
255+
private def warn(benchmark: Benchmark, message: String): Unit = {
256+
val border = "=" * 80
257+
benchmark.out.println(s"\n$border\n$message\n$border")
258+
}
259+
172260
protected def addParquetScanCases(
173261
benchmark: Benchmark,
174262
query: String,
@@ -291,3 +379,19 @@ trait CometBenchmarkBase
291379
.select((($"value" - 500) / 100.0) cast decimal as Symbol("dec"))
292380
}
293381
}
382+
383+
object CometBenchmarkBase {
384+
385+
/**
386+
* SplitMix64 finalizer, used to turn a row id into a well distributed pseudo-random value.
387+
*
388+
* This lives in the companion object rather than the trait because referencing a trait method
389+
* from a Dataset closure captures `this`, which is not serializable.
390+
*/
391+
def mix64(value: Long): Long = {
392+
var z = value + 0x9e3779b97f4a7c15L
393+
z = (z ^ (z >>> 30)) * 0xbf58476d1ce4e5b9L
394+
z = (z ^ (z >>> 27)) * 0x94d049bb133111ebL
395+
z ^ (z >>> 31)
396+
}
397+
}

‎spark/src/test/scala/org/apache/spark/sql/benchmark/CometStringExpressionBenchmark.scala‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,11 @@ object CometStringExpressionBenchmark extends CometBenchmarkBase {
7575
StringExprConfig("rlike", "select c1 rlike '[0-9]+' from parquetV1Table"),
7676
StringExprConfig("rpad", "select rpad(c1, 150, 'x') from parquetV1Table"),
7777
StringExprConfig("rtrim", "select rtrim(c1) from parquetV1Table"),
78-
StringExprConfig("space", "select space(2) from parquetV1Table"),
78+
// `space` takes its length from a column rather than a literal. Given a literal, Comet's
79+
// native `space` receives a `ColumnarValue::Scalar`, builds one string per batch and lets
80+
// DataFusion broadcast it, while Spark's `StringSpace` calls `UTF8String.blankString` once
81+
// per row. The plans look symmetric but the work is not.
82+
StringExprConfig("space", "select space(c2) from parquetV1Table"),
7983
StringExprConfig("startswith", "select startswith(c1, '1') from parquetV1Table"),
8084
StringExprConfig("substring", "select substring(c1, 1, 100) from parquetV1Table"),
8185
StringExprConfig("translate", "select translate(c1, '123456', 'aBcDeF') from parquetV1Table"),
@@ -86,9 +90,14 @@ object CometStringExpressionBenchmark extends CometBenchmarkBase {
8690
runBenchmarkWithTable("String expressions", 1024) { v =>
8791
withTempPath { dir =>
8892
withTempTable("parquetV1Table") {
93+
// c2 gives expressions that take a length or a count something to vary over per row.
94+
// `pmod` keeps it non-negative, so `space(c2)` builds a string on every row rather
95+
// than returning empty for the negative half of the input.
8996
prepareTable(
9097
dir,
91-
spark.sql(s"SELECT REPEAT(CAST(value AS STRING), 10) AS c1 FROM $tbl"))
98+
spark.sql(
99+
"SELECT REPEAT(CAST(value AS STRING), 10) AS c1," +
100+
s" CAST(PMOD(value, 200) AS INT) AS c2 FROM $tbl"))
92101

93102
val extraConfigs = Map(
94103
CometConf.getExprAllowIncompatConfigKey("Upper") -> "true",

0 commit comments

Comments
 (0)