Describe the bug
CometColumnarToRowBenchmark measures vanilla Spark in all three of its cases, so its published
numbers do not compare what they claim to compare.
The benchmark overrides getSparkSession and builds its own SparkConf without setting
spark.shuffle.manager. spark.comet.shuffle.enabled defaults to true, and
CometSparkSessionExtensions.isCometLoaded bails out in exactly that combination:
if (COMET_SHUFFLE_ENABLED.get(conf) && !isCometShuffleManagerEnabled(conf)) {
logWarning("Comet extension is disabled because spark.shuffle.manager is not set to ...")
return false
}
That disables Comet for the whole session - no scan conversion, no operator conversion - so
addC2RBenchmarkCases runs three identical vanilla Spark plans and only the case labels differ.
CometBenchmarkBase.getSparkSession sets the shuffle manager for this reason and carries a comment
explaining it. The override in CometColumnarToRowBenchmark drops it.
Evidence
Two independent checks, both on main (commit 2949fd0, Spark 4.1, release build, M3 Max):
1. The executed plan. Capturing the executed plan for a session configured exactly like this
benchmark's shows Spark's operators in the arms labelled Comet:
*(1) ColumnarToRow
+- FileScan parquet [...] Batched: true, ...
Adding spark.shuffle.manager -> CometShuffleManager to the same session and changing nothing else
makes the same arms plan CometColumnarToRow and CometNativeColumnarToRow.
2. The numbers. Running the benchmark as-is, all three arms land within ~1% in every one of the
ten groups:
Columnar to Row - Struct Types: Best Time(ms) Avg Time(ms) Stdev(ms) Relative
Spark (ColumnarToRowExec) 186 196 12 1.0X
Comet JVM (CometColumnarToRowExec) 186 196 9 1.0X
Comet Native (CometNativeColumnarToRowExec) 186 202 18 1.0X
Columnar to Row - Array Types: Best Time(ms) Avg Time(ms) Stdev(ms) Relative
Spark (ColumnarToRowExec) 197 201 3 1.0X
Comet JVM (CometColumnarToRowExec) 196 200 2 1.0X
Comet Native (CometNativeColumnarToRowExec) 196 200 3 1.0X
That result is not plausible on its own terms: the doc for
spark.comet.exec.columnarToRow.native.enabled says the native conversion "carries a fixed JNI
cost per batch and is slower than the JVM implementation for small batches", yet here it is
identical to both other arms everywhere.
Steps to reproduce
make benchmark-org.apache.spark.sql.benchmark.CometColumnarToRowBenchmark
Expected behavior
The Comet arms should plan Comet operators. Fix is either to set the shuffle manager in the
override, or to call super.getSparkSession and adjust from there the way CometReadBenchmark
does.
Additional context
Two related notes:
-
A second, quieter version of the same trap is in this benchmark:
fixedWidthOnlyBenchmark and primitiveTypesBenchmark both include tinyint / smallint
columns, which Comet's scan rejects unless spark.comet.parquet.unsignedSmallIntCheck is
false. Even with the shuffle manager fixed, those two groups fall back to Spark's scan in
every arm. CometTestBase sets that config to false; the benchmarks do not.
-
The generalisable fix is a plan assertion, so a silently-Spark arm fails loudly instead of
producing a plausible-looking table. Something like: run each arm once before benchmark.run(),
capture the executed plan through a QueryExecutionListener, and print a warning into the
results file when the arm did not plan the operator its label names. Worth considering for
CometBenchmarkBase so every benchmark inherits it.
I audited the other benchmarks under spark/src/test/scala/org/apache/spark/sql/benchmark/ for the
same problem. Only this one is affected: CometReadBenchmark calls super.getSparkSession,
CometTPCQueryBase sets the shuffle manager, and CometC2RIsolatedBench never builds a
SparkSession at all.
Describe the bug
CometColumnarToRowBenchmarkmeasures vanilla Spark in all three of its cases, so its publishednumbers do not compare what they claim to compare.
The benchmark overrides
getSparkSessionand builds its ownSparkConfwithout settingspark.shuffle.manager.spark.comet.shuffle.enableddefaults totrue, andCometSparkSessionExtensions.isCometLoadedbails out in exactly that combination:That disables Comet for the whole session - no scan conversion, no operator conversion - so
addC2RBenchmarkCasesruns three identical vanilla Spark plans and only the case labels differ.CometBenchmarkBase.getSparkSessionsets the shuffle manager for this reason and carries a commentexplaining it. The override in
CometColumnarToRowBenchmarkdrops it.Evidence
Two independent checks, both on
main(commit 2949fd0, Spark 4.1, release build, M3 Max):1. The executed plan. Capturing the executed plan for a session configured exactly like this
benchmark's shows Spark's operators in the arms labelled Comet:
Adding
spark.shuffle.manager -> CometShuffleManagerto the same session and changing nothing elsemakes the same arms plan
CometColumnarToRowandCometNativeColumnarToRow.2. The numbers. Running the benchmark as-is, all three arms land within ~1% in every one of the
ten groups:
That result is not plausible on its own terms: the doc for
spark.comet.exec.columnarToRow.native.enabledsays the native conversion "carries a fixed JNIcost per batch and is slower than the JVM implementation for small batches", yet here it is
identical to both other arms everywhere.
Steps to reproduce
Expected behavior
The Comet arms should plan Comet operators. Fix is either to set the shuffle manager in the
override, or to call
super.getSparkSessionand adjust from there the wayCometReadBenchmarkdoes.
Additional context
Two related notes:
A second, quieter version of the same trap is in this benchmark:
fixedWidthOnlyBenchmarkandprimitiveTypesBenchmarkboth includetinyint/smallintcolumns, which Comet's scan rejects unless
spark.comet.parquet.unsignedSmallIntCheckisfalse. Even with the shuffle manager fixed, those two groups fall back to Spark's scan inevery arm.
CometTestBasesets that config to false; the benchmarks do not.The generalisable fix is a plan assertion, so a silently-Spark arm fails loudly instead of
producing a plausible-looking table. Something like: run each arm once before
benchmark.run(),capture the executed plan through a
QueryExecutionListener, and print a warning into theresults file when the arm did not plan the operator its label names. Worth considering for
CometBenchmarkBaseso every benchmark inherits it.I audited the other benchmarks under
spark/src/test/scala/org/apache/spark/sql/benchmark/for thesame problem. Only this one is affected:
CometReadBenchmarkcallssuper.getSparkSession,CometTPCQueryBasesets the shuffle manager, andCometC2RIsolatedBenchnever builds aSparkSessionat all.