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