apache / apache/datafusion-comet
[EPIC] Reimplement Scala expression microbenchmarks so they measure expression cost, not scan and result transfer
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
## Background
We now have ~25 Scala end-to-end microbenchmark suites for expressions, almost all of which funnel through `runExpressionBenchmark` in `spark/src/test/scala/org/apache/spark/sql/benchmark/CometBenchmarkBase.scala`. I reviewed them to check whether they actually isolate expression evaluation cost. They largely do not — a large and non-uniform share of each timed region is Parquet scan plus columnar-to-row conversion of the result set, and there are several fairness bugs that make individual rows misleading.
These benchmarks are what we use to justify optimization work and to catch regressions, so it is worth reimplementing the harness rather than patching individual suites.
## Problem 1: scan and result transfer are inside the timed region
Every timed case is:
```scala
spark.sql("SELECT expr(c1) FROM parquetV1Table").noop()
```
Two costs that are not expression evaluation are inside the timer:
**Parquet scan.** Read on every iteration, and critically the two arms use *different* readers — Spark's vectorized reader vs. Comet's native scan. So the contamination does not cancel. The `relative` column is a ratio of sums, `(scan_spark + expr_spark + c2r_spark) / (scan_comet + expr_comet + c2r_comet)`, which for any expression cheaper than the scan converges to the *scan* ratio rather than the expression ratio.
**Result transfer + columnar-to-row.** The `noop` sink writer is a `DataWriter[InternalRow]`, so a `ColumnarToRowExec` sits on top of every plan. This is known and tolerated — `CometPlanChecker.findFirstNonCometOperator` explicitly whitelists `ColumnarToRowExec`, `CometColumnarToRowExec`, `CometNativeColumnarToRowExec` and `WholeStageCodegenExec`. Since `spark.comet.exec.columnarToRow.native.enabled` defaults to `false`, the Comet arm pays Arrow FFI export to JVM `CometVector` to `UnsafeRow`, while Spark emits rows directly from a single fused codegen stage.
The second cost tracks the *output* type of the expression, so contamination varies a lot within a single suite:
| Output type | Affected suites | c2r cost |
| --- | --- | --- |
| int / bool / long | comparison, hash, `length`/`ascii`/`instr`, `like`/`rlike`, `unix_timestamp`, cast-to-numeric | small |
| String | all cast-to-string suites, most of `CometStringExpressionBenchmark`, `to_json`, `to_csv`, `get_json_object`, string-result `CASE WHEN` | large |
| struct / array | `from_json`, `sort_array`, `make_interval` | dominant |
So cross-expression rankings within a single results file are distorted too, not only the absolute numbers.
Worst case is `CometArrayExpressionBenchmark`: 4M rows of 16- and 32-element int arrays through `.noop()`, where building `UnsafeArrayData` per row very likely costs more than `sort_array` itself. The suite's own `element_at(sort_array(x), 1)` variant collapses the output to a single int, and the gap between those two rows is essentially the c2r cost.
`CometAggregateExpressionBenchmark` is the one suite that is clean on this axis, because aggregate queries emit one row per group and nothing crosses the boundary.
## Problem 2: fairness bugs
**ConstantFolding is excluded only for the Comet arm.** `cometExecConfigs` sets `spark.sql.optimizer.excludedRules` to `ConstantFolding`; the Spark case runs with folding enabled. Concrete casualty: `select space(2) from parquetV1Table` in `CometStringExpressionBenchmark` — Spark folds it to a literal and does no per-row work while Comet evaluates it per row, so that row reports a fabricated Comet regression. The conf is also *set* rather than appended, clobbering any pre-existing exclusions.
**`CometPredicateExpressionBenchmark` measures Parquet filter pushdown, not `In`.** The query is `select * from parquetV1Table where c1 in ('positive','zero')` over a column with three distinct values. Both engines push that into the reader, so the `In` expression may never be evaluated as an expression at all. `CometComparisonExpressionBenchmark` gets this right by putting predicates in the SELECT list instead.
**Two suites use a cardinality too low to measure anything.** `CometStringExpressionBenchmark` and `CometRegExpBenchmark` both use 1024 rows — a single batch. Spark's `Benchmark` runs 2s of warmup and then at least 2s of timed iterations, each iteration being a full Spark job, so these are measuring job submission and Comet's per-query native plan construction rather than expression throughput. The `relative` column will read approximately 1.0x regardless of the expression. That is 31 string expressions and 5 regex patterns currently producing noise.
## Problem 3: uncontrolled inputs
**Parquet dictionary encoding is left at default in every expression suite**, while `CometColumnarToRowBenchmark`, `CometExecBenchmark` and `CometShuffleBenchmark` all set `parquet.enable.dictionary=false` for exactly this reason. The expression suites diverged from an existing convention in the same package. This matters because kernels with a dictionary fast path only compute over distinct values, and distinct counts happen to range from 3 (`CometPredicateExpressionBenchmark`) to 1M (`CometConditionalExpressionBenchmark.c4`) across suites — so some expressions get a large unrepresentative win by accident. `CometArithmeticBenchmark` is the only suite that varies this, and only indirectly via `useDictionary` reducing distinct value counts.
**Input data is non-deterministic.** `runBenchmarkWithTable` builds the base table from an unseeded `Random.nextLong()`, so committed results are not reproducible. This is not merely cosmetic: `CometStringExpressionBenchmark` derives `REPEAT(CAST(value AS STRING), 10)` from it, giving 10-200 character strings that vary run to run, which means `lpad(c1, 150, 'x')` and `rpad` are sometimes padding and sometimes truncating.
**Fallback warnings never reach the results file.** `runExpressionBenchmark` reports a non-native plan with `println`, which goes to the console, whereas `Benchmark(output = output)` writes to the `.txt`. Anyone reading a committed results file cannot tell that a row labelled "Comet" actually ran on Spark.
## Proposed work
Rebuild the harness so that scan and result-transfer costs are either removed from the timed region or explicitly reported, then migrate the suites onto it.
- [ ] Add a no-expression baseline case to the shared harness. Take a `baselineQuery` (default: project the raw input columns) and emit `Spark (baseline)` / `Comet (baseline)` rows alongside every measurement, so the scan-plus-c2r floor is visible in every committed results file. This is mechanical and immediately stops people misreading a 1.1x as an expression win.
- [ ] Add an aggregate-sink measurement mode for expressions whose output is a string or a complex type, e.g. `SELECT sum(xxhash64()) FROM t`, so the result never crosses the JNI or row boundary. Comet aggregates natively, output is one row, c2r goes to zero. The added hash per row is small relative to `UnsafeRow`/`UnsafeArrayData` construction and is identical in both arms.
- [ ] (#5371) Apply `excludedRules` to both arms rather than only the Comet arm, and append to the existing value instead of overwriting it.
- [ ] (#5371) Assert that the benchmarked expression is actually present in the Spark baseline plan. Note: with `excludedRules` applied to both arms, the `space(2)` case is already fixed by that change alone, and this check cannot catch the `In` case below, because Spark retains `FilterExec` above the scan even when the filter is pushed down. What it does catch is expressions removed by rules that are not excluded, such as `SimplifyCasts` on a no-op cast.
- [ ] Raise `CometStringExpressionBenchmark` and `CometRegExpBenchmark` to at least 1M rows.
- [ ] Move the `In` predicate in `CometPredicateExpressionBenchmark` into the SELECT list, or disable `spark.sql.parquet.filterPushdown` for that suite.
- [ ] Set `parquet.enable.dictionary` explicitly in `CometBenchmarkBase.getSparkSession` to match the three other suites in the package, and make dictionary-encoded vs. plain a deliberate variant where it is interesting rather than an accident of the generated data.
- [ ] (#5371) Seed the `Random` in `runBenchmarkWithTable` so results are reproducible. Note: the closure runs per row on the executor, so a driver-side seed has no effect; #5371 uses a pure function of the row id instead.
- [ ] (#5371) Route the not-fully-native warning to `output` so it lands in the results file, and consider failing the run outright.
- [ ] Migrate the existing suites onto the new harness and regenerate the committed results.
`CometRegExpBenchmark` is worth calling out as already having close to the right shape: its `Comet (Scan)` case isolates the exec contribution far better than a plain Spark-vs-Comet pair does. It just runs at 1024 rows.
## Progress
- #5371 (draft) covers items 3, 4, 8 and 9.
## Corrections to the above
Two premises in this issue turned out to be wrong, found while working on #5371:
- This issue refers throughout to "committed results files". `spark/benchmarks` is listed in `.gitignore`, so no Scala microbenchmark results are tracked in git. The only committed benchmark results are the TPC JSON files under `benchmarks/results/`. The `.txt` files are local artifacts that get pasted into PRs and issues by hand, which is where a console-only warning does its damage. The last checklist item therefore has nothing to regenerate.
- The plan assertion cannot catch the `In` predicate case, for the reason noted inline above. Moving the `In` into the SELECT list stays a separate manual fix.
## Found along the way
- #5372: nine benchmark rows labelled `Comet` are actually measuring Spark (`translate`, and eight `ShortType` cast cases). Surfaced by routing the fallback warning into the results file.
Contributor guide
Assessment
This issue has not been assessed yet.