test: make expression benchmark harness fair and reproducible

[andygrove]

Which issue does this PR close?

Part of #5363. This covers checklist items 3, 4, 8 and 9 of that epic; the epic stays open.

Rationale for this change

runExpressionBenchmark in CometBenchmarkBase backs 27 benchmark suites. Four defects make its output untrustworthy, and they are worth fixing before the larger harness rework in #5363 so that later results can be compared against something sound.

Constant folding was excluded for the Comet arm only. cometExecConfigs set spark.sql.optimizer.excludedRules to ConstantFolding; the Spark case ran with folding enabled. For select space(2) from parquetV1Table in CometStringExpressionBenchmark, the Spark plan was:

*(1) Project [   AS space(2)#1366]

a folded literal doing no per-row work, while Comet evaluated space(2) for every row. That row reported a Comet regression that does not exist. The conf was also assigned rather than appended, clobbering any pre-existing exclusions.

Nothing checked that the benchmarked expression survived optimization. Rules that are not excluded, such as SimplifyCasts on a no-op cast, can remove the expression entirely and leave a case that measures only the scan.

The not-fully-native warning went to println. Benchmark(output = output) writes the results table to the .txt, but the warning only reached the console, so a reader of a results table cannot tell that a row labelled "Comet" actually ran on Spark.

Table data came from an unseeded scala.util.Random, so runs are not comparable. This is not only cosmetic: CometStringExpressionBenchmark derives REPEAT(CAST(value AS STRING), 10) from it, so lpad(c1, 150, 'x') is sometimes padding and sometimes truncating.

What changes are included in this PR?

All changes are in CometBenchmarkBase.scala. No call sites change.

• spark.sql.optimizer.excludedRules is now built once and applied to both arms, appending to the existing value rather than overwriting it.
• A structural check warns when the Spark baseline plan does no work beyond scanning and projecting bare attributes or literals. Any other node counts as work, so the join and aggregate suites, which legitimately project bare attributes, are unaffected.
• Both that warning and the existing not-fully-native warning go through a helper that writes to output as well as the console, so they land in the results file directly above the affected table.
• The base table is generated from a SplitMix64 mix of the row id instead of Random.nextLong(). Seeding a driver-side Random would not have worked, since the closure runs per row on the executor. The helper lives in a companion object because referencing a trait method from a Dataset closure captures this, which is not serializable.

Two things reviewers may want to weigh in on:

• The checks warn rather than fail. Failing would block regenerating results for suites that currently fall back, which later items in [EPIC] Reimplement Scala expression microbenchmarks so they measure expression cost, not scan and result transfer #5363 address.
• The plan check runs the Spark arm once, untimed, before benchmarking. Previously only the Comet arm was pre-run. This costs an extra run per case but warms both arms symmetrically.

Results files are not regenerated here. The final item of #5363 regenerates them once, after the baseline-case and aggregate-sink work lands.

How are these changes tested?

Benchmarks do not run in CI, so this was verified by running them locally. Compiles clean under the spark-4.1, spark-3.5 and spark-3.4 profiles.

• Reproducibility. SELECT sum(hash(value)), count(*) over the generated table returned sum=-293787312920 count=65536 on three independent runs across JVM restarts.
• The space(2) fix. With folding excluded on both arms the Spark plan becomes *(1) Project [space(2) AS space(2)#1367], evaluated per row, rather than the folded literal above.
• Both warning paths. Exercised with a temporary suite containing a bare-column projection and a case with spark.comet.exec.project.enabled=false. Both warnings appeared in the generated .txt above the correct table, with the offending plan. A control case using abs(c1) produced no warning. The temporary suite was removed.
• No false positives. CometStringExpressionBenchmark (31 expressions), CometCastNumericToNumericBenchmark and CometPredicateExpressionBenchmark were run in full. None produced a trivial-plan warning.

Running the suites with the warnings visible surfaced fallbacks that were previously console-only:

• translate in CometStringExpressionBenchmark falls back to a JVM Project.
• Eight c_short cases in CometCastNumericToNumericBenchmark do the same, for example Project [cast(c_short#19 as int) AS c_short#528].

Those rows have been reporting Spark timings under a "Comet" label. I will file a separate issue rather than address them here.

Two corrections to the epic text that came out of this work, noted for whoever picks up the remaining items:

• [EPIC] Reimplement Scala expression microbenchmarks so they measure expression cost, not scan and result transfer #5363 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 pasted into PRs and issues by hand, which is where a console-only warning does its damage.
• The epic expects the plan assertion to catch the In predicate in CometPredicateExpressionBenchmark. It does not, and cannot: Spark retains FilterExec above the Parquet scan even when the filter is pushed down, so that plan contains real work. Confirmed by running the suite. Moving the In into the SELECT list remains a separate manual fix.

[sunchao]

Thanks for improving the benchmark harness. The deterministic inputs and persisted fallback warnings are helpful. I left one [P2] comment about making the existing space(2) case measure comparable work.

[sunchao]

[P2] Could we make this benchmark depend on an input column rather than just disable ConstantFolding? The existing SELECT space(2) FROM parquetV1Table still compares different amounts of work: Spark's StringSpace calls UTF8String.blankString(2) once per row, while Comet's DataFusion space UDF receives ColumnarValue::Scalar, evaluates it once per batch, and broadcasts the result. The structural plan check also accepts the retained ProjectExec, so it does not flag the mismatch. Materializing an integer column and benchmarking space(column) would make both engines evaluate the expression across the rows.

[andygrove]

Fixed. Thanks for suggesting that.