[rich7420]

Which issue does this PR close?

Closes #2837.

Rationale for this change

Map inputs previously fell back to Spark. Support all four explode variants with Spark's entry order, positions, schemas and NULL/empty behavior.

What changes are included in this PR?

Expose map entries as a list sharing Arrow buffers and field metadata. Reuse the bounded list operator for maps, arrays and both chain directions. Evaluate computed inputs once per batch, and split oversized rows before repeating carried columns. Retain Arrow kernels: a safe repetition probe lost at repeat five (161 vs 155 µs).

Add map SQL coverage, native-path/schema assertions, nested benchmarks and support docs. Stored float-key tests preserve the legacy normalization setting; producer normalization remains #6549.

How are these changes tested?

Local: GenerateExecSuite 82, SQL files 4, ExecRuleSuite 125, focused V2 writer test 1, release Rust explode tests 32 and planner tests 2 passed. Restoring the old chain guard makes all eight map→array native assertions fail; removing it passes. Release JNI build, semantic scalafix CHECK, Spotless, cargo fmt and operator clippy passed.

At d2e378083: M3 Pro/36 GiB, Spark 4.1.3/JDK 21, release JNI, 8 GiB heap, local[1]/one native worker. Single maps: 262,144 rows; depth-eight chains: 102,400 rows, fan-out 4×5; 20% NULL/empty. Load tests scale datasets 12×. Custom harness uses these benchmark fixtures, five-second warm-up per arm and 60 interleaved samples. ConstantFolding/InferFiltersFromGenerate excluded in all arms. Every execution checks results and native operator counts. Mean±stdev, milliseconds:

Case	Spark	Native scan + Spark generators	Comet
explode ×2	55.2±4.0	38.3±5.6	37.1±3.9
explode ×10	96.0±8.7	57.8±5.9	55.8±5.8
explode ×100	542.1±24.3	360.9±20.9	305.6±11.1
posexplode ×10	101.6±16.3	57.8±4.5	55.9±4.3
explode_outer ×10	103.3±7.1	63.3±13.2	80.9±9.1
posexplode_outer ×10	103.6±8.1	62.7±6.2	90.1±6.1
map → array	123.5±5.5	85.9±4.0	101.8±3.9
array → array	114.9±7.8	83.5±6.8	99.6±6.1
array → map	146.2±11.3	99.8±31.2	112.2±6.2

Outer and chained cases have higher idle-task latency than the native-scan control. NULL/empty outer rows gather entries instead of slicing. With local[12], 12 workers/input tasks and 30 pairs, outer CPU time fell from 1168/1219 to 1008/1099 ms; wall differences were inconclusive. For map→array, wall time was 201±16 versus 194±28 ms and native CPU was 3.9% higher (delta 56 ms, paired-bootstrap 95% CI [26,90]; earlier run [-25,43], inconclusive). Array→map used 1611 versus 1924 CPU ms against an external prior-map-admission overlay using the same operator build; wall means were noisy (245±60 versus 312±102 ms). These are workload-specific results.

Upstream CI is running. Fork Actions is enabled, but no PR-tier run has started. Merged #7335 fixes the writer's order-dependent comparison. Spark 3.4's DPP failure also appears on upstream nightly; root cause is unconfirmed. No failing test was disabled.

@sunchao @comphead @andygrove Please re-review the implementation and measurements.

[sunchao]

Summary

• Prior state and problem: Map inputs to the four explode variants fell back to Spark.
• Design approach: Reuse ExplodeExec by exposing map entries as List<Struct<key, value>>.
• Correctness / compatibility analysis: Checked Spark sources and generator tests across supported versions 3.4–4.2. Entry order, positions, nullability, empty/null outer rows, nested values and sliced offsets matched in the reviewed paths. No introduced P1/P2 issues found within this review.
• Key design decisions: The dedicated MapEntriesExpr preserves field metadata and shares buffers. Reusing existing unnest machinery keeps the implementation small. Materializing positional inputs preserves evaluation once per batch.
• Implementation sketch: Enable MapType, wrap maps before list processing, then flatten entry structs into key/value columns. Regression tests and support documentation cover the new behavior.
• Behavioral changes worth calling out: Supported map generators now execute natively under the existing explode configuration.
• Suggested improvements: None meeting the reproducible P1/P2 threshold.

Reviewed all 13 changed files in the full diff from de2e70291460ed0310428493d8c3aa5223293d08 to 76768f2ee6e1861e63ab1eed330569011904860a. The PR remained non-draft. The snapshot and live discussion contained no reviews, comments or threads.

Routed skills: review-comet-pr and review-comet-expression-pr.

Exact-head CI: Upstream CI passed Linux Rust, Spark 4.1 Comet suites and required checks. Upstream Spark SQL, Iceberg and macOS jobs were skipped. Fork CI passed Comet suites across Spark 3.4–4.2 and all seven Spark 4.1 SQL shards. Both tested merge commits have the same source tree as the reviewed head.

Validation: An isolated harness compiled unchanged exact-head modules against the lockfile’s Arrow/DataFusion versions. All 32 tests passed, including additional map checks covering 32 combinations of nested entries, outer/positional modes and batch sizes, plus scalar broadcasting and buffer preservation. Full JVM/native integration builds were not repeated locally. Integration coverage relies on verified CI. No map performance benchmark was run, so relative speedup remains unverified.

[andygrove]

I would like to do a deeper review of this PR before it is merged.

[andygrove]

Thanks for this. I compared the native map path with Spark's explode on 3.4 through 4.2. NULL and empty maps, the outer forms, positions, entry order, sliced inputs and batch splits all match, and the array path is unchanged. I didn't find anything that makes released behavior worse. I have a few test and benchmark requests, and there's one older float-key issue worth tracking.

1. Float map keys, which is not new in this PR

On Spark 4.0+, ArrayBasedMapBuilder normalizes floating-point keys before it checks for duplicates (-0.0 becomes 0.0, and NaN is canonicalized). The native map_from_arrays and map_from_entries don't. So a -0.0 key comes back as -0.0, and keys 0.0, -0.0 give two entries where Spark raises DUPLICATED_MAP_KEY. This already happens on branch-1.1, for example with a plain SELECT map_from_arrays(k, v), so this PR doesn't cause it. Map explode becomes one more place where it shows up. Could we open an issue to gate float and double keys in CometMapFromArrays and CometMapFromEntries on Spark 4.0+, through a shim that honors spark.sql.legacy.disableMapKeyNormalization? An explode case with -0.0 keys would be a good test for that fix.

2. spark/src/test/scala/org/apache/comet/exec/CometGenerateExecSuite.scala:228

The test this replaces was the only one with a JVM-fed map, a LocalTableScan with COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED=true. Could we keep it as a native check with checkSparkAnswerAndOperator, so the path from the JVM through FFI to MapEntriesExpr has coverage? While you're here, could the batch boundary test also cover map<int, boolean> values? Sliced boolean children go through zero_offsets (#6288). A batch where every map is NULL or empty under the outer forms would also help.

3. spark/src/test/resources/sql-tests/expressions/map/explode.sql:50

Could we add maps built by the native producers (map_from_arrays, map_from_entries, str_to_map), and a map<double, int> table with -0.0 and NaN keys? With the gate from the first comment, the map_from_arrays case becomes a MinSparkVersion: 4.0 check that the key comes back as 0.0.

4. spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala:62

Now that map explode runs natively by default, could you add a map case here and put Comet vs Spark numbers in the PR description? The nesting group's comment says a map was the natural model, so that group could use one.

5. spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:2637

This change isn't related to map explode. Could it go in its own PR that links the failing Spark 3.4 run? That way the #6116 author can weigh in on relaxing the direct-cause assertion, and reverting one change won't take the other with it.

[andygrove]

I filed #6549 for the float map key issue from my first comment, so no need to open one. It reproduces on branch-1.1 with map_from_entries, and with duplicate zero keys in map_from_arrays.

[rich7420]

@andygrove could you take a look , please

[sunchao]

Reviewed 8f93e97a8da5d1f3ccc879ff7deb38cfd00305b0 against base/merge base 330b29cf1013351cd9d7e93d3d37a6bc0b6d956b, with five reviewers covering native execution, planner/schema handling, Spark semantics, tests/CI, and an independent performance/memory audit. Also compared generator admission with branch-1.2 at 8af6dfbedf49b585e15dfe86cdaa8ee5d225c40b.

Requesting changes for one newly exposed map-generator failure, detailed inline. No additional actionable issue survived reconciliation with the existing discussion. The historical nested-workload timing concern remains disclosed in the PR and was not independently remeasured.

Validation: the isolated harness compiled unchanged current-head native modules against Arrow 59.3.1/DataFusion 55.2.0. Its 25 tests passed, including three additional map-expression probes and a test confirming the reported failure. A separate 20,000-case short-string differential probe passed. Plain Spark 4.1.3 successfully ran the wide-map aggregate comparison. These are native-operator and plain-Spark checks, not a full local Comet JNI integration run.

CI run 40182676698 tested a merge tree identical to this head. Verified logs show 2,648 Rust tests, 1,782 Spark 4.1 execution tests and 2,282 expression tests passing, including the new map suites. Applied run-spark-4.1-tests and run-all-spark-profiles to request the skipped coverage. Those additional runs are still in progress.

[comphead]

The map path itself is small: MapEntriesExpr shares the map's buffers and the existing list unnest does the rest, and the output nullability matches ExplodeBase in 3.4.3 through 4.1.3. The offset-overflow fix from the last round is in place. Most of the remaining lines change how every explode repeats carried columns, so four comments are about scope and the cost of the new split rule. git merge-tree also shows this conflicts with #7253 in five files, two hunks of them semantic in planner/explode.rs, since both PRs reuse the struct-flatten slot.

[comphead]

span is the largest buffer of any carried column across the whole input batch, but what can overflow is the bytes of the rows in one chunk times their repetition. Take an 8,192-row batch carrying a 1 KiB string. Its values buffer is 8 MiB, so the limit is 255, and every row with more than 255 elements is then built alone in 255-element pieces, with next_chunk_rows also ending the chunk before such a row. On main, 300-element arrays with that payload are built about 27 rows at a time, roughly 8 MB of copied strings, far from the offset limit. split_limit is set for every depth-1 List, so this changes array explode too, and fixed-width carried columns count toward span although they cannot overflow an offset. Could the limit come from the offset-addressed bytes of the rows in the chunk, or apply only when a row's own carried bytes times its length can overflow? I derived this from the code and have not measured it. A bench case with a 1 KiB carried string and a few hundred elements per row would show it.

[sunchao]

I reproduced the batch fragmentation on the existing array path at 259c891c2. With a batch size of 8,192, the input was 33 rows. One carried a 16 MiB string and an array with one element. Each of the other 32 carried a one-byte string and an array with 300 elements. The base (330b29cf) and previous head (8f93e97a) emitted 2 batches. This head emitted 97, with exactly the same 9,601 rows and 16,786,816 payload bytes. The wide string is emitted only once, but its size caps every other row at 127 repetitions, so each 300-element row gets three builds.

A second operator probe carried a single one-byte string sliced from a larger Arrow buffer. With 64 MiB of unselected bytes behind that slice, exploding 1,024 elements changed from 1 batch to 34 for just 1 KiB of logical string output. Across 200 runs the previous head took about 4.9 ms total and this head took 48.5 ms. The no-padding control stayed at one batch on both heads. These are isolated operator timings with unoptimized dependencies, not Spark benchmark numbers.

Could the split decision use the bytes and child-offset spans actually repeated in the candidate chunk? That would retain the wide-map fix without penalizing small array rows for another row's payload or a slice's unused backing buffer. The 33-row input would make a regression test for both output and batching.

[comphead]

This keeps a map generator on Spark whenever another generator sits above it, while an array generator in the same position stays native, and array-only generator chain stays native asserts that. The stated reason is carried-column repetition between generators, which costs the same whether the lower generator reads a map or an array. The description calls the 97 ms against 90 ms comparison historical, and c4525cb85f has since changed how carried values repeat. Could you post nested numbers at this head for the map chain and the equivalent array<struct> chain? If the gap is gone, the rule, its tests and the operators.md paragraph can go. If it remains, it is a chained-generator issue rather than a map one. The benchmark also replaced the native array<struct> depth-8 chain with this map chain, which now times Spark generators, so keeping both cases would answer this.

[comphead]

repeat_short_strings, with its StringArray::new_unchecked, and repeat_primitive change how every explode repeats carried columns, arrays included, and map support does not need them. There are no current-head numbers for them, and the new explode_chained_carry and fan_out_* Criterion groups have no posted results. Could they move to their own PR with before and after numbers from benches/explode.rs? That keeps this PR to map support and lets the unsafe path be reverted on its own, as andygrove asked for the ParquetReadSuite change.

[comphead]

This and the posexplode(m) case in posexplode.sql were fallback checks, and as plain queries they now repeat explode_maps in map/explode.sql, which has the same entries, empty and NULL shapes for all four generators. Could they be dropped? with batches containing only null and empty maps in CometGenerateExecSuite also looks like a fit for that fixture with -- Config: spark.comet.batchSize=4. I have not run that conversion.

[sunchao]

The original wide-map overflow is fixed at 259c891c2. The exact native operator returned all 2,049 entries in nine batches of at most 256 rows, and the Spark integration tests passed for all four map generator variants. I resolved my original thread.

I'm requesting changes for a performance regression on existing array explode queries. A single 16 MiB carried value that is emitted once changes how the operator groups the other 32 rows carrying one-byte strings. The base (330b29cf) produced 2 batches, and this head produced 97 for the same 9,601 rows and about 16 MiB of payload. Please apply the offset limit to the data actually repeated by a chunk, keeping the new map overflow protection. I put the reproducer and separate timing results in the existing split-limit discussion.

The five reviewers covered native splitting, planner and positions, Spark fallback, CI and tests, and shared array performance. All 28 tests in the local native harness passed, including the original wide-map probe and 320 combinations of slicing, parallel lists, nested values and outer behavior. The operator timings are local measurements, and I did not rerun the JVM benchmarks. On the hosted merge build, Rust, every Spark execution profile and the Spark 4.1 SQL shards passed. The Spark 4.1 expressions job stopped on a Maven Central HTTP 429 before running tests. Its retry is queued.

[sunchao]

The revised explode path preserves map entry order and the original positions across output batches, including chained generators. One regression test puts a near-2-GiB allocation into the ordinary unit-test run. Separately, the red Spark 3.4 scan job failed the AQE DPP broadcast-stage-order test, which also failed on #6784. I could not attribute that failure to map explode. CI log

[sunchao]

This test repeats a 1 MiB UTF-8 value 2,049 times and materializes the output. The first 2,047-row batch alone needs 2,047 MiB for the string payload. Could the test assert the real 32-bit boundary arithmetically, then drive the stream with a small injectable byte limit to verify splitting and output values? That would keep the regression in the default suite without requiring a multi-gigabyte allocation.