Repository navigation
Conversation
…onstruction `map_from_arrays` and `map_from_entries` built their maps without the entry checks Spark's `ArrayBasedMapBuilder` performs, so a `NULL` key inside the keys array produced a map with a `NULL` key instead of raising `NULL_MAP_KEY`, and `spark.sql.mapKeyDedupPolicy=LAST_WIN` fell the whole expression back to Spark. DataFusion 55 added `datafusion.spark.map_key_dedup_policy` and taught the `datafusion-spark` map kernels to follow it, which is the missing half. Forward Spark's `spark.sql.mapKeyDedupPolicy` to it across JNI, and pass the session's `ConfigOptions` into `ScalarFunctionExpr` so a kernel that reads a setting sees the session's value rather than DataFusion's defaults. New `SparkMapFromArrays` / `SparkMapFromEntries` / `SparkStrToMap` wrappers add the checks the upstream kernels do not perform and restate their errors as the Spark error classes `SparkErrorConverter` turns back into `QueryExecutionErrors`: a `NULL` key raises `NULL_MAP_KEY` ahead of any duplicate-key check, key and value arrays of different lengths raise `MAP_KEY_VALUE_DIFF_SIZES`, and a duplicate key under `EXCEPTION` raises `DUPLICATED_MAP_KEY` naming the key. `CometMapFromArrays` now emits `map_from_arrays`, which is null intolerant like Spark's, so the `CaseWhen` guard against NULL input arrays is no longer needed. A floating-point map key stays a documented difference: Spark normalizes `-0.0` to `+0.0` and canonicalizes `NaN` before storing a key, while the native builders compare the raw Arrow values. `spark.comet.exec.strictFloatingPoint` declines those key types. Closes apache#4680
github-actions
Bot
added
bug
peterxcli
changed the title
…d-dedup-policy # Conflicts: # native/spark-expr/src/comet_scalar_funcs.rs # native/spark-expr/src/lib.rs # native/spark-expr/src/map_funcs/mod.rs
| self.inner | ||
| .invoke_with_args(args) | ||
| .map_err(|error| as_spark_error(error, DuplicateKeyFormat::Bare)) |
There was a problem hiding this comment.
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.
| if let Some(nulls) = &key_nulls { | ||
| if nulls.slice(start, end - start).null_count() > 0 { | ||
| return Err(SparkError::NullMapKey.into()); | ||
| } |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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
left a comment
andygrove
left a comment
There was a problem hiding this comment.
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.
The upstream `datafusion-spark` map kernels read each row's entries at its own
offset but build the mask selecting the surviving keys from zero, then apply
that mask to the list's whole values array. Arrow's `filter` accepts a predicate
shorter than the array it filters, so on a sliced argument the mismatch silently
returns keys belonging to earlier rows rather than raising: keys `[[10], [20]]`
and values `[[100], [200]]`, both sliced to the second row, built `{10: 200}`
instead of `{20: 200}`. A `LIMIT` above a projection produces such an argument.
Compact any list argument whose values hold more than its offsets address
before validating or delegating, so the kernels see the layout they assume.
`map_from_entries` reached the same helper before this branch, so the bug is not
new to `map_from_arrays`; a fix belongs upstream as well.
Reported by @rich7420.
Spark's `ArrayBasedMapBuilder` inserts entries one at a time, so for keys `[1, 1, NULL]` under `EXCEPTION` it raises `DUPLICATED_MAP_KEY` on the second entry and never reaches the null. The validation pre-scanned a whole row for null keys before delegating, so it reported `NULL_MAP_KEY` instead, and its comments described the precedence as categorical rather than positional. Walk each row's keys in insertion order and raise on the first offending entry, so the two errors order the way Spark orders them, across rows as well as within one. The walk runs only when the keys carry a `NULL`: without one the kernel's own duplicate check already names the key Spark would. Under `LAST_WIN` a duplicate overwrites rather than raising, so only the null check applies. Reported by @rich7420.
`ArrayBasedMapBuilder` keys its dedup map on `TypeUtils.getInterpretedOrdering` once the key type contains a string, so under `UTF8_LCASE` the keys 'a' and 'A' are one key. The native builders compare the raw Arrow bytes and would keep both, missing the duplicate Spark reports or the overwrite Spark performs under `LAST_WIN`. `MapKeySupport` already declines a collated key for `map_extract` for the same reason; `MapBuilderSupport` only gated floating-point keys. Report `Incompatible` for a collated key type in both constructors. `CometMapFromArrays` falls back to Spark, while `CometMapFromEntries` mixes in `CodegenDispatchFallback` and stays in the Comet pipeline running Spark's own generated code. The new fixture pins both routes. Reported by @andygrove.
The note claimed Spark normalizes a floating-point map key before storing it, full stop. Two corrections, both checked against Spark's sources: `ArrayBasedMapBuilder` gained `keyNormalizer` in 4.0, alongside `spark.sql.legacy.disableMapKeyNormalization`. The 3.5 builder has no normalizer and no reference to `NormalizeFloatingNumbers`, so on 3.4 and 3.5 the native builders already match Spark and there is nothing to warn about. On 4.0+ the two functions differ. `MapFromArrays` calls `ArrayBasedMapBuilder.from`, which returns the input arrays untouched when no key repeated, so a lone `-0.0` key stays `-0.0` in Spark as it does natively; only duplicate detection diverges. `MapFromEntries` puts entries one at a time and always calls `build()`, so Spark stores the normalized key and returns `+0.0` where Comet returns `-0.0`. The gate stays unconditional. Declining on 3.4 and 3.5 costs only a fallback that `spark.comet.exec.strictFloatingPoint` users opted into. Reported by @andygrove.
…gines The length mismatch test avoided naming Spark's condition because I assumed the `_LEGACY_ERROR_TEMP_*` number moved between Spark versions, and added `checkSparkErrorParity` to `CometTestBase` to work around it. The assumption was never checked and is wrong: `mapDataKeyArrayLengthDiffersFromValueArrayLengthError` raises `_LEGACY_ERROR_TEMP_2128` in 3.4.3, 3.5.8 and 4.1.3 alike. Name the condition in the test and drop the helper, which leaves `CometTestBase` untouched by this branch. Reported by @andygrove.
6861d8b to
0021e46
Compare
|
@andygrove @rich7420 addressed review! ptal again, thanks! |
…tures `routing_map_legacy_disabled.sql` and `routing_map_legacy_enabled.sql` arrived with apache#5918 and pin how `map_from_entries` routes under `spark.sql.mapKeyDedupPolicy=LAST_WIN`. They encode the behavior this branch removes: `MapFromEntries` reported `Incompatible` under `LAST_WIN`, so `spark.comet.exec.scalaUDF.codegen.enabled` decided whether it fell back to Spark or ran through the JVM codegen dispatcher. The native builder now reads the policy from `datafusion.spark.map_key_dedup_policy`, so the expression is `Compatible` and stays native under either setting of that flag. Expect native in both fixtures. No routing coverage is lost. `map_from_entries` is still `Incompatible` for a `BinaryType` key or value, and `routing_maps_disabled.sql` and `routing_maps_enabled.sql` exercise its fallback and dispatch routes that way. `str_to_map` keeps its expectations in both fixtures: it declines for `spark.sql.legacy.truncateForEmptyRegexSplit`, which this branch does not touch.
| (builder, binaryExpr) => builder.setAnd(binaryExpr)) | ||
| // Native `map_from_arrays` is null intolerant like Spark's: a NULL keys or values array | ||
| // yields a NULL map for that row, so no CaseWhen guard is needed here. | ||
| scalarFunctionExprToProto("map_from_arrays", keysExpr, valuesExpr) |
There was a problem hiding this comment.
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.
…d-dedup-policy # Conflicts: # native/spark-expr/src/comet_scalar_funcs.rs
…rcuit Spark's `BinaryExpression.eval` returns NULL as soon as the left input is NULL and never evaluates the right one, so a failing cast in the values argument does not run for a row whose keys array is NULL. Comet evaluates both argument subtrees, so removing the `CaseWhen` guard made `map_from_arrays(k, array(CAST(v AS INT)))` raise `CAST_INVALID_INPUT` under ANSI where Spark returns NULL. Restore the guard. Its `AND` over `IsNotNull` on both arguments lets the native side skip the values expression once the keys array is known NULL, which matches Spark's evaluation order. The earlier commit removed it on the grounds that native `map_from_arrays` is null intolerant; that is true of the result but says nothing about which subtrees get evaluated. Add a regression test that fails with `CAST_INVALID_INPUT` without the guard and passes with it.
|
Triage note: #5844 also makes I think the native route is the one we want here, provided DataFusion's kernel matches |
…the values expression The previous commit restored `CASE WHEN keys IS NOT NULL AND values IS NOT NULL` around `map_from_arrays`, and its regression test passed only because the table held a single row. DataFusion's `AND` skips its right side when the left side is false on every row of the batch, or on at most a fifth of them; in a batch where most rows do have keys it evaluates `values IS NOT NULL` on the whole batch, so a failing cast in the values array still runs for the NULL-keys row and raises under ANSI where Spark returns NULL. Nest one `CaseWhen` per argument instead, as apache#5846 does. DataFusion evaluates a THEN branch only on the rows its WHEN selected, so the values expression is never evaluated for a row whose keys array is NULL, which is what `BinaryExpression.eval` does. Rewrite the regression test as a five-row table written in one partition so every row shares a batch and most rows have keys. It fails with `CAST_INVALID_INPUT` against the `AND` guard and passes with the nested guards; the three `map_from_arrays` tests from apache#5846 pass as well.
…batch validation The map_from_arrays validator checked every row's key and value lengths before the kernel looked for a duplicate key, so a duplicate in row 0 followed by a length mismatch in row 1 raised _LEGACY_ERROR_TEMP_2128 where Spark raises DUPLICATED_MAP_KEY. The kernel already checks lengths and duplicates row by row, in Spark's order, so both validators now return early unless a row that gets built holds a NULL key; that row is bound to fail, so only then are the rows up to it walked, lengths first. The kernel's length-mismatch message is translated to SparkError::MapKeyValueDiffSizes like its duplicate-key message, with a pin test. map_from_entries first rules out the common case with one AND and a popcount (a NULL key only under NULL entries), and the walk reuses the duplicate from HashSet::replace instead of cloning every key. The sliced-list workaround now rebases only a list whose first offset is nonzero, zero-copy (shifted offsets and a values slice), links apache/datafusion#25419, and says to delete it once Comet picks up apache/datafusion#25431. An all-scalar call is evaluated once and returned as a scalar, as the kernel alone does, instead of building number_rows rows. Cleanups: derive Default for the wrappers and drop their unused new(), read key nulls with logical_nulls(), drop the unreachable LargeList and FixedSizeList arms and row-count check, drop the clippy allow (clippy is clean without it), parse the policy into the typed ConfigOptions field instead of set_str, and inline session_config_options() at its only caller. The Rust tests that only exercised the upstream kernel are gone; new ones pin the orderings above, a NULL key in a row whose values are NULL or that holds a NULL entry, and the all-scalar result.
The native kernel returns a NULL map for a NULL values array, and Spark evaluates the values whenever the keys are non-NULL, so the inner `CASE WHEN values IS NOT NULL` no longer did anything but evaluate the values twice. The serde now emits `CASE WHEN keys IS NOT NULL THEN map_from_arrays(keys, values) END`, with no ELSE: create_case_expr wraps an ELSE in a Cast, which hides it from DataFusion's NULL-ELSE shortcut. With the values serialized once and evaluated only on the rows Spark evaluates them on, only a nondeterministic keys expression needs declining. The check moves inline into CometMapFromArrays, since NullGuardSupport had this one caller and described dispatcher routing it does not have. The nondeterministic-child fixture now runs a stateful values expression natively against Spark. Both constructors list the strictFloatingPoint decline in getIncompatibleReasons(), and the floating-point note no longer says the decline falls back to Spark, since map_from_entries runs Spark's code through the JVM dispatcher instead.
…cated map coverage The two dedup-policy timing tests checked `df.select(...)`, a new QueryExecution, so they never re-ran `df` and would have passed with the policy fixed on the first execution. They are now one test that plans `df` under one policy, collects `df` itself under the other and again under the first, in both directions, outside whole-stage codegen, for Spark and Comet. The error half uses checkSparkError's structuredError, now a protected CometTestBase helper, so a CometNativeException anywhere in the cause chain fails it. New coverage: map_builders_strict_fp.sql pins where each constructor goes for a DOUBLE key under spark.comet.exec.strictFloatingPoint (map_from_arrays falls back, map_from_entries dispatches, a DOUBLE value stays native); an end-to-end LIMIT/OFFSET test for the sliced-list rebase, which fails without it; fixture queries for a duplicate ahead of a NULL key; and str_to_map.sql matches the converted "Duplicate map key a was found" rather than the error class, which the raw upstream message also carries. Removed as duplicates: the Scala nondeterministic-child test and the matching query in map_from_arrays_dedup_policy.sql, the LAST_WIN halves of the duplicate-key tests, and the "null entry" test (now two table rows in map_from_entries.sql). The dedup-policy fixtures keep one literal query for the all-scalar path and assert expect_native on the query that pins entry order, which also replaces the map_from_entries checks in the routing_map_legacy_* files.
The audit said a NULL key raises NULL_MAP_KEY ahead of any duplicate-key check; the builders report whichever of a length mismatch, a NULL key and a duplicate comes first in the first failing row, as Spark does. A length mismatch reaches users as _LEGACY_ERROR_TEMP_2128, not MAP_KEY_VALUE_DIFF_SIZES. The guard description follows the single keys guard, and the nondeterministic-child limitation now covers only the keys. The map_from_entries bullet that repeated map_from_arrays' now points to it, and policy forwarding is described once, which the code comments point to. The floating-point notes say where each constructor goes under spark.comet.exec.strictFloatingPoint and that NaNs only stay apart natively when their bit patterns differ.
|
This is a light fully automated review since there are so many PRs open. The floating point note at |
# Conflicts: # docs/source/contributor-guide/expression-audits/map_funcs.md # native/spark-expr/src/comet_scalar_funcs.rs # spark/src/main/scala/org/apache/comet/serde/maps.scala # spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala
For a struct or array key type, ArrayBasedMapBuilder finds duplicate keys with interpreted ordering on every Spark version, which treats -0.0 and +0.0 (and all NaNs) as equal at any nesting level. The native builders compare the nested values bit by bit, so map_from_arrays and map_from_entries kept both keys where Spark raises DUPLICATED_MAP_KEY, or keeps one entry under LAST_WIN. Report such key types as Incompatible: map_from_arrays falls back to Spark and map_from_entries goes through the codegen dispatcher, with the allowIncompatible opt-ins unchanged. Scope the floating-point note and the map_funcs audit to top-level keys, where Spark 3.4 and 3.5 still treat NaNs with different bit patterns as one key, and add fixtures for both dedup policies.
|
I would like to do a deeper review of this PR before it is merged. |
# Conflicts: # native/spark-expr/src/comet_scalar_funcs.rs
|
@peterxcli please resolve the conflicts |
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
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?
# Conflicts: # native/spark-expr/src/lib.rs # native/spark-expr/src/map_funcs/map_from_arrays.rs # native/spark-expr/src/map_funcs/mod.rs
…nt fields The bench added on main by apache#6541 called return_type, which main's MapFunc-backed SparkMapFromArrays answers. This branch backs it with datafusion-spark's MapFromArrays, which only implements return_field_from_args, so the bench panicked when run. Build the argument fields, take the return field from return_field_from_args, and pass both to invoke_with_args, as DataFusion's planner does.
The CASE WHEN keys IS NOT NULL guard serializes the keys a second time. When expr.left is not nullable, the guard selects every row, so the serde now emits the map_from_arrays call alone. It declines a nondeterministic keys expression only when it emits the guard. Thus map_from_arrays(array(monotonically_increasing_id()), array(2)) runs natively again, as in 1.1, and the keys are evaluated once.
…sion The native map builders compare a FLOAT or DOUBLE key by its raw bits. Spark treats NaNs with different bit patterns as one key on every version: Spark 3.4 and 3.5 through Double.equals in the builder's HashMap, and Spark 4.0 and later through the key normalizer. Only 4.0 and later also merge -0.0 and +0.0, and only map_from_entries stores the normalized key when no key repeats. Rewrite the floating-point key note and the map_funcs audit to say so, and to say that a key that repeats under LAST_WIN shows the difference. Refer to apache#6549 for the remaining work.
Add map_from_arrays and map_from_entries cases with FLOAT and DOUBLE keys. Under LAST_WIN, each pair of keys that Spark merges becomes one entry, so the answer shows the merge. The pairs are signed zeros, NaNs with different bit patterns, and struct keys that wrap them. One more case per builder stores a single key under the default policy. The known gaps refer to apache#6549: signed zeros and the key that map_from_entries stores on Spark 4.0 and later, and NaN pairs on every version. The struct-key cases match, because the builders decline those key types. Case gains a conf field for the per-case SQL settings.
Both rows in the expressions table now say that both spark.sql.mapKeyDedupPolicy values run natively. They also list the key types that fall back or go through the codegen dispatcher: collated string keys, struct or array keys with floating-point fields, and floating-point keys under spark.comet.exec.strictFloatingPoint. For map_from_arrays, a nullable nondeterministic keys array falls back too.
…ries The benchmark now builds 8192 rows of Int32 and Utf8 keys at 1, 8 and 64 entries per row, with Int32 values and no duplicate key. The same file compiles against the MapFunc wrapper on main, so it compares the two kernels. The config is built once, outside the timed loop.
# Conflicts: # spark/src/test/scala/org/apache/comet/CometFloatSemanticsSuite.scala
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
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.mdis 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 innative/core/src/execution/jni_api.rs.
I'm requesting changes for items 1 and 2. Happy to re-review once they are addressed.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
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_WINcaused fallback or dispatch for the array constructors, while nativestr_to_mapstill used the default duplicate policy. - Design approach: Reuse
datafusion-sparkkernels 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 newKnownGapcases 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.1ate9efd9f764ee0a59b7898ff028d6985d4a7a28e1. NativeLAST_WINand structured errors are intended changes. Removing the previousLAST_WINfallback 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:
CometExecIteratorserializes the policy, JNI parses it into session options, and the planner passes those options to scalar functions.map_builders.rsvalidates 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
ScalarValueand 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.
|
Hi @peterxcli let me know if I can help out with this one |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Reviewed commit f3de2f78546ca8941a49605ce329c22dfc2fd34d. New P1/P2 findings are detailed inline.
| statement | ||
| INSERT INTO test_map_builders_strict_fp VALUES (1.0D, 1), (double('-0.0'), 2), (2.5D, 3) | ||
|
|
||
| query expect_fallback(Map construction on a floating-point key) |
There was a problem hiding this comment.
[P2] [P2] Expect codegen dispatch in the new map routing fixtures. With the default spark.comet.exec.scalaUDF.codegen.enabled=true, CometMapFromArrays mixes in CodegenDispatchFallback, so this strict-floating-point query executes Spark’s generated code inside CometProject. It does not produce the operator fallback required by expect_fallback, and the test fails despite returning the correct answer. The same mismatch affects map_builders_collation.sql, map_builders_nested_fp.sql, map_builders_nested_fp_last_win.sql, and map_from_arrays_nondeterministic_child.sql. These five failures independently keep required expression CI red. Please use expect_dispatch(map_from_arrays) for these cases and align their routing explanations.
Evidence: Exact-head expression CI job https://github.com/apache/datafusion-comet/actions/runs/39112998076/job/117639227900 fails this query with Expected fallback reason 'Map construction on a floating-point key' but no fallback reasons were found, showing CometProject and codegen-dispatched execution. The four sibling fixtures fail with the corresponding missing-fallback assertion. Verified that all five fixtures, maps.scala, QueryPlanSerde.scala, and the test harness match HEAD in the tested merge revision 62dc23651a2a23cf14ad402cbc3008767c03af47. With the native library built, the focused reproduction is ./mvnw test -Dtest=none -Dsuites="org.apache.comet.CometSqlFileTestSuite map_builders_strict_fp".
6 participants
Which issue does this PR close?
Rationale for this change
Spark builds every map through
ArrayBasedMapBuilder, which refuses aNULLkey and resolves duplicate keys according tospark.sql.mapKeyDedupPolicy. Comet'smap_from_arraysandmap_from_entriesdid neither. ANULLinside the keys array produced a map with aNULLkey instead of an error, and setting the policy toLAST_WINpushed the whole expression back to Spark.DataFusion 55 supplies what was missing. Its
datafusion.spark.map_key_dedup_policyoption takes the sameEXCEPTIONandLAST_WINvalues as the Spark config, and thedatafusion-sparkmap kernels already follow it. Once Comet passes the setting through,LAST_WINruns natively, and the remaining checks thatArrayBasedMapBuilderperforms cost only a few lines on top.What changes are included in this PR?
spark.sql.mapKeyDedupPolicynow crosses JNI.CometExecIterator.serializeCometSQLConfssends it explicitly, sincecometSqlConfscarries only keys underspark.comet., andprepare_datafusion_session_contextapplies it to the session asdatafusion.spark.map_key_dedup_policy.A second change was needed before that setting could reach a kernel at all.
create_scalar_function_exprhanded everyScalarFunctionExpra freshConfigOptions::default(), so any kernel reading a session option saw DataFusion's defaults. It now passes the session's ownConfigOptions.native/spark-expr/src/map_funcs/map_builders.rsadds three wrappers,SparkMapFromArrays,SparkMapFromEntriesandSparkStrToMap. Each calls the matchingdatafusion-sparkkernel, adds the checks that kernel skips, and translates its errors into the Spark error classes thatSparkErrorConverterconverts back intoQueryExecutionErrors:NULLkey raisesNULL_MAP_KEY, before any check for duplicates, matching the order Spark applies them;MAP_KEY_VALUE_DIFF_SIZES;EXCEPTIONraisesDUPLICATED_MAP_KEYand names the key.str_to_mapneeds only the last of these, because splitting a string never yields aNULLkey. Passing the config through also fixed itsLAST_WINcase, which used to raise an error where Spark returns a map.CometMapFromArraysnow emitsmap_from_arrays. It used to emit the genericmapwrapped inCaseWhen(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 theirLAST_WINIncompatiblebranch.One difference with Spark remains.
ArrayBasedMapBuildernormalizes a floating point key before storing it, so-0.0becomes+0.0and everyNaNcollapses into one. The native builders compare the raw Arrow values, so a map built from both-0.0and+0.0keeps two entries where Spark reports a duplicate key. The compatibility notes record this, andspark.comet.exec.strictFloatingPointmakes 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
CometMapExpressionSuiterun each case through both engines and compare the exception type, error class and SQLSTATE, along with the answers each engine returns underLAST_WIN.Among the SQL fixtures, the two
*_dedup_policy.sqlfiles used to assert theLAST_WINfallback and now assert native execution.map_from_arrays.sql,map_from_entries.sqlandstr_to_map.sqlgained theEXCEPTIONerror cases, andstr_to_map_dedup_policy.sqlis new. That also retires theTODO: Add LAST_WIN policy tests when spark.sql.mapKeyDedupPolicy config is supportednote instr_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, soCometTestBase.checkSparkErrornow builds on a newcheckSparkErrorParityhelper.The
ConfigOptionschange affects every scalar function, so the full 487-fixture suite ran green as well.