[GLUTEN-10134][Vl] elt ansi function overlay

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala

@@ -1250,6 +1250,68 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with Logging {
     GenericExpressionTransformer(substraitExprName, child, expr)
   }
 
+  override def genEltTransformer(
+      substraitExprName: String,
+      children: Seq[ExpressionTransformer],
+      expr: Elt): ExpressionTransformer = {
+    // Velox's elt derives whether an out-of-range index raises an error from the session's
+    // 'spark.sql.ansi.enabled', while Spark captures it in Elt.failOnError at analysis time.
+    // The two normally agree; fall back when they don't, so the ANSI behavior never diverges.
+    if (expr.failOnError != SQLConf.get.ansiEnabled) {
+      GlutenExceptionUtil
+        .throwsNotFullySupported(
+          ExpressionNames.ELT,
+          EltRestrictions.NOT_SUPPORT_FAIL_ON_ERROR_MISMATCH
+        )
+    }
+    GenericExpressionTransformer(substraitExprName, children, expr)
+  }
+
+  override def genConvTransformer(
+      substraitExprName: String,
+      children: Seq[ExpressionTransformer],
+      expr: Conv): ExpressionTransformer = {
+    // Velox derives whether an overflow raises an error from the session's
+    // 'spark.sql.ansi.enabled', while Spark captures it in Conv.ansiEnabled at analysis time.
+    // The two normally agree; fall back when they don't, so the ANSI behavior never diverges.
+    if (expr.ansiEnabled != SQLConf.get.ansiEnabled) {
+      GlutenExceptionUtil
+        .throwsNotFullySupported(
+          ExpressionNames.CONV,
+          ConvRestrictions.NOT_SUPPORT_ANSI_ENABLED_MISMATCH
+        )
+    }
+    GenericExpressionTransformer(substraitExprName, children, expr)
+  }
+
+  override def genElementAtTransformer(
+      substraitExprName: String,
+      children: Seq[ExpressionTransformer],
+      expr: ElementAt): ExpressionTransformer = {
+    // Only the array input reads failOnError: Spark returns NULL for a key a map does not
+    // contain whatever the ANSI mode is, and so does Velox.
+    if (expr.left.dataType.isInstanceOf[ArrayType]) {
+      if (expr.defaultValueOutOfBound.isDefined) {
+        GlutenExceptionUtil
+          .throwsNotFullySupported(
+            ExpressionNames.ELEMENT_AT,
+            ElementAtRestrictions.NOT_SUPPORT_DEFAULT_VALUE_OUT_OF_BOUND
+          )
+      }
+      // Velox derives whether an out-of-bound index raises an error from the session's
+      // 'spark.sql.ansi.enabled', while Spark captures it in ElementAt.failOnError at
+      // analysis time. The two normally agree; fall back when they don't.
+      if (expr.failOnError != SQLConf.get.ansiEnabled) {
+        GlutenExceptionUtil
+          .throwsNotFullySupported(
+            ExpressionNames.ELEMENT_AT,
+            ElementAtRestrictions.NOT_SUPPORT_FAIL_ON_ERROR_MISMATCH
+          )
+      }
+    }
+    GenericExpressionTransformer(substraitExprName, children, expr)
+  }
+
   override def genBase64StaticInvokeTransformer(
       substraitExprName: String,
       child: ExpressionTransformer,

backends-velox/src/main/scala/org/apache/gluten/expression/ExpressionRestrictions.scala

@@ -83,6 +83,44 @@ object Unbase64Restrictions extends ExpressionRestrictions {
   override val restrictionMessages: Array[String] = Array(NOT_SUPPORT_FAIL_ON_ERROR)
 }
 
+object EltRestrictions extends ExpressionRestrictions {
+  val NOT_SUPPORT_FAIL_ON_ERROR_MISMATCH: String =
+    s"${ExpressionNames.ELT} whose failOnError disagrees with the session's " +
+      s"'${SQLConf.ANSI_ENABLED.key}' is not supported, since Velox derives the ANSI " +
+      s"behavior of elt from the session config"
+
+  override val functionName: String = ExpressionNames.ELT
+
+  override val restrictionMessages: Array[String] = Array(NOT_SUPPORT_FAIL_ON_ERROR_MISMATCH)
+}
+
+object ConvRestrictions extends ExpressionRestrictions {
+  val NOT_SUPPORT_ANSI_ENABLED_MISMATCH: String =
+    s"${ExpressionNames.CONV} whose ansiEnabled disagrees with the session's " +
+      s"'${SQLConf.ANSI_ENABLED.key}' is not supported, since Velox derives the ANSI " +
+      s"behavior of conv from the session config"
+
+  override val functionName: String = ExpressionNames.CONV
+
+  override val restrictionMessages: Array[String] = Array(NOT_SUPPORT_ANSI_ENABLED_MISMATCH)
+}
+
+object ElementAtRestrictions extends ExpressionRestrictions {
+  val NOT_SUPPORT_FAIL_ON_ERROR_MISMATCH: String =
+    s"${ExpressionNames.ELEMENT_AT} over an array whose failOnError disagrees with the " +
+      s"session's '${SQLConf.ANSI_ENABLED.key}' is not supported, since Velox derives the " +
+      s"ANSI behavior of element_at from the session config"
+
+  val NOT_SUPPORT_DEFAULT_VALUE_OUT_OF_BOUND: String =
+    s"${ExpressionNames.ELEMENT_AT} with a default value for an out-of-bound index is not " +
+      s"supported in Velox, which always returns NULL for such an index"
+
+  override val functionName: String = ExpressionNames.ELEMENT_AT
+
+  override val restrictionMessages: Array[String] =
+    Array(NOT_SUPPORT_FAIL_ON_ERROR_MISMATCH, NOT_SUPPORT_DEFAULT_VALUE_OUT_OF_BOUND)
+}
+
 object Base64Restrictions extends ExpressionRestrictions {
   val NOT_SUPPORT_DISABLE_CHUNK_BASE64_STRING: String =
     s"${ExpressionNames.BASE64} with chunkBase64String disabled is not supported"
@@ -125,6 +163,9 @@ object ExpressionRestrictions {
       ToJsonRestrictions,
       Unbase64Restrictions,
       Base64Restrictions,
+      EltRestrictions,
+      ConvRestrictions,
+      ElementAtRestrictions,
       FormatNumberRestrictions
     )
   }

backends-velox/src/test/scala/org/apache/gluten/functions/MathFunctionsValidateSuite.scala

@@ -20,6 +20,7 @@ import org.apache.gluten.config.GlutenConfig
 import org.apache.gluten.execution.{BatchScanExecTransformer, ProjectExecTransformer}
 
 import org.apache.spark.SparkConf
+import org.apache.spark.SparkException
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.internal.SQLConf
 
@@ -64,6 +65,34 @@ class MathFunctionsValidateSuiteAnsiOn extends FunctionsValidateSuite {
       checkGlutenPlan[ProjectExecTransformer]
     }
   }
+
+  test("conv") {
+    runQueryAndCompare(
+      "select conv(cast(l_orderkey as string), 10, 16), conv('big', 36, 16) from lineitem") {
+      checkGlutenPlan[ProjectExecTransformer]
+    }
+
+    // 2^64 - 1 is the largest input that does not overflow, and the sign is applied after
+    // the digits are accumulated, so neither of these raises an error.
+    runQueryAndCompare(
+      "select conv('18446744073709551615', 10, 10), conv('-1', 10, 16) from lineitem") {
+      checkGlutenPlan[ProjectExecTransformer]
+    }
+
+    // An out-of-range base gives NULL rather than an error.
+    runQueryAndCompare("select conv('15', 1, 10), conv('15', 10, 37) from lineitem") {
+      checkGlutenPlan[ProjectExecTransformer]
+    }
+
+    // Digits that do not fit in an unsigned 64-bit integer overflow, which raises an
+    // error in ANSI mode instead of saturating. l_orderkey is at least 1, so the
+    // concatenated input always has more than 16 hexadecimal digits.
+    intercept[SparkException] {
+      sql(
+        "select conv(concat(cast(l_orderkey as string), '0000000000000000'), 16, 10)" +
+          " from lineitem").collect()
+    }
+  }
 }
 
 class MathFunctionsValidateSuite extends FunctionsValidateSuite {
@@ -343,6 +372,22 @@ class MathFunctionsValidateSuite extends FunctionsValidateSuite {
     compareResultsAgainstVanillaSpark("select round(44, -1)", true, { _ => })
   }
 
+  test("conv") {
+    withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+      runQueryAndCompare(
+        "SELECT conv(cast(l_orderkey as string), 10, 16), conv('big', 36, 16) from lineitem") {
+        checkGlutenPlan[ProjectExecTransformer]
+      }
+
+      // Digits that do not fit in an unsigned 64-bit integer saturate with ANSI mode off.
+      runQueryAndCompare(
+        "SELECT conv(concat(cast(l_orderkey as string), '0000000000000000'), 16, 10)," +
+          " conv('9223372036854775807', 36, 16) from lineitem") {
+        checkGlutenPlan[ProjectExecTransformer]
+      }
+    }
+  }
+
   test("shiftleft") {
     runQueryAndCompare("SELECT shiftleft(int_field1, 1) from datatab limit 1") {
       checkGlutenPlan[ProjectExecTransformer]

backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala

@@ -177,6 +177,70 @@ class ScalarFunctionsValidateSuite extends FunctionsValidateSuite {
     }
   }
 
+  test("elt") {
+    withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+      // int_field1 is 1, 2, 3, so every input gets selected by some row.
+      runQueryAndCompare("SELECT elt(int_field1, string_field1, 'b', 'c') FROM datatab") {
+        checkGlutenPlan[ProjectExecTransformer]
+      }
+      // A NULL index, an out-of-range index and a NULL selected input all give NULL
+      // with ANSI mode off.
+      runQueryAndCompare(
+        "SELECT elt(NULL, 'a', 'b'), elt(int_field1 + 3, 'a', 'b'), " +
+          "elt(1, string_field1, 'b') FROM datatab") {
+        checkGlutenPlan[ProjectExecTransformer]
+      }
+      runQueryAndCompare(
+        "SELECT elt(int_field1, cast(string_field1 as binary), cast('b' as binary)) FROM datatab") {
+        checkGlutenPlan[ProjectExecTransformer]
+      }
+    }
+  }
+
+  test("element_at") {
+    withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+      // An index past either end of the array gives NULL with ANSI mode off, and a key
+      // the map does not contain gives NULL whatever the ANSI mode is.
+      runQueryAndCompare(
+        "SELECT element_at(array(l_orderkey, l_partkey), 1)," +
+          " element_at(array(l_orderkey, l_partkey), -1)," +
+          " element_at(array(l_orderkey, l_partkey), 3)," +
+          " element_at(array(l_orderkey, l_partkey), -3)," +
+          " element_at(map(1, 'a', 2, 'b'), 3) FROM lineitem") {
+        checkGlutenPlan[ProjectExecTransformer]
+      }
+
+      // An index of 0 is an error whatever the ANSI mode is.
+      intercept[SparkException] {
+        sql("SELECT element_at(array(l_orderkey, l_partkey), 0) FROM lineitem").collect()
+      }
+    }
+  }
+
+  test("size") {
+    withTempPath {
+      path =>
+        Seq[Seq[Integer]](Seq(1, 2, 3), Seq.empty, null)
+          .toDF("i")
+          .write
+          .parquet(path.getCanonicalPath)
+        spark.read.parquet(path.getCanonicalPath).createOrReplaceTempView("size_tbl")
+
+        withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+          // With ANSI mode off, spark.sql.legacy.sizeOfNull decides between -1 and NULL
+          // for a null collection.
+          Seq("true", "false").foreach {
+            legacySizeOfNull =>
+              withSQLConf(SQLConf.LEGACY_SIZE_OF_NULL.key -> legacySizeOfNull) {
+                runQueryAndCompare("SELECT size(i), cardinality(i) FROM size_tbl") {
+                  checkGlutenPlan[ProjectExecTransformer]
+                }
+              }
+          }
+        }
+    }
+  }
+
   test("shiftright") {
     runQueryAndCompare("SELECT shiftright(int_field1, 1) from datatab limit 1") {
       checkGlutenPlan[ProjectExecTransformer]

backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuiteAnsiOn.scala

@@ -0,0 +1,107 @@
+/*
+ * 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.
+ */
+package org.apache.gluten.functions
+
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.ProjectExecTransformer
+
+import org.apache.spark.SparkConf
+import org.apache.spark.SparkException
+import org.apache.spark.sql.internal.SQLConf
+
+class ScalarFunctionsValidateSuiteAnsiOn extends FunctionsValidateSuite {
+
+  disableFallbackCheck
+
+  import testImplicits._
+
+  override protected def sparkConf: SparkConf = {
+    super.sparkConf
+      .set(GlutenConfig.GLUTEN_ANSI_FALLBACK_ENABLED.key, "false")
+      .set(SQLConf.ANSI_ENABLED.key, "true")
+  }
+
+  test("elt") {
+    // int_field1 is 1, 2, 3, so every index is within range here.
+    runQueryAndCompare("SELECT elt(int_field1, 'a', 'b', 'c') FROM datatab") {
+      checkGlutenPlan[ProjectExecTransformer]
+    }
+
+    // A NULL index gives NULL rather than an error, and a NULL selected input stays NULL.
+    runQueryAndCompare("SELECT elt(NULL, 'a', 'b'), elt(1, string_field1, 'b') FROM datatab") {
+      checkGlutenPlan[ProjectExecTransformer]
+    }
+
+    // An out-of-range index raises an error in ANSI mode. int_field1 - 1 is 0 for the
+    // first row, and int_field1 + 3 is beyond the number of inputs for every row.
+    intercept[SparkException] {
+      sql("SELECT elt(int_field1 - 1, 'a', 'b', 'c') FROM datatab").collect()
+    }
+    intercept[SparkException] {
+      sql("SELECT elt(int_field1 + 3, 'a', 'b', 'c') FROM datatab").collect()
+    }
+  }
+
+  test("element_at") {
+    // In-bound indices, including the negative ones counting from the end of the array,
+    // are unaffected by ANSI mode.
+    runQueryAndCompare(
+      "SELECT element_at(array(l_orderkey, l_partkey), 1)," +
+        " element_at(array(l_orderkey, l_partkey), -1) FROM lineitem") {
+      checkGlutenPlan[ProjectExecTransformer]
+    }
+
+    // A key the map does not contain gives NULL, in ANSI mode as well.
+    runQueryAndCompare("SELECT element_at(map(1, 'a', 2, 'b'), 3) FROM lineitem") {
+      checkGlutenPlan[ProjectExecTransformer]
+    }
+
+    // An index past either end of the array raises an error in ANSI mode.
+    intercept[SparkException] {
+      sql("SELECT element_at(array(l_orderkey, l_partkey), 3) FROM lineitem").collect()
+    }
+    intercept[SparkException] {
+      sql("SELECT element_at(array(l_orderkey, l_partkey), -3) FROM lineitem").collect()
+    }
+    // An index of 0 is an error whatever the ANSI mode is.
+    intercept[SparkException] {
+      sql("SELECT element_at(array(l_orderkey, l_partkey), 0) FROM lineitem").collect()
+    }
+  }
+
+  test("size") {
+    withTempPath {
+      path =>
+        Seq[Seq[Integer]](Seq(1, 2, 3), Seq.empty, null)
+          .toDF("i")
+          .write
+          .parquet(path.getCanonicalPath)
+        spark.read.parquet(path.getCanonicalPath).createOrReplaceTempView("size_tbl")
+
+        // Spark's legacySizeOfNull is 'spark.sql.legacy.sizeOfNull AND NOT ANSI mode', so
+        // size(null) is NULL here whatever spark.sql.legacy.sizeOfNull says.
+        Seq("true", "false").foreach {
+          legacySizeOfNull =>
+            withSQLConf(SQLConf.LEGACY_SIZE_OF_NULL.key -> legacySizeOfNull) {
+              runQueryAndCompare("SELECT size(i) FROM size_tbl") {
+                checkGlutenPlan[ProjectExecTransformer]
+              }
+            }
+        }
+    }
+  }
+}

cpp/velox/CMakeLists.txt

@@ -176,6 +176,8 @@ set(VELOX_SRCS
     memory/VeloxMemoryManager.cc
     operators/functions/RegistrationAllFunctions.cc
     operators/functions/delta/DeltaBitmapAggregator.cc
+    operators/functions/overlay/ElementAt.cc
+    operators/functions/overlay/Elt.cc
     operators/functions/overlay/RegisterFunctionOverlay.cc
     operators/functions/RowConstructorWithNull.cc
     operators/functions/SparkExprToSubfieldFilterParser.cc