Skip to content

test: report expression cost against a scan-and-transfer baseline [WIP / experimental] - #5375

Draft
andygrove wants to merge 4 commits into
apache:mainfrom
andygrove:bench-baseline-case
Draft

test: report expression cost against a scan-and-transfer baseline [WIP / experimental]#5375
andygrove wants to merge 4 commits into
apache:mainfrom
andygrove:bench-baseline-case

Conversation

@andygrove

Copy link
Copy Markdown
Member

Which issue does this PR close?

Part of #5363, covering checklist item 1. The epic stays open.

Stacked on #5371. That PR is the first two commits here; review and merge it first. This PR's own change is the third commit, test: report expression cost against a scan-and-transfer baseline.

Rationale for this change

Every timed case in runExpressionBenchmark is spark.sql("SELECT expr(c1) FROM t").noop(), so the timed region contains a Parquet scan and a columnar-to-row conversion in addition to the expression. The two arms use different readers, Spark's vectorized reader against Comet's native scan, so the contamination does not cancel: the reported ratio converges on the scan ratio for any expression cheaper than the scan. Nothing in the results file told a reader how much of the number was floor.

What changes are included in this PR?

Measure the floor and subtract it. For a single-table query the baseline is the optimized plan's leaf output, which is already column-pruned, projected straight back out. SELECT abs(c1) FROM parquetV1Table yields SELECT c1 FROM parquetV1Table: the same scan, the same columns, no expression work. The derived query is printed above each table so a reader can audit what the floor actually measured.

Cache it. The baseline is measured once per (query, config) pair. The 31 tables in CometStringExpressionBenchmark share one measurement per arm rather than adding 31, and the whole suite runs in about 4.5 minutes. Without caching, adding two cases per table would roughly double total benchmark runtime; CometCastNumericToNumericBenchmark alone would go from about 9 minutes to about 18.

Report through our own renderer. Benchmark.addCase/Benchmark.run renders from its own case list, accepts no precomputed Result, and has nowhere to put a derived column, so a cached baseline can never appear in its output. Measurement still uses Spark's warmup, iteration and standard deviation logic via the public Benchmark.measure; only the rendering is ours.

OpenJDK 64-Bit Server VM 17.0.17+10 on Mac OS X 26.3.1
Apple M3 Ultra
baseline: SELECT `c1` FROM parquetv1table
levenshtein_threshold (1024 rows):           Best(ms)   Stdev(ms)   Baseline(ms)    Expr(ms)  Relative(total)
--------------------------------------------------------------------------------------------------------------
Spark                                            10.4         1.3           12.8           -             1.0X
Comet                                             8.8         1.0            6.7         2.1             1.2X
[-] the expression cost is below the measurement floor

A difference smaller than the combined standard deviations, or a negative one, is reported as below the measurement floor rather than as a number, because a number invites a conclusion the data does not support. Queries reading more than one table, which is the 13 join cases, get no baseline and fall back to totals. The Relative header names which quantity the ratio is over, so a total ratio is never misread as an expression ratio.

BenchmarkTable.render is a pure function over Spark's Result case class, so it needs no SparkSession. That makes the column rules unit-testable, which matters here: nothing in this area was testable before, and #5371 could only be verified by running benchmarks and reading the output by hand.

Spark's Avg, Rate(M/s) and Per Row(ns) columns are dropped to make room. Avg is redundant with Best plus Stdev, and both rate columns are derivable from Best and the row count now carried in the title.

This diverges in output format from the 13 suites that call new Benchmark directly, such as CometReadBenchmark and the TPC suites. That seems right rather than unfortunate: the expression suites are precisely the ones the epic says report the wrong number, so they should report a different one.

How are these changes tested?

New BenchmarkTableSuite, 8 tests, running in CI and registered in both pr_build_linux.yml and pr_build_macos.yml:

  • expression cost is total minus baseline, with Relative computed on it
  • a missing baseline blanks the derived columns and falls back to Relative on total
  • when only one arm has an expression cost, every ratio is over the total
  • a negative expression cost is blanked, with the footnote
  • a difference within the combined standard deviations is blanked, and the same difference is reported once the measurement is tighter
  • the baseline query and the row count appear in the output

The third of those is a regression test. The first live run reported levenshtein as 0.6X under a Relative(total) header, which was Spark's total divided by Comet's expression cost. It now reads 0.4X.

Compiles under the spark-4.1, spark-3.5 and spark-3.4 profiles.

End to end, CometStringExpressionBenchmark and CometHashJoinBenchmark were run in full. The baseline is measured once per arm, confirmed by all 31 string tables reporting an identical baseline rather than 31 separately measured values. The join suite blanks the derived columns and prints the no-baseline footnote.

One substantive finding falls straight out of this, confirming the third bullet of Problem 2 in the epic: at 1024 rows, all 31 string expressions are below the measurement floor, with a Spark baseline of 12.8ms against totals in the 10-16ms range, and Comet's total frequently under its own baseline. That suite has been reporting noise. The harness now says so in the results file instead of printing a ratio. Raising the cardinality is item 5 of the epic and is left to a later PR.

`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 apache#5363.
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.
@andygrove andygrove changed the title test: report expression cost against a scan-and-transfer baseline test: report expression cost against a scan-and-transfer baseline [WIP / experimental] Aug 15, 2026
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.
Every timed case in `runExpressionBenchmark` is
`spark.sql("SELECT expr(c1) FROM t").noop()`, so the timed region contains a
Parquet scan and a columnar-to-row conversion as well as the expression. The
two arms use different readers, so the contamination does not cancel and the
reported ratio converges on the scan ratio for any expression cheaper than the
scan. Nothing in the results file said how much of the number was floor.

Measure that floor and subtract it. For a single-table query the baseline is
the optimized plan's leaf output, which is already column-pruned, projected
straight back out: `SELECT abs(c1) FROM t` yields `SELECT c1 FROM t`. The
baseline is measured once per (query, config) pair and cached, so the 31 tables
in `CometStringExpressionBenchmark` share one measurement per arm rather than
adding 31; without caching this would roughly double total benchmark runtime.

Reporting moves off `Benchmark.addCase`/`Benchmark.run`, which renders from its
own case list, accepts no precomputed `Result` and has nowhere to put a derived
column. Measurement still uses Spark's warmup, iteration and stdev logic
through the public `Benchmark.measure`. `BenchmarkTable.render` is pure, so the
column rules are unit-tested rather than eyeballed in a results file.

A difference smaller than the combined standard deviations, or a negative one,
is reported as below the measurement floor rather than as a number. Queries
reading more than one table get no baseline and fall back to totals, with the
header naming which quantity the ratio is over.

This immediately confirms the third bullet of Problem 2 in the epic: at 1024
rows every one of the 31 string expressions is below the measurement floor, and
Comet's total is often under its own baseline. The suite has been reporting
noise, and now says so.

Part of apache#5363.
@andygrove
andygrove force-pushed the bench-baseline-case branch from 730f490 to 7dd418c Compare August 15, 2026 22:48
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant