test: add explode operator microbenchmark

[andygrove]

Which issue does this PR close?

N/A. This adds a benchmark for an existing operator, so there is no behavior change to track.

Rationale for this change

CometExplodeExec has no benchmark. Generator cost is driven by dimensions that other operator
benchmarks do not exercise: how far each input row fans out, whether the generator emits a
position column or preserves rows with null and empty arrays, the element type being unnested,
and how many columns have to be replicated alongside the generated one. Without numbers for
those, there is no way to tell whether a change to the operator helped, and no way to see which
shapes Comet declines to convert.

What changes are included in this PR?

A new CometExplodeBenchmark with 12 cases over 256K input rows, in four groups:

• Fan-out: explode over array<bigint> of length 2, 10, and 100.
• Generator variants: explode, posexplode, explode_outer, posexplode_outer.
• Element type: array<bigint>, array<string>, array<struct<a bigint, b string>>.
• Carried columns: explode alone against explode plus three passthrough columns.

Two details worth calling out for reviewers:

One in ten rows holds a null array and another one in ten holds an empty array. Without that,
explode_outer and explode would produce identical output and the variant group would be
measuring the same query twice. The empty array is built with slice, not array(), because
array() types as array<null>.

Each array column gets its own temp view rather than sharing one wide table, so a case is never
charged for scanning an array column it does not read.

Reported times are whole-query totals and include the Parquet scan and result transfer, which is
noted in the class comment: at fan-out 2 the scan is a large share of the total and the ratio
understates the difference between the two implementations. The Rate and Per Row columns are
per input row, not per generated row, since fan-out varies across cases.

How are these changes tested?

Run locally on an Apple M3 Ultra with local[5]:

SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometExplodeBenchmark

All 12 cases complete, and none trips the harness warning for a plan that is not fully Comet
native, including the array<struct> case. Comet is roughly 2x Spark across every case, and
flat across fan-out, generator variant, and element type (Best ms, Spark then Comet):

Case	Spark	Comet	Relative
fan-out 2 / 10 / 100	20 / 27 / 150	10 / 14 / 75	2.1X / 1.9X / 2.0X
explode / posexplode / explode_outer / posexplode_outer	26 / 25 / 25 / 25	14 / 14 / 12 / 13	1.9X / 1.8X / 2.1X / 1.9X
bigint / string / struct	25 / 32 / 42	13 / 18 / 24	2.0X / 1.8X / 1.8X
alone / plus 3 carried columns	25 / 30	12 / 18	2.0X / 1.7X

Correctness of the query forms themselves is already covered by CometGenerateExecSuite, which
tests explode and explode_outer over simple, empty, and null arrays.

[sunchao]

Summary

This first review evaluates the new explode benchmark at commit bf0324024d27985096bc590153110a3fe9b7c5df. The coverage dimensions are useful, but four benchmark-design issues prevent several reported comparisons from isolating the native explode operator.

Prior state and problem

CometExplodeExec previously lacked a dedicated Scala microbenchmark, making it difficult to characterize fan-out, generator variants, nested element types, and columns carried through generation. Existing expression benchmarks already include Parquet scan and result-transfer costs, which become especially significant for an operator that multiplies its input cardinality.

Design approach

The change adds a standalone CometExplodeBenchmark that materializes deterministic 256K-row Parquet-backed temporary views and compares Spark with Comet through the shared runExpressionBenchmark harness. Its cases cover fan-outs of 2, 10, and 100; four generator variants; bigint, string, and struct elements; and zero versus three carried columns.

Correctness / compatibility analysis

The generated data includes exactly 26,215 null arrays and 26,215 empty arrays, with 209,714 nonempty arrays, and the relevant native explode operator is enabled by default. However, Catalyst filters null and empty arrays before non-outer generators, the noop sink forces Comet's generated columnar output through row conversion, and mismatched source distributions and projected Parquet schemas introduce additional work that is unrelated to the intended operator comparisons.

Key design decisions

The benchmark intentionally stores each array shape in a separate Parquet-backed view, reports rates per original input row, and delegates warmup, iteration control, and Spark-versus-Comet configuration to the common benchmark helper. Null and empty arrays are deliberately included to distinguish outer generators, while the carried-column case reuses a wider source view.

Implementation sketch

arrayColumn builds typed arrays with deterministic null and empty branches; createView writes each generated dataset and registers a temporary view. Four runBenchmark groups then call runExpressionBenchmark, whose Spark and Comet cases execute the corresponding SQL query via Spark's row-oriented noop writer.

Behavioral changes worth calling out

This is a benchmark-only change and does not modify production explode semantics. The benchmark-only CI route compiles and lints against Spark 4.0, while the broader Spark integration matrix is intentionally skipped; reported rates use 262,144 original input rows even though optimized non-outer generators receive only 209,714 rows.

Suggested improvements

Keep exploded results columnar or use a matched conversion baseline; apply generator-filter optimization controls symmetrically so all variants receive the intended inputs; align string distributions and Parquet encoding across element-type datasets; and hold the projected scan schema constant, or subtract a matched scan baseline, when comparing carried columns. The inline P2 comments identify the exact affected cases and their concrete measurement consequences.

[sunchao]

[P2] Match string cardinality across the element-type datasets

arr_str10 contains 262,151 distinct str_<id+x> values, while this struct's string field contains only the 10 values s1 through s10, across the same 2,097,140 non-null elements. The default Parquet writer enables dictionaries with a 1 MiB dictionary-page threshold, so the struct field stays dictionary-encoded while the high-cardinality standalone string field exceeds that threshold. Consequently the advertised string-versus-struct comparison also measures substantially different Parquet encoding, decoding, compression, and scan work. Make struct.b row-varying like the standalone string case, or explicitly match the value distribution and encoding before comparing element types.

[sunchao]

[P2] Isolate explode from per-generated-row columnar-to-row conversion

runExpressionBenchmark executes .noop(), which writes to a DataWriter[InternalRow]. Therefore the Comet case converts every generated columnar output row to a Spark row, whereas Spark's GenerateExec already emits rows; the existing CometColumnarToRowBenchmark deliberately uses this same .noop() pattern to measure that conversion. Here fan-out 2 produces 419,428 rows, while fan-out 100 produces 20,971,400 rows, so conversion work scales 50x with the dimension attributed to explode and also varies with strings, structs, and carried columns. This is not fixed result-transfer overhead, and a scan-only baseline will not remove it. Consume generated batches without a row boundary, or include a matched columnar-to-row baseline so the benchmark isolates CometExplodeExec.

[sunchao]

[P2] Prevent optimizer filtering from changing generator inputs

Catalyst's InferFiltersFromGenerate inserts size(arr) > 0 AND arr IS NOT NULL below explode and posexplode, but explicitly does not apply to their outer variants. With these generated datasets, the non-outer operators therefore receive only 209,714 rows, while explode_outer and posexplode_outer receive all 262,144; none of the 52,430 null or empty arrays reaches the non-outer operator. The purported generator-variant comparison consequently mixes an extra upstream filter and different operator input cardinalities, and its non-outer rate is normalized using rows the generator never processes. Exclude InferFiltersFromGenerate symmetrically for Spark and Comet, preserving the helper's existing ConstantFolding exclusion, or otherwise give both variants matched operator inputs.

[sunchao]

[P2] Keep the Parquet scan schema constant when measuring carried columns

SELECT explode(arr) prunes k, s, and v from the Parquet scan, whereas this query must scan and decode all three additional columns before explode runs. Across 262,144 input rows, that adds 786,432 scalar values, including 262,144 distinct strings, although both cases produce the same 2,097,140 generated rows. The measured difference therefore combines extra Parquet I/O and string decoding with carried-column replication, so it cannot isolate the stated carried-column dimension. Keep the projected scan schema identical between the two cases, or subtract a schema-matched scan baseline.