[rich7420]

Which issue does this PR close?

Closes #2837.

Rationale for this change

Map inputs to all four explode variants previously fell back to Spark. Enable native generation while preserving entry order, positions, schemas and NULL/empty behavior.

What changes are included in this PR?

• Reuse bounded list unnest through a buffer-sharing MapEntriesExpr; evaluate computed inputs once per batch. Reduce carried short-string copying while retaining Arrow take for other shapes.
• Cover JVM-fed maps, sliced nullable Boolean children, batch boundaries, NULL/empty outer batches and native map producers, with native-plan assertions.
• Add map/nested benchmarks and support documentation. Float-key producer normalization remains in Native map construction doesn't match Spark 4.0+ float key normalization (-0.0 keys, missing DUPLICATED_MAP_KEY) #6549; stored-key tests here use its legacy setting.

How are these changes tested?

At 4194af733, upstream CI passed, including Rust, native build, lint, all Spark 4.1 suites and TPC-H/TPC-DS. Supplemental Spark SQL and cross-profile CI: requested via fork #32, but fork Actions is blocked and no current-head run started. Please apply run-spark-4.1-tests and run-all-spark-profiles upstream before queueing.

Historical operator/JNI checks, Spark 4.1 Generate/SQL tests and benchmark result/plan assertions passed. Benchmarks at ff55b13cd used Apple M3 Pro, Spark 4.1.3/JDK 21, local[1], 8 GiB heap and release native code, with 2-second warmup/measurement per arm:

Historical case	Spark mean ± stdev (ms)	Comet mean ± stdev (ms)
Map fan-out 2	41 ± 4	25 ± 4
Map fan-out 10	71 ± 5	47 ± 6
Nested map followed by inner array	91 ± 5	87 ± 4

A longer 10-second warmup/measurement control still averaged 97 ms with native generators versus 90 ms with Spark generators above a native scan, a noisy 7.8% remaining gap. This concern and Andy's requested deeper review remain open before merging. No new timings are claimed for the current head. Benchmark source.

[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.

[sunchao]

[P2] Bound one map's expansion before repeating carried strings

Could we keep this shape on Spark until the native operator can split an individual map before repeating its carried columns? A single input row with a 2,049-entry map<int,int> and a 1 MiB STRING payload is enough to fail. PendingInput::next_chunk_rows never splits one input row, so repeat_arrs calls Arrow take for the entire map before BatchSplitStream runs. The repeated string reaches the 32-bit offset limit even with a configured batch size of 256.

I reproduced Arrow error: Offset overflow error: 2147483648 on the first output poll using this head's unchanged MapEntriesExpr and ExplodeExec, Arrow 59.3.1 and DataFusion 55.2.0. A small control passes. With the same input stored in Parquet, plain Spark 4.1.3 successfully computes SELECT sum(CAST(length(concat(payload, CAST(key AS STRING))) AS BIGINT)) FROM wide_map LATERAL VIEW explode(m) e AS key, value. Its executed plan keeps payload through Generate and consumes the rows in the aggregate.

The base and branch-1.2 leave map generators on Spark, so this admission change exposes a query failure even though array explode already has the underlying limitation. Please add a wide carried-column regression and split within a map before materialization, or retain a safe fallback for this shape.

[rich7420]

ok , thanks for catching this