This is an automated email from the ASF dual-hosted git repository.
comphead pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 1142c3a709 chore: fallback for `spark.sql.mapKeyDedupPolicy` ==
`LAST_WIN` (#4863)
1142c3a709 is described below
commit 1142c3a709805f210bedd3a2fff0f2b434c2e387
Author: Oleks V <[email protected]>
AuthorDate: Thu Jul 9 14:09:14 2026 -0700
chore: fallback for `spark.sql.mapKeyDedupPolicy` == `LAST_WIN` (#4863)
---
.../main/scala/org/apache/comet/serde/maps.scala | 38 ++++++++++++++++---
.../map/map_from_arrays_dedup_policy.sql | 41 ++++++++++++++++++++
.../map/map_from_entries_dedup_policy.sql | 44 ++++++++++++++++++++++
3 files changed, 117 insertions(+), 6 deletions(-)
diff --git a/spark/src/main/scala/org/apache/comet/serde/maps.scala
b/spark/src/main/scala/org/apache/comet/serde/maps.scala
index d663941c51..a5ae24d215 100644
--- a/spark/src/main/scala/org/apache/comet/serde/maps.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/maps.scala
@@ -75,8 +75,32 @@ 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."
+
+ def isLastWin: Boolean =
+ SQLConf.get
+ .getConf(SQLConf.MAP_KEY_DEDUP_POLICY)
+ .toString
+ .equalsIgnoreCase(SQLConf.MapKeyDedupPolicy.LAST_WIN.toString)
+}
+
object CometMapFromArrays extends CometExpressionSerde[MapFromArrays] {
+ override def getIncompatibleReasons(): Seq[String] =
+ Seq(MapKeyDedupPolicySupport.incompatibleReason)
+
+ override def getSupportLevel(expr: MapFromArrays): SupportLevel = {
+ if (MapKeyDedupPolicySupport.isLastWin) {
+ Incompatible(Some(MapKeyDedupPolicySupport.incompatibleReason))
+ } else {
+ Compatible(None)
+ }
+ }
+
override def convert(
expr: MapFromArrays,
inputs: Seq[Attribute],
@@ -127,16 +151,18 @@ object CometMapFromEntries
"`BinaryType` is not supported as a map value in `map_from_entries`"
override def getIncompatibleReasons(): Seq[String] =
- Seq(keyUnsupportedReason, valueUnsupportedReason)
+ Seq(keyUnsupportedReason, valueUnsupportedReason,
MapKeyDedupPolicySupport.incompatibleReason)
override def getSupportLevel(expr: MapFromEntries): SupportLevel = {
if (SupportLevel.containsType(expr.dataType.keyType, classOf[BinaryType]))
{
- return Incompatible(Some(keyUnsupportedReason))
- }
- if (SupportLevel.containsType(expr.dataType.valueType,
classOf[BinaryType])) {
- return Incompatible(Some(valueUnsupportedReason))
+ Incompatible(Some(keyUnsupportedReason))
+ } else if (SupportLevel.containsType(expr.dataType.valueType,
classOf[BinaryType])) {
+ Incompatible(Some(valueUnsupportedReason))
+ } else if (MapKeyDedupPolicySupport.isLastWin) {
+ Incompatible(Some(MapKeyDedupPolicySupport.incompatibleReason))
+ } else {
+ Compatible(None)
}
- Compatible(None)
}
}
diff --git
a/spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_dedup_policy.sql
b/spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_dedup_policy.sql
new file mode 100644
index 0000000000..fffaf5f9a9
--- /dev/null
+++
b/spark/src/test/resources/sql-tests/expressions/map/map_from_arrays_dedup_policy.sql
@@ -0,0 +1,41 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Verifies that `map_from_arrays` falls back to Spark when
`spark.sql.mapKeyDedupPolicy` is set
+-- to `LAST_WIN`. Spark's ArrayBasedMapBuilder keeps the last occurrence of
each duplicate key;
+-- Comet's native `map` scalar has no LAST_WIN path, so it must fall back. The
default `EXCEPTION`
+-- mode agrees with Comet and is covered by `map_from_arrays.sql`.
+
+-- Config: spark.sql.mapKeyDedupPolicy=LAST_WIN
+
+statement
+CREATE TABLE test_map_from_arrays_dedup(k array<string>, v array<int>) USING
parquet
+
+statement
+INSERT INTO test_map_from_arrays_dedup VALUES
+ (array('a', 'b', 'c'), array(1, 2, 3)),
+ (array('a', 'a', 'b'), array(1, 2, 3)),
+ (array('x', 'x'), array(10, 20))
+
+-- literal duplicate keys under LAST_WIN: Spark keeps the last value; Comet
must fall back.
+query expect_fallback(mapKeyDedupPolicy)
+SELECT map_from_arrays(array('a', 'a', 'b'), array(1, 2, 3))
+
+-- column input falls back the same way; the incompat branch is triggered by
the SQLConf value,
+-- not per-row content.
+query expect_fallback(mapKeyDedupPolicy)
+SELECT map_from_arrays(k, v) FROM test_map_from_arrays_dedup
diff --git
a/spark/src/test/resources/sql-tests/expressions/map/map_from_entries_dedup_policy.sql
b/spark/src/test/resources/sql-tests/expressions/map/map_from_entries_dedup_policy.sql
new file mode 100644
index 0000000000..feba795193
--- /dev/null
+++
b/spark/src/test/resources/sql-tests/expressions/map/map_from_entries_dedup_policy.sql
@@ -0,0 +1,44 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Verifies that `map_from_entries` falls back to Spark when
`spark.sql.mapKeyDedupPolicy` is set
+-- to `LAST_WIN`. `CometMapFromEntries` mixes in `CodegenDispatchFallback`, so
its native
+-- `Incompatible` normally routes through the JVM codegen dispatcher; we
disable the dispatcher
+-- here so the incompat branch surfaces as a genuine Spark fallback rather
than in-pipeline
+-- codegen. The default `EXCEPTION` mode agrees with Comet and is covered by
+-- `map_from_entries.sql`.
+
+-- Config: spark.sql.mapKeyDedupPolicy=LAST_WIN
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=false
+
+statement
+CREATE TABLE test_map_from_entries_dedup(entries array<struct<key:string,
value:int>>) USING parquet
+
+statement
+INSERT INTO test_map_from_entries_dedup VALUES
+ (array(struct('a', 1), struct('b', 2), struct('c', 3))),
+ (array(struct('a', 1), struct('a', 2), struct('b', 3))),
+ (array(struct('x', 10), struct('x', 20)))
+
+-- literal duplicate keys under LAST_WIN: Spark keeps the last value; Comet
must fall back.
+query expect_fallback(mapKeyDedupPolicy)
+SELECT map_from_entries(array(struct('a', 1), struct('a', 2), struct('b', 3)))
+
+-- column input falls back the same way; the incompat branch is triggered by
the SQLConf value,
+-- not per-row content.
+query expect_fallback(mapKeyDedupPolicy)
+SELECT map_from_entries(entries) FROM test_map_from_entries_dedup
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
