Merge remote-tracking branch 'apache/main' into feat/bloom-filter-intermediate-buffer-compat

Author: andygrove

docs/source/contributor-guide/spark_expressions_support.md

@@ -411,7 +411,7 @@
 - [x] randn
 - [ ] random
 - [ ] randstr
-- [ ] rint
+- [x] rint
 - [x] round
 - [x] sec
 - [x] shiftleft

docs/source/user-guide/latest/expressions.md

@@ -174,6 +174,7 @@ of expressions that be disabled.
 | Rand           | `rand`         |
 | Randn          | `randn`        |
 | Remainder      | `%`            |
+| Rint           | `rint`         |
 | Round          | `round`        |
 | Sec            | `sec`          |
 | Signum         | `signum`       |

native/core/src/execution/jni_api.rs

@@ -60,6 +60,7 @@ use datafusion_spark::function::map::str_to_map::SparkStrToMap;
 use datafusion_spark::function::math::expm1::SparkExpm1;
 use datafusion_spark::function::math::factorial::SparkFactorial;
 use datafusion_spark::function::math::hex::SparkHex;
+use datafusion_spark::function::math::rint::SparkRint;
 use datafusion_spark::function::math::trigonometry::SparkCsc;
 use datafusion_spark::function::math::trigonometry::SparkSec;
 use datafusion_spark::function::math::width_bucket::SparkWidthBucket;
@@ -605,6 +606,7 @@ fn register_datafusion_spark_function(session_ctx: &SessionContext) {
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkTryParseUrl::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkFactorial::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkSec::default()));
+    session_ctx.register_udf(ScalarUDF::new_from_impl(SparkRint::default()));
 }
 
 /// Prepares arrow arrays for output.

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -125,6 +125,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
     classOf[Rand] -> CometRand,
     classOf[Randn] -> CometRandn,
     classOf[Remainder] -> CometRemainder,
+    classOf[Rint] -> CometScalarFunction("rint"),
     classOf[Round] -> CometRound,
     classOf[Sec] -> CometScalarFunction("sec"),
     classOf[Signum] -> CometScalarFunction("signum"),

spark/src/test/resources/sql-tests/expressions/math/rint.sql

@@ -0,0 +1,60 @@
+-- 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.
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
+-- Spark's Rint extends UnaryMathExpression with inputTypes = Seq(DoubleType).
+-- It returns the double value closest to the argument equal to a mathematical integer
+-- (Java's Math.rint, IEEE 754 round-half-to-even / banker's rounding).
+
+statement
+CREATE TABLE test_rint(v double) USING parquet
+
+statement
+INSERT INTO test_rint VALUES
+  (0.0),
+  (-0.0),
+  (1.0),
+  (-1.0),
+  (0.4),
+  (0.5),
+  (0.6),
+  (1.5),
+  (2.5),
+  (3.5),
+  (-0.4),
+  (-0.5),
+  (-0.6),
+  (-1.5),
+  (-2.5),
+  (-3.5),
+  (12.3456),
+  (-12.3456),
+  (1.7976931348623157E308),
+  (-1.7976931348623157E308),
+  (4.9E-324),
+  (cast('NaN' as double)),
+  (cast('Infinity' as double)),
+  (cast('-Infinity' as double)),
+  (NULL)
+
+query
+SELECT rint(v) FROM test_rint
+
+-- column with arithmetic
+query
+SELECT rint(v + 0.5) FROM test_rint

spark/src/test/scala/org/apache/comet/CometCodegenFuzzSuite.scala

@@ -308,10 +308,16 @@ class CometCodegenFuzzSuite
     }
   }
 
-  private def probeCardinality(accessor: String, viewName: String): Unit = {
+  private def probeCardinality(accessor: String, dt: DataType, viewName: String): Unit = {
+    // `Size` only supports `ArrayType` in Comet, so for `MapType` we route through `map_keys` to
+    // reach a `Size(ArrayType)`. Spark still calls `getMap` on the column vector to extract the
+    // keys, which is the accessor path this probe is intended to exercise.
+    val sizeExpr = dt match {
+      case _: MapType => s"size(map_keys($accessor))"
+      case _ => s"cardinality($accessor)"
+    }
     assertCodegenRan {
-      checkSparkAnswerAndOperator(
-        s"SELECT $cardinalityProbeUdf(cardinality($accessor)) FROM $viewName")
+      checkSparkAnswerAndOperator(s"SELECT $cardinalityProbeUdf($sizeExpr) FROM $viewName")
     }
   }
 
@@ -323,13 +329,14 @@ class CometCodegenFuzzSuite
   private def probeComplexColumn(field: StructField, viewName: String): Unit = {
     field.dataType match {
       case _: ArrayType | _: MapType =>
-        probeCardinality(field.name, viewName)
+        probeCardinality(field.name, field.dataType, viewName)
 
       case st: StructType =>
         for (subField <- st.fields) {
           val accessor = s"${field.name}.${subField.name}"
           subField.dataType match {
-            case _: ArrayType | _: MapType => probeCardinality(accessor, viewName)
+            case _: ArrayType | _: MapType =>
+              probeCardinality(accessor, subField.dataType, viewName)
             case dt if !isComplexType(dt) =>
               val udfName = s"id_${field.name}_${subField.name}"
               registerIdentityUdfFor(dt, udfName).foreach { _ =>

spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala

@@ -646,7 +646,15 @@ class CometCodegenSuite
   }
 
   test("ScalaUDF over Decimal(38, 10) routes through the BigDecimal slow path") {
-    spark.udf.register("decIdLong", (d: java.math.BigDecimal) => d)
+    // Pin the return type to Decimal(38, 10). TypeTag inference for `BigDecimal` would default to
+    // Decimal(38, 18), and under Spark 4 ANSI the encoder's CheckOverflow throws on the 28-digit
+    // boundary value below when rescaling 10 -> 18.
+    spark.udf.register(
+      "decIdLong",
+      new UDF1[java.math.BigDecimal, java.math.BigDecimal] {
+        override def call(d: java.math.BigDecimal): java.math.BigDecimal = d
+      },
+      DecimalType(38, 10))
     withDecimalTable(
       "DECIMAL(38, 10)",
       Seq(