fix: enforce null-key rejection and mapKeyDedupPolicy in native map construction

[peterxcli]

Which issue does this PR close?

• Closes [Bug] map_from_arrays / map_from_entries do not enforce null-key rejection or spark.sql.mapKeyDedupPolicy #4680.
• Closes str_to_map does not honour mapKeyDedupPolicy=LAST_WIN #5857.

Rationale for this change

Spark builds every map through ArrayBasedMapBuilder, which refuses a NULL key and resolves duplicate keys according to spark.sql.mapKeyDedupPolicy. Comet's map_from_arrays and map_from_entries did neither. A NULL inside the keys array produced a map with a NULL key instead of an error, and setting the policy to LAST_WIN pushed the whole expression back to Spark.

DataFusion 55 supplies what was missing. Its datafusion.spark.map_key_dedup_policy option takes the same EXCEPTION and LAST_WIN values as the Spark config, and the datafusion-spark map kernels already follow it. Once Comet passes the setting through, LAST_WIN runs natively, and the remaining checks that ArrayBasedMapBuilder performs cost only a few lines on top.

What changes are included in this PR?

spark.sql.mapKeyDedupPolicy now crosses JNI. CometExecIterator.serializeCometSQLConfs sends it explicitly, since cometSqlConfs carries only keys under spark.comet., and prepare_datafusion_session_context applies it to the session as datafusion.spark.map_key_dedup_policy.

A second change was needed before that setting could reach a kernel at all. create_scalar_function_expr handed every ScalarFunctionExpr a fresh ConfigOptions::default(), so any kernel reading a session option saw DataFusion's defaults. It now passes the session's own ConfigOptions.

native/spark-expr/src/map_funcs/map_builders.rs adds three wrappers, SparkMapFromArrays, SparkMapFromEntries and SparkStrToMap. Each calls the matching datafusion-spark kernel, adds the checks that kernel skips, and translates its errors into the Spark error classes that SparkErrorConverter converts back into QueryExecutionErrors:

• a NULL key raises NULL_MAP_KEY, before any check for duplicates, matching the order Spark applies them;
• key and value arrays of unequal length raise MAP_KEY_VALUE_DIFF_SIZES;
• a duplicate key under EXCEPTION raises DUPLICATED_MAP_KEY and names the key.

str_to_map needs only the last of these, because splitting a string never yields a NULL key. Passing the config through also fixed its LAST_WIN case, which used to raise an error where Spark returns a map.

CometMapFromArrays now emits map_from_arrays. It used to emit the generic map wrapped in CaseWhen(IsNotNull(left) AND IsNotNull(right), ...) so that a NULL input array yielded a NULL map; the Spark kernel already behaves that way, so the wrapper came out. Both serdes also drop their LAST_WIN Incompatible branch.

One difference with Spark remains. ArrayBasedMapBuilder normalizes a floating point key before storing it, so -0.0 becomes +0.0 and every NaN collapses into one. The native builders compare the raw Arrow values, so a map built from both -0.0 and +0.0 keeps two entries where Spark reports a duplicate key. The compatibility notes record this, and spark.comet.exec.strictFloatingPoint makes Comet decline a floating point key type for anyone who needs the guarantee.

How are these changes tested?

The 21 native unit tests cover the wrappers. Two of them pin the exact wording DataFusion uses when it reports a duplicate key, because the wrapper reads that message to recover the key it should name. If DataFusion rewords the message, those tests fail rather than the error quietly degrading into a generic execution failure.

Seven new tests in CometMapExpressionSuite run each case through both engines and compare the exception type, error class and SQLSTATE, along with the answers each engine returns under LAST_WIN.

Among the SQL fixtures, the two *_dedup_policy.sql files used to assert the LAST_WIN fallback and now assert native execution. map_from_arrays.sql, map_from_entries.sql and str_to_map.sql gained the EXCEPTION error cases, and str_to_map_dedup_policy.sql is new. That also retires the TODO: Add LAST_WIN policy tests when spark.sql.mapKeyDedupPolicy config is supported note in str_to_map.sql.

The test for mismatched array lengths compares the two engines against each other instead of naming an error condition. Spark still reports that case through a _LEGACY_ERROR_TEMP_* condition whose number moves between Spark versions, so CometTestBase.checkSparkError now builds on a new checkSparkErrorParity helper.

The ConfigOptions change affects every scalar function, so the full 487-fixture suite ran green as well.

[rich7420]

This can return a key from a preceding row. With keys [[10], [20]] and values [[100], [200]], slicing both to the second row returns {10: 200} instead of {20: 200}. The previous MapFunc returns {20: 200}.

The helper applies a zero-based keys_mask to the unsliced flat_keys, while value indices include the starting offset. I reproduced this through a native GlobalLimitExec -> ProjectionExec component test on DataFusion 55.0.0; the relevant kernels are unchanged in 55.1.0.

Please fix the offset handling in the helper or normalize the inputs before delegation, and add a sliced-list regression test. The newly enabled LAST_WIN path for map_from_entries is affected too.

[rich7420]

Verified

[rich7420]

For keys [1, 1, NULL] under EXCEPTION, Spark 4.1.3 reports DUPLICATED_MAP_KEY, but this pre-scan reports NULL_MAP_KEY. Spark inserts entries in order and fails on the second key before reaching the null.

Please preserve that check order in both builders and update the comments claiming null-key errors always take precedence.

[rich7420]

Verified at 3085702: duplicate/NULL error ordering is fixed. Please also update the two map-constructor sections in map_funcs.md that still say "ahead of any duplicate-key check".

[andygrove]

MapBuilderSupport.keySupport only looks at the floating point gate, but MapKeySupport.keySupport a few lines above declines a non-default string collation for map lookups. Spark's ArrayBasedMapBuilder picks a collation-aware TreeMap for any StringType that is not supportsBinaryEquality, so under UTF8_LCASE the keys 'a' and 'A' are the same key, while the native builder compares raw Arrow values.

SELECT map_from_arrays(array(CAST('a' AS STRING COLLATE UTF8_LCASE), CAST('A' AS STRING COLLATE UTF8_LCASE)), array(i, i)) FROM t looks like it reaches the native path, since both casts have Literal children so CometCast folds them and supportedDataType accepts a collated StringType. Spark 4 raises DUPLICATED_MAP_KEY there and Comet returns a two-entry map. The LAST_WIN side worries me more, because the old isLastWin branch declined and sent that case back to Spark, so it used to be correct and is not after this change. Would it make sense for MapBuilderSupport.keySupport to call hasNonDefaultStringCollation the way MapKeySupport.keySupport does, with a fixture next to element_at_map_collation.sql to pin it?

The floating point note also reads as though the only difference is duplicate detection, and I think it is off in both directions. Spark only normalizes map keys from 4.0.0 onwards. spark.sql.legacy.disableMapKeyNormalization is marked .version("4.0.0") and the 3.5 ArrayBasedMapBuilder has no keyNormalizer at all, so on 3.4 and 3.5 the native builder already matches and spark.comet.exec.strictFloatingPoint declines for nothing. On 4.0 and later MapFromArrays calls mapBuilder.from(...), which reuses the original key array whenever the row has no duplicates, so a lone -0.0 key stays -0.0 in Spark too and "a -0.0 key is stored as +0.0" is not observable through map_from_arrays. MapFromEntries is the one that stores the normalized key, because it goes through put and build(), so SELECT map_from_entries(array(struct(-0.0D AS key, 1 AS value))) gives {0.0 -> 1} on Spark 4 and {-0.0 -> 1} on Comet with no duplicate anywhere. The NaN half overstates it too, since ScalarValue compares floats by to_bits, so two double('NaN') keys do collapse natively. Could the note be reworded around what actually differs per function and scoped to Spark 4.0 and later? A fixture for a floating point map key would help, since neither the default path nor the new strict decline is exercised today.

One smaller thing. The comment on checkSparkErrorParity says the mismatched-length case goes through a _LEGACY_ERROR_TEMP_* condition whose number moves between versions, but reading it out of the shipped jars it is _LEGACY_ERROR_TEMP_2128 on 3.4.3, 3.5.8, 4.0.1, 4.1.3 and 4.2.0 alike. If that holds then checkSparkError(df, "_LEGACY_ERROR_TEMP_2128") and expect_error(_LEGACY_ERROR_TEMP_2128) pin it directly and the new helper is not needed. Is there a version where the number actually differs?

On sequencing, #5846 also rewrites CometMapFromArrays.convert and adds its own SparkMapFromArrays re-export, and #5844 adds the CodegenDispatchFallback for LAST_WIN that this PR would make unnecessary. I have commented on both pointing here. Worth agreeing an order with @sunchao and @LinSimon-901101 so the same lines are not landed twice.

[peterxcli]

@andygrove @rich7420 addressed review! ptal again, thanks!

[rich7420]

Please preserve short-circuit evaluation. With ANSI enabled and a Parquet row (k = NULL, v = 'bad'):

SELECT map_from_arrays(k, array(CAST(v AS INT))) FROM t;

Spark returns NULL; Comet raises CAST_INVALID_INPUT because the cast runs before the NULL check. Reproduced on Spark 4.1.3; restoring the CASE guard makes it pass. Please add this regression test.

[andygrove]

Triage note: #5844 also makes map_from_arrays work under mapKeyDedupPolicy=LAST_WIN, but by mixing CodegenDispatchFallback into CometMapFromArrays rather than running it natively. It keeps MapKeyDedupPolicySupport, which this PR deletes, and rewrites the same map_from_arrays_dedup_policy.sql fixture, so the two cannot both land as written.

I think the native route is the one we want here, provided DataFusion's kernel matches ArrayBasedMapBuilder on ordering as well as dedup. Worth a word with @LinSimon-901101 so that PR is not reworked further against a shape that is about to disappear. Separately, #5846 is fixing per-row length validation and null short-circuiting in the same native map_from_arrays path, so there is a three-way sequencing question in map_funcs/mod.rs worth sorting out with @sunchao as well.

[comphead]

ON it

[comphead]

Reviewed at 88c877b. Line numbers below refer to that commit. Everything here comes from reading the code. Nothing was built or run.

The approach looks right to me. Wrapping the datafusion-spark kernels and reading the policy per task matches Spark outside whole-stage codegen. The items below are mostly about test strength, redundancy and per-batch cost.

Major

1. The two dedup-policy timing tests never re-run df (CometMapExpressionSuite.scala:255-333)

• Problem: checkAnswer(df.select("a", "e", "s"), …) (lines 274, 287 and 318) creates a new QueryExecution, so every "run again" step plans a fresh query instead of re-running df. The .select came in with the padding columns in 362f288. Before that, 6962053 called checkAnswer(df, …).
• Consequence:
  • The "freeze the policy at first execution" design that 362f288 rejected would pass both tests.
  • The first half of the "materialize then execute" test would pass even with plan-time capture.
  • So the tests don't pin the design the PR settled on.
• Fix:
  • Assert on df itself: compare the first three fields of df.collect(), or add the padding values to the expected row.
  • Then one sequence per config covers both rejected designs: materialize under policy A, execute under B, re-execute under A.
  • assertDuplicateMapKey re-implements the structuredError helper nested inside checkSparkError, but drops its "no CometNativeException in the cause chain" check. Make that helper protected and reuse it.

Minor

2. The length pre-check changes which error Spark would report (map_builders.rs:259-293)

• Problem: When no key is NULL, validate_map_from_arrays checks every row's lengths before the kernel looks for duplicate keys.
  • Take row 0 = [1,1]→[a,b] and row 1 = [2]→[c,d].
  • Comet raises _LEGACY_ERROR_TEMP_2128 (length mismatch).
  • Spark raises DUPLICATED_MAP_KEY for row 0.
• Why it's avoidable: the kernel already checks lengths row by row, in Spark's order (utils.rs:199-204 in datafusion-spark 55.1).
• Fix: return early when no row that gets built has a NULL key, as the entries validator already does. Translate the kernel's length message in as_spark_error, with a pin test like the one for the duplicate-key message. This also removes two Vec<usize> offset copies per batch.

3. The inner values IS NOT NULL guard no longer does anything (maps.scala:242-265)

• Why it's redundant:
  • #5846 needs the inner guard because DataFusion's MapFunc only honours NULL key rows.
  • This PR switches to the datafusion-spark kernel. That kernel returns NULL for a NULL values row (utils.rs:196-199), and the wrapper skips such rows (map_builders.rs:281).
  • Spark's BinaryExpression.eval evaluates right whenever left is non-null. So CASE WHEN keys IS NOT NULL THEN map_from_arrays(keys, values) END keeps Spark's evaluation order.
• What it saves: values is evaluated once instead of twice, and one filter-plus-merge disappears on batches where a values array is NULL. NullGuardSupport could then check only expr.left.
• Knock-on: rows with a NULL values array would then reach the wrapper. Its NULL walk should only consider rows that actually get built (see item 4).
• Separate small fix: drop setElseExpr(null). create_case_expr wraps the ELSE in a Cast, and DataFusion's NULL-ELSE shortcut only recognizes a bare null Literal.

4. map_from_entries takes the slow path on batches that won't fail (map_builders.rs:295-327)

• Problem: the per-entry ScalarValue walk starts whenever the key child has any NULL. A NULL struct entry usually has a NULL key underneath it, and that row just becomes a NULL map, not an error.
• Cost: under EXCEPTION, one such row makes every other row in the batch pay a try_from_array plus a HashSet insert per entry. The kernel then repeats the same work.
• Fix:
  • Only enter the walk when some valid entry has a NULL key. That is true when the null count of NullBuffer::union(Some(&key_nulls), structs.nulls()) exceeds the null count of structs.nulls(), which costs one AND and a popcount.
  • Do that check before line 302 builds the offsets Vec.
  • In check_keys_in_order, seen.insert(key.clone()) clones every key in case it is needed for the error message. HashSet::replace hands back the duplicate without the clone.

5. The sliced-list workaround has no exit and copies too much (map_builders.rs:218-257)

• No exit: there is no link to apache/datafusion#25419. It is fixed on DataFusion main by apache/datafusion#25431, which is not in 55.1. explode.rs and planner.rs already carry "delete once Comet moves to a DataFusion release carrying apache/datafusion#N" notes, so there's a pattern to follow.
• Copies too often:
  • Only a nonzero first offset breaks the kernel's mask. Arrow's filter accepts a shorter predicate, so a head slice (for example from LIMIT) is safe.
  • But entries_start_at_zero also requires the last offset to equal values.len(), so head slices get deep-copied anyway.
• Copies too much: take copies every nested child. A zero-copy rebase (shifted offsets plus values.slice(first, last - first)) is what #25431 does, and it costs O(rows).
• Dead code: the FixedSizeList arm can never fire.

6. Doc errors

• Wrong error class: map_funcs.md:53 and the module doc of map_builders.rs say Spark raises MAP_KEY_VALUE_DIFF_SIZES. Spark users actually see _LEGACY_ERROR_TEMP_2128, because ShimSparkErrorConverter maps the native error to mapDataKeyArrayLengthDiffersFromValueArrayLengthError(). The PR's own test pins _LEGACY_ERROR_TEMP_2128.
• Wrong comment on expand_scalars: it says the kernel "expands them anyway". That is false when every argument is a scalar. make_scalar_function computes one row and returns a Scalar, while the wrapper computes number_rows rows. It rarely matters in production because Spark constant-folds all-literal calls. CometConcatWs shows the in-repo way to handle this case.

Simplification

Tests to remove or move (the second bullet applies once #5846 lands):

• Nondeterministic-child duplicates: remove the Scala test "a nondeterministic child falls back under LAST_WIN", plus the last query and its table in map_from_arrays_dedup_policy.sql. They repeat map_from_arrays_nondeterministic_child.sql, and getSupportLevel never reads the policy.
• Overlap with #5846: remove the Scala tests "a null keys array skips…", "…different lengths are rejected" and "a null input array gives a null map". #5846 has the same three cases.
• LAST_WIN halves: remove the LAST_WIN halves of both "duplicate key follows…" tests. The map_keys/map_values column queries in the fixtures already pin that behavior.
• "a null entry gives a null map": turn the Scala test into a table row in map_from_entries.sql.
• Rust tests to drop:
  • the five LAST_WIN value tests (*_honours_last_win ×3, *_keeps_…_under_last_win ×2)
  • *_rejects_null_key ×2
  • rejects_key_value_length_mismatch
  • These only exercise the upstream kernel, and the fixtures check the same behavior against Spark. Also merge the two unquoted-key message pin tests into one.
• Routing fixtures: in all three routing_map_legacy_*.sql files (including _opt_in, which the PR didn't touch), remove:
  • the map_from_entries query, its comment and the e column
  • the LAST_WIN and MapFromEntries.allowIncompatible config lines
  • First move the "stays native" check into the *_dedup_policy.sql files with expect_native(map_from_entries) and expect_native(str_to_map). A plain query also passes when the expression goes through the JVM dispatcher, so it doesn't prove native execution.
• Weaker or duplicate fixture queries:
  • literal queries that repeat a table row (keep one literal query per file for the scalar path)
  • the column query that doesn't pin entry order
  • str_to_map(s, ',', ':'), which is the same expression as str_to_map(s)
  • the unused k and v columns in test_map_from_arrays_nondet

Keep the checkSparkError, slice, error-ordering and message-pin tests. expect_error checks neither native execution nor that the error was converted. For example, in str_to_map.sql, expect_error(DUPLICATED_MAP_KEY) also matches the raw upstream message. expect_error(Duplicate map key a was found) should pin the converted one, but I haven't verified that. The slice fix can be tested end to end with the suite's existing ORDER BY … LIMIT … OFFSET pattern, which I'd use through map_from_entries.

Code

• Wrapper structs: use #[derive(Debug, Default, PartialEq, Eq, Hash)], as CometCollectList and CometConcatWs do. The three new() functions have no callers.
• element_validity: it can be logical_nulls().filter(|n| n.null_count() > 0), which also catches nulls inside dictionary values.
• Unreachable code:
  • The LargeList and FixedSizeList arms can't be reached, because Comet maps every Spark ArrayType to List.
  • The row-count check can't fire after expand_scalars.
• Lint allow: clippy::allow_attributes in the allow list is DataFusion's lint policy, not Comet's. agg_funcs/mode.rs uses ScalarValue hash keys with no allow at all. Worth confirming with clippy.
• jni_api.rs: prefer options_mut().spark.map_key_dedup_policy = policy.parse()? over set_str, which unwraps. A future rename then fails to compile instead of panicking on every plan.
• planner.rs:3673: session_config_options() has one caller, yet its doc describes it as the general way to get the options.
  • Other sites still pass ConfigOptions::default(): planner.rs lines 814, 1128, 1147, 1163 and 1193, five sites in temporal.rs, and modulo_expr.rs:282.
  • Either inline it (as at :2342), or capture the options once in PhysicalPlanner::new and use them everywhere (fine as a follow-up).
  • No kernel at those other sites reads config today, so this is a trap for later, not a bug.
• maps.scala: list the strict floating-point decline in getIncompatibleReasons(), as CometSortArray and CometMapSort do. The note also says "fall back to Spark", but map_from_entries goes through the JVM dispatcher instead.
• NullGuardSupport: it has one caller, and its doc describes dispatcher routing that caller doesn't have. CometElementAt (arrays.scala:651) already declines the same case inline. Either inline the object until #5867 adds more callers, or use it from CometElementAt too.

Comments: several explanations are repeated many times. Policy forwarding is explained in more than a dozen places, and the nondeterministic-child and AND-guard reasoning and the floating-point note in several each. Keep one copy in the audit doc and point to it. Audit lines 51 and 63 are identical bullets. Replace the pull/5854#discussion_r… links (2 in maps.scala, 4 in the suite) with issue links. Main code has only one other link to a PR review thread.

Questions

• Floating-point keys: map lookups decline them always (MapKeySupport), but the builders decline them only under strictFloatingPoint. Both divergences from Spark are silent. Is the difference intentional?
• Key text in DUPLICATED_MAP_KEY: the key is rendered by ScalarValue's Display, not by Spark's toString of the internal value. A DOUBLE key prints 1 instead of Spark's 1.0, and a DATE key prints 2024-01-01 instead of the day count. Spark's own checkError(parameters = …) tests would catch this. Is that acceptable?
• Upstream issue: would you file a datafusion-spark issue for the missing NULL-key check and for a typed duplicate-key error, and link it from a TODO? Until it's fixed, the two list builders could treat an upstream [DUPLICATED_MAP_KEY] as a signal and re-run check_keys_in_order on that rare path to find the key. That would drop the parsing of the unquoted message.

Assessment

• Scope:
  • Right-sized for the fix, apart from NullGuardSupport being added ahead of #5867.
  • The planner change passes the session options to every UDF built by create_scalar_function_expr. Comet's session differs from the defaults only in: target partitions, batch size, parquet pushdown, the skip-partial-aggregation threshold, the dedup policy, and spark.comet.datafusion.* pass-through.
  • Of those, only the dedup policy is read by any scalar kernel, though a pass-through override would now reach UDFs too.
• Spark compatibility: I checked ArrayBasedMapBuilder and MapFromEntries in Spark 3.4.3, 3.5.8 and 4.1.1.
  • Matches Spark: NULL keys, the dedup policy, the first-position order under LAST_WIN, and reading the policy per task outside whole-stage codegen.
  • Documented gaps: re-execution inside whole-stage codegen, and floating-point keys on 4.0+.
  • Undocumented gaps: item 2 and the key rendering.
• Tests:
  • Apart from item 1, the new tests look like they would fail if the fix were reverted.
  • ci.md recommends the run-spark-4.1-tests label for serde and planner changes, and it hasn't been applied.
  • The area:writer, area:shuffle and area:Iceberg labels look left over from an earlier revision.
• Performance: there's no benchmark, which is fine for a correctness fix. Items 3 to 5 add per-batch costs on the common path. I estimated those from the code and did not measure them.

Already raised by others and not repeated here: the stale test comment, the "ahead of any duplicate-key check" wording, the strictFloatingPoint fixture, and deleting #5846's map_from_arrays.rs and "map" arm on rebase.

[andygrove]

This is a light fully automated review since there are so many PRs open.

The floating point note at spark/src/main/scala/org/apache/comet/serde/maps.scala:144, and the known limitation at docs/source/contributor-guide/expression-audits/map_funcs.md:53, say Spark 3.4 and 3.5 already match because they do not normalize. I think that only holds for a top-level FLOAT or DOUBLE key. For a struct or array key type, ArrayBasedMapBuilder on every version from 3.4.3 through 4.1.1 keys its dedup TreeMap on TypeUtils.getInterpretedOrdering, and the double ordering there is SQLOrderingUtil.compareDoubles, which treats -0.0 and +0.0 as equal. The upstream map_deduplicate_keys keys a HashMap on ScalarValue, whose struct and list equality compares the float bits, so it keeps them apart. With a column ks array<struct<a: double>> holding array(named_struct('a', -0.0D), named_struct('a', 0.0D)), map_from_arrays(ks, array(1, 2)) raises DUPLICATED_MAP_KEY in Spark and returns two entries in Comet. Under LAST_WIN Spark returns one entry, and before this change that policy was declined, so this case used to be correct. Could the note scope the 3.4 and 3.5 claim to top-level keys and call out nested keys on every version? Or, since MapKeySupport.keySupport already declines a floating point type at any nesting level, would it make sense for MapBuilderSupport.keySupport to decline a complex key type that contains a float or double?

[andygrove]

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

[rich7420]

@peterxcli please resolve the conflicts

[andygrove]

I took a deeper look at this against branch-1.1 and the Spark sources for 3.4.3 through 4.2.0. Most of what changes for users is intended and documented: str_to_map stops failing under LAST_WIN, and NULL keys, duplicate keys and length mismatches now raise Spark's own errors. A few things need another look before this merges:

1. spark/src/main/scala/org/apache/comet/serde/maps.scala:171

Thanks for working through all the earlier rounds. Reading the policy per task looks right to me now. One thing about LAST_WIN with float keys. In 1.1 a LAST_WIN query always got Spark's answer, because map_from_arrays fell back and map_from_entries went through the dispatcher. With this change a FLOAT or DOUBLE key runs natively, and the native dedup compares raw bits. Spark merges more keys than that, and not only on 4.0+. On every version a struct or array key goes into a TreeMap ordered by SQLOrderingUtil.compareDoubles, so named_struct('a', 0.0D) and named_struct('a', -0.0D) are one key. On 3.4 and 3.5 a top-level double is boxed into a HashMap, and Double.equals folds every NaN into one.

The nested cases are the floating-point edge cases the user guide already accepts by default behind spark.comet.exec.strictFloatingPoint, and the decline under that flag covers them, so I'm fine with that policy here. But the notes at maps.scala:144 and map_funcs.md:53 and :64 say Spark 3.4 and 3.5 already match Comet, and they don't for nested keys or for top-level NaNs with different bits. Could you correct those notes, and add a few map_from_arrays and map_from_entries cases with float keys to CometFloatSemanticsSuite, so the behavior is pinned?

2. spark/src/main/scala/org/apache/comet/serde/maps.scala:196

The decline makes sense for a nullable keys expression. When expr.left.nullable is false, though, the guard is true on every row, so both copies of the keys see the same rows. map_from_arrays(array(monotonically_increasing_id()), array(2)) ran natively in 1.1 and matched Spark, and here it falls back. 1.1 could drift when the values array had NULLs, and this PR fixes that, because the guard now only tests the keys. Could the serde skip the guard when the keys can't be NULL, and decline a nondeterministic keys expression only when it emits the guard? That also saves evaluating the keys twice for the common map_from_arrays(array(...), ...) shape.

3. native/spark-expr/src/map_funcs/map_builders.rs:167 (performance)

This swaps the kernel behind map_from_arrays on the default path, from DataFusion's MapFunc to datafusion-spark's map_deduplicate_keys. The new kernel builds a ScalarValue per key, allocating a String for every string key, and looks each one up in a HashMap. I can't tell from the code which side is faster, since the old path had its own per-row slices and concat. Could you add a criterion bench under native/spark-expr/benches/, with 8192 rows of Int32 and Utf8 keys at 1, 8 and 64 entries per row, and post the numbers for the old wrapper and the new one, as optimizing_expressions.md describes?

4. docs/source/user-guide/latest/expressions.md:412

This row still says LAST_WIN goes through the dispatcher. After this PR it runs natively, and it's collated keys, and floating-point keys under spark.comet.exec.strictFloatingPoint, that dispatch. Could the map_from_entries row and the map_from_arrays row at :411 say what routes where now? This page is hand-edited, so it won't pick up the serde changes on its own.

5. General

The floating-point behavior here differs between Spark 3.x and 4.x, and only the Spark 4.1 Comet suites ran on this head. Could you apply run-all-spark-profiles after the rebase?

[andygrove]

Thanks for the thorough work here, and for adding the bench. The null key, duplicate key, collation and policy timing handling all look right to me now. I compared the PR against branch-1.1 and found two things that I think need to change before this merges.

1. map_from_arrays is slower on the default path

I ran the bench shapes from this PR against the merge base with the same inputs: 8192 rows, no duplicate keys, release profile.

Keys per row	Int32 old	Int32 new	Utf8 old	Utf8 new
1	974 µs	245 µs	1.06 ms	380 µs
8	1.34 ms	2.15 ms	1.54 ms	3.34 ms
64	2.87 ms	17.2 ms	3.88 ms	26.4 ms

The new path wins at one key per row, but it is about 1.6x to 2.2x slower at 8 keys and about 6x to 7x slower at 64 keys. This is the default EXCEPTION path, so every map_from_arrays user pays for it. I suspect the per-key ScalarValue and HashMap work in the upstream map_deduplicate_keys, but I did not profile it. Could you look at where the time goes? Would a typed fast path for primitive and string keys, or an upstream improvement, bring 8 and 64 entries back to the old numbers? Please post before and after numbers in the PR.

2. LAST_WIN with a float key now diverges from Spark by default

In 1.1, LAST_WIN always got Spark's answer. After this PR a top-level FLOAT or DOUBLE key runs natively and compares raw bits. On Spark 4.0 and later -0.0 and +0.0 are one key in Spark, and NaNs with different bits are one key on every version. map_from_entries also stores 0.0 where Comet stores -0.0. CometFloatSemanticsSuite records these as KnownGaps for #6549, and a known gap on a default path needs a fallback. Nested float keys are already Incompatible by default, so the top-level case looks inconsistent with that.

docs/source/contributor-guide/floating_point.md on main (it landed with #6447) says an expression that does not follow Spark's float rule should report Incompatible for floating-point input until it does. Could MapBuilderSupport.keySupport do that for top-level float keys until #6549 is fixed? That would also cover the EXCEPTION divergence that already exists in 1.1. Normalizing the key natively with the float_semantics helpers would work too. Either way the expressions.md rows and the audit will need a small update.

3. Smaller asks

• The whole-stage codegen re-execution difference in map_funcs.md is the only known limitation without an issue link. Could you file one and link it?
• Could you rebase and apply run-all-spark-profiles? Float behavior differs between Spark 3.x and 4.x, and only Spark 4.1 ran. The only conflict I found is in native/core/src/execution/jni_api.rs.

I'm requesting changes for items 1 and 2. Happy to re-review once they are addressed.

[sunchao]

Reviewed the full 30-file diff at 58e00bc6baea7e4772332825914e55914d6eecf1 against 3bc2faa934f04799686705b952d59ed32a707dda. The PR is not a draft. Applied review-comet-pr, review-comet-expression-pr, and review-comet-ffi-pr, and considered the existing discussion and review threads.

No additional introduced P1/P2 issues found within this review. The already-reported floating-point correctness and wider-map performance concerns remain unresolved, so this is not a clean verdict. Those concerns are not duplicated as new findings.

Summary

• Prior state and problem: Map construction did not consistently enforce Spark’s null-key checks and structured errors. LAST_WIN caused fallback or dispatch for the array constructors, while native str_to_map still used the default duplicate policy.
• Design approach: Reuse datafusion-spark kernels through three wrappers. Forward Spark’s duplicate policy into the native session, supply session options to scalar expressions, and add the missing validation and error conversion.
• Correctness: Checked length validation, duplicate/null error precedence, null input arrays and entries, first-position ordering under LAST_WIN, scalar broadcasting, sliced lists, and left-before-right evaluation against Spark sources. The existing top-level floating-point issue remains: native bitwise equality separates signed zeros on Spark 4.x and distinct NaN payloads on every supported version. The new KnownGap cases explicitly confirm these divergences.
• Compatibility analysis: Compared Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0 sources and the affected release paths on branch-1.1 at e9efd9f764ee0a59b7898ff028d6985d4a7a28e1. Native LAST_WIN and structured errors are intended changes. Removing the previous LAST_WIN fallback exposes the already-reported floating-point regression. Collated and nested floating-point keys now decline by default.
• Key design decisions: Reading the policy per native task plan matches Spark outside whole-stage codegen. The documented repeated-Dataset difference inside whole-stage codegen remains, as discussed previously. The nullable-key guard preserves short-circuiting, and excluding nullable nondeterministic keys prevents independent stateful evaluations.
• Implementation sketch: CometExecIterator serializes the policy, JNI parses it into session options, and the planner passes those options to scalar functions. map_builders.rs validates relevant null keys, rebases sliced list offsets without copying values, delegates construction, and translates errors. Registration replaces the former map implementation.
• Performance: Omitting the guard for non-nullable keys avoids duplicate evaluation. Ordinary batches avoid the null-key validation walk, and slice rebasing shares value buffers. However, the existing exact-head review reports approximately 1.6–2.2× slower construction at eight entries and 6–7× at 64 entries for 8,192-row batches. The upstream per-key ScalarValue and hashing path remains unchanged. I did not independently rerun those benchmarks.
• Design: Wrapping upstream kernels is a reasonable, focused approach. Sharing session configuration fixes the underlying policy propagation problem. The JNI changes do not alter Arrow ownership, buffer lifetimes, or batch-transfer mechanisms.
• Abstraction & complexity: The shared list-builder invocation and validation helpers serve both constructors appropriately. Error-message parsing is an upstream coupling, with tests pinning the expected messages. No additional abstraction issue meeting the P1/P2 bar was identified.
• Behavioral changes worth calling out: Eligible constructors now run natively under LAST_WIN, retain the first key position with the last value, and reject invalid keys through Spark errors. Nullable nondeterministic keys fall back. Top-level floating-point keys remain native unless strict mode is enabled, which leaves the existing correctness blocker.
• Suggested improvements: Address the existing floating-point concern through normalization or default fallback, and resolve the measured wider-map regression with before/after benchmarks. The previously requested Spark-profile and Spark SQL coverage is still outstanding. No additional P1/P2-level improvement is proposed.

Exact-head CI: 25 successful checks and 15 skipped, including successful Required Checks. Run 37257017863 tested merge revision 49e8d8a5af2e7e0b3ab7e20d1115d1d84417aee9. The map builders, serde, policy forwarding, and relevant regression cases match the reviewed head. It passed 2,079 native tests and 2,098 Spark 4.1 expression tests. The expression suite also recorded one unrelated cancellation and 12 ignored tests. Spark SQL suites and non-default Spark execution profiles were not exercised.

Local validation: All 18 map-builder tests passed using cargo test --offline --locked -p datafusion-comet-spark-expr map_funcs::map_builders --no-default-features, with the locked DataFusion 55.1.0 dependencies. No local end-to-end Spark/JNI run or independent performance benchmark was performed. Project source was unchanged.