From 768b3e90f261c7aea58bdb98dc698b90deeeae34 Mon Sep 17 00:00:00 2001 From: Kazantsev Maksim Date: Sun, 14 Dec 2025 16:24:01 +0400 Subject: [PATCH 1/5] impl map_from_entries --- native/core/src/execution/jni_api.rs | 2 + .../apache/comet/serde/QueryPlanSerde.scala | 3 +- .../scala/org/apache/comet/serde/maps.scala | 29 +++++++++++- .../comet/CometMapExpressionSuite.scala | 45 +++++++++++++++++++ 4 files changed, 77 insertions(+), 2 deletions(-) diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs index a24d9930597..4f53cea3e68 100644 --- a/native/core/src/execution/jni_api.rs +++ b/native/core/src/execution/jni_api.rs @@ -46,6 +46,7 @@ use datafusion_spark::function::datetime::date_add::SparkDateAdd; use datafusion_spark::function::datetime::date_sub::SparkDateSub; use datafusion_spark::function::hash::sha1::SparkSha1; use datafusion_spark::function::hash::sha2::SparkSha2; +use datafusion_spark::function::map::map_from_entries::MapFromEntries; use datafusion_spark::function::math::expm1::SparkExpm1; use datafusion_spark::function::string::char::CharFunc; use datafusion_spark::function::string::concat::SparkConcat; @@ -337,6 +338,7 @@ fn register_datafusion_spark_function(session_ctx: &SessionContext) { session_ctx.register_udf(ScalarUDF::new_from_impl(SparkSha1::default())); session_ctx.register_udf(ScalarUDF::new_from_impl(SparkConcat::default())); session_ctx.register_udf(ScalarUDF::new_from_impl(SparkBitwiseNot::default())); + session_ctx.register_udf(ScalarUDF::new_from_impl(MapFromEntries::default())); } /// Prepares arrow arrays for output. diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 54df2f1688d..a99cf3824bf 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -125,7 +125,8 @@ object QueryPlanSerde extends Logging with CometExprShim { classOf[MapKeys] -> CometMapKeys, classOf[MapEntries] -> CometMapEntries, classOf[MapValues] -> CometMapValues, - classOf[MapFromArrays] -> CometMapFromArrays) + classOf[MapFromArrays] -> CometMapFromArrays, + classOf[MapFromEntries] -> CometMapFromEntries) private val structExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = Map( classOf[CreateNamedStruct] -> CometCreateNamedStruct, 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 2e217f6af0b..498aa3594cf 100644 --- a/spark/src/main/scala/org/apache/comet/serde/maps.scala +++ b/spark/src/main/scala/org/apache/comet/serde/maps.scala @@ -19,9 +19,12 @@ package org.apache.comet.serde +import scala.annotation.tailrec + import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.types.{ArrayType, MapType} +import org.apache.spark.sql.types.{ArrayType, BinaryType, DataType, MapType, StructType} +import org.apache.comet.serde.CometArrayReverse.containsBinary import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType} object CometMapKeys extends CometExpressionSerde[MapKeys] { @@ -89,3 +92,27 @@ object CometMapFromArrays extends CometExpressionSerde[MapFromArrays] { optExprWithInfo(mapFromArraysExpr, expr, expr.children: _*) } } + +object CometMapFromEntries extends CometScalarFunction[MapFromEntries]("map_from_entries") { + val keyUnsupportedReason = "Using BinaryType as Map keys is not allowed in map_from_entries" + val valueUnsupportedReason = "Using BinaryType as Map values is not allowed in map_from_entries" + + private def containsBinary(dataType: DataType): Boolean = { + dataType match { + case BinaryType => true + case StructType(fields) => fields.exists(field => containsBinary(field.dataType)) + case ArrayType(elementType, _) => containsBinary(elementType) + case _ => false + } + } + + override def getSupportLevel(expr: MapFromEntries): SupportLevel = { + if (containsBinary(expr.dataType.keyType)) { + return Incompatible(Some(keyUnsupportedReason)) + } + if (containsBinary(expr.dataType.valueType)) { + return Incompatible(Some(valueUnsupportedReason)) + } + Compatible(None) + } +} diff --git a/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala index 88c13391a67..01b9744ed6f 100644 --- a/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala @@ -25,7 +25,9 @@ import org.apache.hadoop.fs.Path import org.apache.spark.sql.CometTestBase import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.BinaryType +import org.apache.comet.serde.CometMapFromEntries import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions} class CometMapExpressionSuite extends CometTestBase { @@ -125,4 +127,47 @@ class CometMapExpressionSuite extends CometTestBase { } } + test("map_from_entries") { + withTempDir { dir => + val path = new Path(dir.toURI.toString, "test.parquet") + val filename = path.toString + val random = new Random(42) + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + val schemaGenOptions = + SchemaGenOptions( + generateArray = true, + generateStruct = true, + primitiveTypes = SchemaGenOptions.defaultPrimitiveTypes.filterNot(_ == BinaryType)) + val dataGenOptions = DataGenOptions(allowNull = false, generateNegativeZero = false) + ParquetGenerator.makeParquetFile( + random, + spark, + filename, + 100, + schemaGenOptions, + dataGenOptions) + } + val df = spark.read.parquet(filename) + df.createOrReplaceTempView("t1") + for (field <- df.schema.fieldNames) { + checkSparkAnswerAndOperator( + spark.sql(s"SELECT map_from_entries(array(struct($field as a, $field as b))) FROM t1")) + } + } + } + + test("map_from_entries - fallback for binary type") { + val table = "t2" + withTable(table) { + sql( + s"create table $table using parquet as select cast(array() as array) as c1 from range(10)") + checkSparkAnswerAndFallbackReason( + sql(s"select map_from_entries(array(struct(c1, 0))) from $table"), + CometMapFromEntries.keyUnsupportedReason) + checkSparkAnswerAndFallbackReason( + sql(s"select map_from_entries(array(struct(0, c1))) from $table"), + CometMapFromEntries.valueUnsupportedReason) + } + } + } From c68c3428676b5d991e7ba9e13464bf2ce1ec84e8 Mon Sep 17 00:00:00 2001 From: Kazantsev Maksim Date: Tue, 16 Dec 2025 16:10:43 +0400 Subject: [PATCH 2/5] Revert "impl map_from_entries" This reverts commit 768b3e90f261c7aea58bdb98dc698b90deeeae34. --- native/core/src/execution/jni_api.rs | 2 - .../apache/comet/serde/QueryPlanSerde.scala | 3 +- .../scala/org/apache/comet/serde/maps.scala | 29 +----------- .../comet/CometMapExpressionSuite.scala | 45 ------------------- 4 files changed, 2 insertions(+), 77 deletions(-) diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs index 4f53cea3e68..a24d9930597 100644 --- a/native/core/src/execution/jni_api.rs +++ b/native/core/src/execution/jni_api.rs @@ -46,7 +46,6 @@ use datafusion_spark::function::datetime::date_add::SparkDateAdd; use datafusion_spark::function::datetime::date_sub::SparkDateSub; use datafusion_spark::function::hash::sha1::SparkSha1; use datafusion_spark::function::hash::sha2::SparkSha2; -use datafusion_spark::function::map::map_from_entries::MapFromEntries; use datafusion_spark::function::math::expm1::SparkExpm1; use datafusion_spark::function::string::char::CharFunc; use datafusion_spark::function::string::concat::SparkConcat; @@ -338,7 +337,6 @@ fn register_datafusion_spark_function(session_ctx: &SessionContext) { session_ctx.register_udf(ScalarUDF::new_from_impl(SparkSha1::default())); session_ctx.register_udf(ScalarUDF::new_from_impl(SparkConcat::default())); session_ctx.register_udf(ScalarUDF::new_from_impl(SparkBitwiseNot::default())); - session_ctx.register_udf(ScalarUDF::new_from_impl(MapFromEntries::default())); } /// Prepares arrow arrays for output. diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index a99cf3824bf..54df2f1688d 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -125,8 +125,7 @@ object QueryPlanSerde extends Logging with CometExprShim { classOf[MapKeys] -> CometMapKeys, classOf[MapEntries] -> CometMapEntries, classOf[MapValues] -> CometMapValues, - classOf[MapFromArrays] -> CometMapFromArrays, - classOf[MapFromEntries] -> CometMapFromEntries) + classOf[MapFromArrays] -> CometMapFromArrays) private val structExpressions: Map[Class[_ <: Expression], CometExpressionSerde[_]] = Map( classOf[CreateNamedStruct] -> CometCreateNamedStruct, 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 498aa3594cf..2e217f6af0b 100644 --- a/spark/src/main/scala/org/apache/comet/serde/maps.scala +++ b/spark/src/main/scala/org/apache/comet/serde/maps.scala @@ -19,12 +19,9 @@ package org.apache.comet.serde -import scala.annotation.tailrec - import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.types.{ArrayType, BinaryType, DataType, MapType, StructType} +import org.apache.spark.sql.types.{ArrayType, MapType} -import org.apache.comet.serde.CometArrayReverse.containsBinary import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType} object CometMapKeys extends CometExpressionSerde[MapKeys] { @@ -92,27 +89,3 @@ object CometMapFromArrays extends CometExpressionSerde[MapFromArrays] { optExprWithInfo(mapFromArraysExpr, expr, expr.children: _*) } } - -object CometMapFromEntries extends CometScalarFunction[MapFromEntries]("map_from_entries") { - val keyUnsupportedReason = "Using BinaryType as Map keys is not allowed in map_from_entries" - val valueUnsupportedReason = "Using BinaryType as Map values is not allowed in map_from_entries" - - private def containsBinary(dataType: DataType): Boolean = { - dataType match { - case BinaryType => true - case StructType(fields) => fields.exists(field => containsBinary(field.dataType)) - case ArrayType(elementType, _) => containsBinary(elementType) - case _ => false - } - } - - override def getSupportLevel(expr: MapFromEntries): SupportLevel = { - if (containsBinary(expr.dataType.keyType)) { - return Incompatible(Some(keyUnsupportedReason)) - } - if (containsBinary(expr.dataType.valueType)) { - return Incompatible(Some(valueUnsupportedReason)) - } - Compatible(None) - } -} diff --git a/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala b/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala index 01b9744ed6f..88c13391a67 100644 --- a/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometMapExpressionSuite.scala @@ -25,9 +25,7 @@ import org.apache.hadoop.fs.Path import org.apache.spark.sql.CometTestBase import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.BinaryType -import org.apache.comet.serde.CometMapFromEntries import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions} class CometMapExpressionSuite extends CometTestBase { @@ -127,47 +125,4 @@ class CometMapExpressionSuite extends CometTestBase { } } - test("map_from_entries") { - withTempDir { dir => - val path = new Path(dir.toURI.toString, "test.parquet") - val filename = path.toString - val random = new Random(42) - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - val schemaGenOptions = - SchemaGenOptions( - generateArray = true, - generateStruct = true, - primitiveTypes = SchemaGenOptions.defaultPrimitiveTypes.filterNot(_ == BinaryType)) - val dataGenOptions = DataGenOptions(allowNull = false, generateNegativeZero = false) - ParquetGenerator.makeParquetFile( - random, - spark, - filename, - 100, - schemaGenOptions, - dataGenOptions) - } - val df = spark.read.parquet(filename) - df.createOrReplaceTempView("t1") - for (field <- df.schema.fieldNames) { - checkSparkAnswerAndOperator( - spark.sql(s"SELECT map_from_entries(array(struct($field as a, $field as b))) FROM t1")) - } - } - } - - test("map_from_entries - fallback for binary type") { - val table = "t2" - withTable(table) { - sql( - s"create table $table using parquet as select cast(array() as array) as c1 from range(10)") - checkSparkAnswerAndFallbackReason( - sql(s"select map_from_entries(array(struct(c1, 0))) from $table"), - CometMapFromEntries.keyUnsupportedReason) - checkSparkAnswerAndFallbackReason( - sql(s"select map_from_entries(array(struct(0, c1))) from $table"), - CometMapFromEntries.valueUnsupportedReason) - } - } - } From cf539183972a203ed8fe6ea8f77a0906880c6bd2 Mon Sep 17 00:00:00 2001 From: Kazantsev Maksim Date: Tue, 1 Sep 2026 22:09:22 +0400 Subject: [PATCH 3/5] work --- docs/source/user-guide/latest/expressions.md | 2 +- .../main/scala/org/apache/comet/serde/math.scala | 6 ++++-- .../resources/sql-tests/expressions/math/abs.sql | 16 ++++++++++++++++ .../sql-tests/expressions/math/abs_ansi.sql | 5 +++++ 4 files changed, 26 insertions(+), 3 deletions(-) diff --git a/docs/source/user-guide/latest/expressions.md b/docs/source/user-guide/latest/expressions.md index 2ba71e93c4d..e744067dd0f 100644 --- a/docs/source/user-guide/latest/expressions.md +++ b/docs/source/user-guide/latest/expressions.md @@ -418,7 +418,7 @@ The type-name conversion functions (`bigint`, `binary`, `boolean`, `date`, `deci | `+` | ✅ | Native | | | `-` | ✅ | Native | | | `/` | ✅ | Native | | -| `abs` | ✅ | Native | Interval types fall back | +| `abs` | ✅ | Hybrid | Interval types route through the JVM codegen dispatcher; numeric types run natively | | `acos` | ✅ | Native | | | `acosh` | ✅ | Native | | | `asin` | ✅ | Native | | diff --git a/spark/src/main/scala/org/apache/comet/serde/math.scala b/spark/src/main/scala/org/apache/comet/serde/math.scala index a0ea76cc061..6cbbade3fa5 100644 --- a/spark/src/main/scala/org/apache/comet/serde/math.scala +++ b/spark/src/main/scala/org/apache/comet/serde/math.scala @@ -169,9 +169,11 @@ object CometUnhex extends CometExpressionSerde[Unhex] with MathExprBase { } } -object CometAbs extends CometExpressionSerde[Abs] with MathExprBase { +object CometAbs extends CometExpressionSerde[Abs] with MathExprBase with CodegenDispatchFallback { - val unsupportedReason: String = "Only integral, floating-point, and decimal types are supported" + val unsupportedReason: String = + "Interval types are not supported natively and are handled via JVM codegen dispatch; " + + "this fallback only applies when the dispatcher is disabled" override def getUnsupportedReasons(): Seq[String] = Seq(unsupportedReason) diff --git a/spark/src/test/resources/sql-tests/expressions/math/abs.sql b/spark/src/test/resources/sql-tests/expressions/math/abs.sql index 5e811e7ea52..31568fd4e60 100644 --- a/spark/src/test/resources/sql-tests/expressions/math/abs.sql +++ b/spark/src/test/resources/sql-tests/expressions/math/abs.sql @@ -27,3 +27,19 @@ SELECT abs(i), abs(l), abs(f), abs(d) FROM test_abs -- literal arguments query SELECT abs(-5), abs(-1.5), abs(0), abs(NULL) + +-- abs() on intervals has no native impl; routed through the JVM codegen dispatcher. +-- Interval values are built inline: native Parquet scan of interval columns is unsupported +-- (https://github.com/apache/datafusion-comet/issues/5060), and a top-level YearMonthIntervalType projection column is still rejected by the +-- projection type gate (https://github.com/apache/datafusion-comet/issues/5061), so the ym result is wrapped in a struct. +query +SELECT abs(make_dt_interval(1, 2, 3, 4.5)) AS dt_pos, + abs(make_dt_interval(-1, -2, -3, -4.5)) AS dt_neg, + abs(make_dt_interval(0, 0, 0, 0)) AS dt_zero, + abs(CAST(NULL AS INTERVAL DAY TO SECOND)) AS dt_null + +-- interval year to month: dispatched the same way; wrapped in a struct because a top-level +-- YearMonthIntervalType column is rejected by the projection output type gate (https://github.com/apache/datafusion-comet/issues/5061) +query +SELECT named_struct('v', abs(make_ym_interval(1, 6))) AS ym_pos, + named_struct('v', abs(make_ym_interval(-1, -6))) AS ym_neg diff --git a/spark/src/test/resources/sql-tests/expressions/math/abs_ansi.sql b/spark/src/test/resources/sql-tests/expressions/math/abs_ansi.sql index 879b24f3f65..1bba74801f4 100644 --- a/spark/src/test/resources/sql-tests/expressions/math/abs_ansi.sql +++ b/spark/src/test/resources/sql-tests/expressions/math/abs_ansi.sql @@ -104,3 +104,8 @@ SELECT abs(v) FROM ansi_test_abs_byte -- literal query expect_error(overflow) SELECT abs(cast(-128 as tinyint)) + +-- overflow: abs on Long.MinValue microseconds throws; the dispatched codegen path must +-- propagate Spark's exception +query expect_error(overflow) +SELECT abs(make_dt_interval(-2147483648)) From a779d471c6cda34f6e2dd1da5131795843ea4876 Mon Sep 17 00:00:00 2001 From: Kazantsev Maksim Date: Thu, 3 Sep 2026 21:31:08 +0400 Subject: [PATCH 4/5] address comments --- .../src/test/resources/sql-tests/expressions/math/abs.sql | 6 ++++++ .../test/resources/sql-tests/expressions/math/abs_ansi.sql | 7 ++++--- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/spark/src/test/resources/sql-tests/expressions/math/abs.sql b/spark/src/test/resources/sql-tests/expressions/math/abs.sql index 31568fd4e60..88be9c9e379 100644 --- a/spark/src/test/resources/sql-tests/expressions/math/abs.sql +++ b/spark/src/test/resources/sql-tests/expressions/math/abs.sql @@ -43,3 +43,9 @@ SELECT abs(make_dt_interval(1, 2, 3, 4.5)) AS dt_pos, query SELECT named_struct('v', abs(make_ym_interval(1, 6))) AS ym_pos, named_struct('v', abs(make_ym_interval(-1, -6))) AS ym_neg + +query +SELECT abs(make_dt_interval(-106751991, -4, 0, -54.775808)) + +query +SELECT abs(make_ym_interval(0, -2147483648)) diff --git a/spark/src/test/resources/sql-tests/expressions/math/abs_ansi.sql b/spark/src/test/resources/sql-tests/expressions/math/abs_ansi.sql index 1bba74801f4..2ce672df721 100644 --- a/spark/src/test/resources/sql-tests/expressions/math/abs_ansi.sql +++ b/spark/src/test/resources/sql-tests/expressions/math/abs_ansi.sql @@ -105,7 +105,8 @@ SELECT abs(v) FROM ansi_test_abs_byte query expect_error(overflow) SELECT abs(cast(-128 as tinyint)) --- overflow: abs on Long.MinValue microseconds throws; the dispatched codegen path must --- propagate Spark's exception query expect_error(overflow) -SELECT abs(make_dt_interval(-2147483648)) +SELECT abs(make_dt_interval(-106751991, -4, 0, -54.775808)) + +query expect_error(overflow) +SELECT abs(make_ym_interval(0, -2147483648)) From e8238e4a6009eb845f3ea26f1b3cab7df46782e1 Mon Sep 17 00:00:00 2001 From: Kazantsev Maksim Date: Thu, 3 Sep 2026 21:37:47 +0400 Subject: [PATCH 5/5] address comments --- spark/src/test/resources/sql-tests/expressions/math/abs.sql | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/spark/src/test/resources/sql-tests/expressions/math/abs.sql b/spark/src/test/resources/sql-tests/expressions/math/abs.sql index 88be9c9e379..9eef45f35da 100644 --- a/spark/src/test/resources/sql-tests/expressions/math/abs.sql +++ b/spark/src/test/resources/sql-tests/expressions/math/abs.sql @@ -44,8 +44,8 @@ query SELECT named_struct('v', abs(make_ym_interval(1, 6))) AS ym_pos, named_struct('v', abs(make_ym_interval(-1, -6))) AS ym_neg -query +query expect_error(overflow) SELECT abs(make_dt_interval(-106751991, -4, 0, -54.775808)) -query +query expect_error(overflow) SELECT abs(make_ym_interval(0, -2147483648))