sunchao commented on code in PR #5854:
URL: https://github.com/apache/datafusion-comet/pull/5854#discussion_r4043896247
##########
spark/src/main/scala/org/apache/comet/serde/maps.scala:
##########
@@ -132,41 +133,92 @@ object CometMapExtract extends
CometExpressionSerde[GetMapValue] {
}
}
-private object MapKeyDedupPolicySupport {
- val incompatibleReason: String =
- s"`${SQLConf.MAP_KEY_DEDUP_POLICY.key}` is set to " +
- s"`${SQLConf.MapKeyDedupPolicy.LAST_WIN}`; Comet's native map
construction " +
- "does not implement LAST_WIN dedup semantics."
-
- val nullKeyReason: String =
- "Spark rejects a `NULL` element inside the keys array with a
`RuntimeException`" +
- " (`Cannot use null as map key`); Comet's native `map_from_arrays` /
`map_from_entries`" +
- " does not detect a per-element `NULL` key and produces a map with a
`NULL` key instead" +
- " ([#4680](https://github.com/apache/datafusion-comet/issues/4680))."
-
- def isLastWin: Boolean =
- SQLConf.get
- .getConf(SQLConf.MAP_KEY_DEDUP_POLICY)
- .toString
- .equalsIgnoreCase(SQLConf.MapKeyDedupPolicy.LAST_WIN.toString)
+/**
+ * Shared gate for the native map constructors (`map_from_arrays`,
`map_from_entries`), which
+ * reproduce Spark's `ArrayBasedMapBuilder`: they reject a `NULL` key with
`NULL_MAP_KEY` and
+ * follow `spark.sql.mapKeyDedupPolicy`, whose value Comet forwards to the
native session as
+ * `datafusion.spark.map_key_dedup_policy`.
+ */
+private object MapBuilderSupport {
+
+ /**
+ * Floating-point keys differ from Spark only on 4.0 and later, and
differently per function.
+ * `ArrayBasedMapBuilder` gained `keyNormalizer` in 4.0 (with
+ * `spark.sql.legacy.disableMapKeyNormalization` to turn it off); 3.4 and
3.5 do not normalize
+ * at all, so the native builders already match there.
+ *
+ * On 4.0+ the normalized key decides duplicates for both functions, so a
map built from both
+ * `-0.0` and `+0.0` is one key in Spark and two natively. What each
function stores then
+ * diverges: `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 too; while
+ * `MapFromEntries` puts entries one at a time and always calls `build()`,
which emits the
+ * normalized keys, so a lone `-0.0` key comes back as `+0.0` in Spark and
as `-0.0` natively.
+ *
+ * A note rather than a decline, because a map keyed on `-0.0` or `NaN` is
rare;
+ * `spark.comet.exec.strictFloatingPoint` declines it for anyone who wants
the guarantee. That
+ * gate is not conditioned on the Spark version: declining on 3.4 and 3.5
costs those users
+ * nothing beyond a fallback they opted into.
+ */
+ val floatingPointKeyNote: String =
+ "On Spark 4.0 and later, `ArrayBasedMapBuilder` normalizes a
floating-point map key before " +
+ "comparing it, so `-0.0` counts as the same key as `+0.0` and all `NaN`s
count as one " +
+ "key. Comet's native map construction compares the raw Arrow values, so
a map built from " +
+ "both `-0.0` and `+0.0` keeps two entries where Spark reports a
duplicate key. " +
+ "`map_from_entries` also stores the normalized key, so Spark returns
`+0.0` for a `-0.0` " +
+ "key where Comet returns `-0.0`; `map_from_arrays` keeps the original
keys in both " +
+ "engines when nothing repeated. Spark 3.4 and 3.5 do not normalize at
all, so they match " +
+ s"Comet already. Set `${COMET_EXEC_STRICT_FLOATING_POINT.key}=true` to
fall back to Spark " +
+ "for a floating-point map key."
+
+ /**
+ * `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 that
+ * Spark reports (or, under `LAST_WIN`, the overwrite Spark performs).
`MapKeySupport` declines
+ * a collated key for `map_extract` for the same reason.
+ */
+ val collationKeyReason: String =
+ "Comet's native map construction compares string keys as `UTF8_BINARY`, so
it cannot honour " +
+ "a non-default collation when it looks for a duplicate key."
+
+ /** The support level for a map constructor whose result has key type
`keyType`. */
+ def keySupport(keyType: DataType): SupportLevel =
+ if (hasNonDefaultStringCollation(keyType)) {
+ Incompatible(Some(collationKeyReason))
+ } else {
+ SupportLevel
+ .strictFloatingPointReason(keyType, "Map construction on a
floating-point key")
+ .map(reason => Incompatible(Some(reason)))
+ .getOrElse(Compatible(None))
+ }
}
object CometMapFromArrays extends CometExpressionSerde[MapFromArrays] {
override def getIncompatibleReasons(): Seq[String] =
- Seq(MapKeyDedupPolicySupport.incompatibleReason)
+ Seq(MapBuilderSupport.collationKeyReason)
override def getCompatibleNotes(): Seq[String] =
- Seq(MapKeyDedupPolicySupport.nullKeyReason)
+ Seq(MapBuilderSupport.floatingPointKeyNote)
- override def getSupportLevel(expr: MapFromArrays): SupportLevel = {
- if (MapKeyDedupPolicySupport.isLastWin) {
- Incompatible(Some(MapKeyDedupPolicySupport.incompatibleReason))
- } else {
- Compatible(None)
- }
- }
+ override def getSupportLevel(expr: MapFromArrays): SupportLevel =
+ MapBuilderSupport.keySupport(expr.dataType.keyType)
Review Comment:
[P2] Preserve fallback for nondeterministic children before enabling LAST_WIN
Could we bring in the nondeterministic-child gate from #5867 before removing
the `LAST_WIN` decline? As noted in the sequencing discussion, the null guards
still serialize each child separately from the map constructor. With default
compatibility settings, this now exposes a case that previously fell back to
Spark:
```sql
SET spark.sql.mapKeyDedupPolicy=LAST_WIN;
SELECT id,
map_from_arrays(
IF(monotonically_increasing_id() % 2 = 0, array(1), NULL),
array(2))
FROM t;
```
With ids 0 through 15 in one Parquet partition, Spark 4.1.3 returns eight
maps and eight nulls. A native component reproduction using the current Comet
expressions and this serde's nested CASE shape returns maps only at ids `0, 4,
8, 12`, turning four expected maps into NULL. The guard's counter consumes all
16 rows, while the constructor's independent counter sees only the eight rows
selected by the guard.
The duplicated-child problem already exists under `EXCEPTION`, but removing
the default `LAST_WIN` fallback introduces it for that policy here. #5867 is
still open. Please retain fallback for nondeterministic children, incorporate
its gate, or evaluate each child once before enabling this route, with a
regression test for the query above. The component dependency/validation
boundary is recorded in the review summary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
