[VL] Support partial fallback generic unmapped catalyst expressions

backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarPartialProjectExec.scala

@@ -297,6 +297,22 @@ object ColumnarPartialProjectExec {
     HiveUDFTransformer.isHiveUDF(h) && !VeloxHiveUDFTransformer.isSupportedHiveUDF(h)
   }
 
+  /**
+   * Returns true for generic catalyst expressions that Velox cannot offload and should be pulled
+   * into ColumnarPartialProject, excluding ScalaUDF/HiveUDF and non-row-evaluable wrappers.
+   */
+  private def isUnmappedGlutenExpr(expr: Expression): Boolean = expr match {
+    case _: ScalaUDF => false
+    case _ if HiveUDFTransformer.isHiveUDF(expr) => false
+    case _: Literal | _: AttributeReference | _: BoundReference => false
+    case _: NamedExpression => false
+    case _ =>
+      expr.deterministic &&
+      !expr.isInstanceOf[Unevaluable] &&
+      !expr.isInstanceOf[LambdaFunction] &&
+      !ExpressionMappings.expressionsMap.contains(expr.getClass)
+  }
+
   private def isBlacklistExpression(e: Expression): Boolean = {
     ExpressionMappings.blacklistExpressionMap.contains(e.getClass)
   }
@@ -307,6 +323,7 @@ object ColumnarPartialProjectExec {
       case _: ScalaUDF => true
       case h if containsUnsupportedHiveUDF(h) => true
       case e if isBlacklistExpression(e) => true
+      case e if isUnmappedGlutenExpr(e) => true
       case p => p.children.exists(c => containsUDFOrBlacklistExpression(c))
     }
   }
@@ -339,6 +356,8 @@ object ColumnarPartialProjectExec {
         replaceByAlias(h, replacedAlias)
       case e if isBlacklistExpression(e) =>
         replaceByAlias(e, replacedAlias)
+      case e if isUnmappedGlutenExpr(e) =>
+        replaceByAlias(e, replacedAlias)
       case au @ Alias(_: ScalaUDF, _) =>
         val replaceIndex = replacedAlias.indexWhere(r => r.exprId == au.exprId)
         if (replaceIndex == -1) {

backends-velox/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala

@@ -20,16 +20,31 @@ import org.apache.gluten.config.GlutenConfig
 import org.apache.gluten.execution.{ColumnarPartialProjectExec, WholeStageTransformerSuite}
 
 import org.apache.spark.SparkConf
+import org.apache.spark.sql.catalyst.FunctionIdentifier
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionInfo, UnaryExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.optimizer.{ConstantFolding, NullPropagation}
 import org.apache.spark.sql.execution.ProjectExec
 import org.apache.spark.sql.functions.udf
+import org.apache.spark.sql.types.{BooleanType, DataType}
 
 import java.io.File
 
 case class MyStruct(a: Long, b: Array[Long])
 
 case class MyStructWithNullValue(a: Option[Long], b: Array[Long])
 
+case class DummyUnmappedExpr(child: Expression) extends UnaryExpression with CodegenFallback {
+  override def dataType: DataType = BooleanType
+
+  override protected def nullSafeEval(input: Any): Any = {
+    input.asInstanceOf[Long] % 2 == 0
+  }
+
+  override protected def withNewChildInternal(newChild: Expression): DummyUnmappedExpr =
+    copy(child = newChild)
+}
+
 class UDFPartialProjectSuite extends WholeStageTransformerSuite {
   disableFallbackCheck
   override protected val resourcePath: String = "/tpch-data-parquet"
@@ -64,9 +79,22 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite {
     spark.udf.register("no_argument", noArgument)
     val concat = udf((x: String) => x + "_concat")
     spark.udf.register("concat_concat", concat)
+    spark.sessionState.functionRegistry.registerFunction(
+      FunctionIdentifier("dummy_unmapped"),
+      new ExpressionInfo(classOf[DummyUnmappedExpr].getName, "dummy_unmapped"),
+      (children: Seq[Expression]) => DummyUnmappedExpr(children.head)
+    )
 
   }
 
+  test("fallback generic unmapped expression via partial project") {
+    runQueryAndCompare(
+      "SELECT sum(cast(dummy_unmapped(cast(l_orderkey as long)) as int)" +
+        " + hash(l_partkey)) from lineitem") {
+      checkGlutenPlan[ColumnarPartialProjectExec]
+    }
+  }
+
   ignore("test plus_one") {
     runQueryAndCompare("SELECT sum(plus_one(cast(l_orderkey as long))) from lineitem") {
       checkGlutenPlan[ColumnarPartialProjectExec]

gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -883,6 +883,7 @@ class VeloxTestSettings extends BackendTestSettings {
   enableSuite[GlutenVariantSuite]
     // TODO: Velox parquet writer marks all struct fields as OPTIONAL (nullable),
     //  but Spark's variant type requires REQUIRED fields. Needs Velox-side fix.
+    .exclude("non-literal variant_get")
     .exclude("SPARK-47546: invalid variant binary")
     .exclude("SPARK-47546: valid variant binary")
   enableSuite[GlutenVariantWriteShreddingSuite]
@@ -926,6 +927,9 @@ class VeloxTestSettings extends BackendTestSettings {
       // Velox's collect_list / collect_set are by design declarative aggregate so plan check
       // for ObjectHashAggregateExec will fail. Overridden
       "SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle",
+      // Negative HLL tests throw SparkException on Gluten because executor-side failures are
+      // wrapped at the driver boundary.
+      "SPARK-16484: hll_*_agg + hll_union negative tests",
       "SPARK-31620: agg with subquery (whole-stage-codegen = true)",
       "SPARK-31620: agg with subquery (whole-stage-codegen = false)"
     )

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql
 import org.apache.gluten.config.GlutenConfig
 import org.apache.gluten.execution.{HashAggregateExecBaseTransformer, HashAggregateExecTransformer}
 
+import org.apache.spark.SparkException
 import org.apache.spark.sql.execution.WholeStageCodegenExec
 import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, ObjectHashAggregateExec, SortAggregateExec}
 import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
@@ -312,6 +313,96 @@ class GlutenDataFrameAggregateSuite extends DataFrameAggregateSuite with GlutenS
     }
   }
 
+  testGluten("SPARK-16484: hll_*_agg + hll_union negative tests") {
+    val df1 = Seq(
+      (1, "a"), (1, "a"), (1, "a"),
+      (1, "b"),
+      (1, "c"), (1, "c"),
+      (1, "d")
+    ).toDF("id", "value")
+    df1.createOrReplaceTempView("df1")
+
+    val df2 = Seq(
+      (1, "a"),
+      (1, "c"),
+      (1, "d"), (1, "d"), (1, "d"),
+      (1, "e"), (1, "e"),
+      (1, "f")
+    ).toDF("id", "value")
+    df2.createOrReplaceTempView("df2")
+
+    def assertWrappedSparkThrowable(
+        e: SparkException,
+        errorClass: String,
+        messageFragments: Seq[String]): Unit = {
+      val combinedMessage =
+        Option(e.getMessage).getOrElse("") + "\n" + Option(e.getCause).map(_.toString).getOrElse("")
+      assert(combinedMessage.contains(errorClass))
+      messageFragments.foreach(fragment => assert(combinedMessage.contains(fragment)))
+    }
+
+    val invalidLow = intercept[SparkException] {
+      val res = df1.groupBy("id")
+        .agg(
+          hll_sketch_agg("value", 1).as("hllsketch")
+        )
+      checkAnswer(res, Nil)
+    }
+    assertWrappedSparkThrowable(
+      invalidLow,
+      "HLL_INVALID_LG_K",
+      Seq("`hll_sketch_agg`", "1"))
+
+    val invalidHigh = intercept[SparkException] {
+      val res = df1.groupBy("id")
+        .agg(
+          hll_sketch_agg("value", 25).as("hllsketch")
+        )
+      checkAnswer(res, Nil)
+    }
+    assertWrappedSparkThrowable(
+      invalidHigh,
+      "HLL_INVALID_LG_K",
+      Seq("`hll_sketch_agg`", "25"))
+
+    val unionError = intercept[SparkException] {
+      val i1 = df1.groupBy("id")
+        .agg(
+          hll_sketch_agg("value").as("hllsketch_left")
+        )
+      val i2 = df2.groupBy("id")
+        .agg(
+          hll_sketch_agg("value", 20).as("hllsketch_right")
+        )
+      val res = i1.join(i2).withColumn("union", hll_union("hllsketch_left", "hllsketch_right"))
+      checkAnswer(res, Nil)
+    }
+    assertWrappedSparkThrowable(
+      unionError,
+      "HLL_UNION_DIFFERENT_LG_K",
+      Seq("`hll_union`", "12", "20"))
+
+    val unionAggError = intercept[SparkException] {
+      val i1 = df1.groupBy("id")
+        .agg(
+          hll_sketch_agg("value").as("hllsketch")
+        )
+      val i2 = df2.groupBy("id")
+        .agg(
+          hll_sketch_agg("value", 20).as("hllsketch")
+        )
+      val res = i1.union(i2).groupBy("id")
+        .agg(
+          hll_union_agg("hllsketch")
+        )
+      checkAnswer(res, Nil)
+    }
+    assertWrappedSparkThrowable(
+      unionAggError,
+      "HLL_UNION_DIFFERENT_LG_K",
+      Seq("`hll_union_agg`", "12", "20"))
+  }
+
   Seq(true, false).foreach {
     value =>
       testGluten(s"SPARK-31620: agg with subquery (whole-stage-codegen = $value)") {

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenVariantSuite.scala

@@ -16,4 +16,19 @@
  */
 package org.apache.spark.sql
 
-class GlutenVariantSuite extends VariantSuite with GlutenSQLTestsTrait {}
+class GlutenVariantSuite extends VariantSuite with GlutenSQLTestsTrait {
+  testGluten("non-literal variant_get") {
+    val df = spark.sql(
+      """
+        |SELECT variant_get(parse_json(value), path, 'string')
+        |FROM VALUES
+        |  ('{"a":"x"}', '$.a'),
+        |  ('{"b":"y"}', '$.a'),
+        |  (NULL, '$.a'),
+        |  ('{"a":null}', '$.a')
+        |AS t(value, path)
+        |""".stripMargin)
+
+    checkAnswer(df, Seq(Row("x"), Row(null), Row(null), Row(null)))
+  }
+}

gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -863,6 +863,7 @@ class VeloxTestSettings extends BackendTestSettings {
   enableSuite[GlutenVariantSuite]
     // TODO: Velox parquet writer marks all struct fields as OPTIONAL (nullable),
     //  but Spark's variant type requires REQUIRED fields. Needs Velox-side fix.
+    .exclude("non-literal variant_get")
     .exclude("SPARK-47546: invalid variant binary")
     .exclude("SPARK-47546: valid variant binary")
   enableSuite[GlutenVariantWriteShreddingSuite]
@@ -910,6 +911,9 @@ class VeloxTestSettings extends BackendTestSettings {
       // Velox's collect_list / collect_set are by design declarative aggregate so plan check
       // for ObjectHashAggregateExec will fail. Overriden
       "SPARK-22223: ObjectHashAggregate should not introduce unnecessary shuffle",
+      // Negative HLL tests throw SparkException on Gluten because executor-side failures are
+      // wrapped at the driver boundary.
+      "SPARK-16484: hll_*_agg + hll_union negative tests",
       "SPARK-31620: agg with subquery (whole-stage-codegen = true)",
       "SPARK-31620: agg with subquery (whole-stage-codegen = false)"
     )

gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenDataFrameAggregateSuite.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql
 import org.apache.gluten.config.GlutenConfig
 import org.apache.gluten.execution.{HashAggregateExecBaseTransformer, HashAggregateExecTransformer}
 
+import org.apache.spark.SparkException
 import org.apache.spark.sql.execution.WholeStageCodegenExec
 import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, ObjectHashAggregateExec, SortAggregateExec}
 import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
@@ -313,6 +314,96 @@ class GlutenDataFrameAggregateSuite extends DataFrameAggregateSuite with GlutenS
     }
   }
 
+  testGluten("SPARK-16484: hll_*_agg + hll_union negative tests") {
+    val df1 = Seq(
+      (1, "a"), (1, "a"), (1, "a"),
+      (1, "b"),
+      (1, "c"), (1, "c"),
+      (1, "d")
+    ).toDF("id", "value")
+    df1.createOrReplaceTempView("df1")
+
+    val df2 = Seq(
+      (1, "a"),
+      (1, "c"),
+      (1, "d"), (1, "d"), (1, "d"),
+      (1, "e"), (1, "e"),
+      (1, "f")
+    ).toDF("id", "value")
+    df2.createOrReplaceTempView("df2")
+
+    def assertWrappedSparkThrowable(
+        e: SparkException,
+        errorClass: String,
+        messageFragments: Seq[String]): Unit = {
+      val combinedMessage =
+        Option(e.getMessage).getOrElse("") + "\n" + Option(e.getCause).map(_.toString).getOrElse("")
+      assert(combinedMessage.contains(errorClass))
+      messageFragments.foreach(fragment => assert(combinedMessage.contains(fragment)))
+    }
+
+    val invalidLow = intercept[SparkException] {
+      val res = df1.groupBy("id")
+        .agg(
+          hll_sketch_agg("value", 1).as("hllsketch")
+        )
+      checkAnswer(res, Nil)
+    }
+    assertWrappedSparkThrowable(
+      invalidLow,
+      "HLL_INVALID_LG_K",
+      Seq("`hll_sketch_agg`", "1"))
+
+    val invalidHigh = intercept[SparkException] {
+      val res = df1.groupBy("id")
+        .agg(
+          hll_sketch_agg("value", 25).as("hllsketch")
+        )
+      checkAnswer(res, Nil)
+    }
+    assertWrappedSparkThrowable(
+      invalidHigh,
+      "HLL_INVALID_LG_K",
+      Seq("`hll_sketch_agg`", "25"))
+
+    val unionError = intercept[SparkException] {
+      val i1 = df1.groupBy("id")
+        .agg(
+          hll_sketch_agg("value").as("hllsketch_left")
+        )
+      val i2 = df2.groupBy("id")
+        .agg(
+          hll_sketch_agg("value", 20).as("hllsketch_right")
+        )
+      val res = i1.join(i2).withColumn("union", hll_union("hllsketch_left", "hllsketch_right"))
+      checkAnswer(res, Nil)
+    }
+    assertWrappedSparkThrowable(
+      unionError,
+      "HLL_UNION_DIFFERENT_LG_K",
+      Seq("`hll_union`", "12", "20"))
+
+    val unionAggError = intercept[SparkException] {
+      val i1 = df1.groupBy("id")
+        .agg(
+          hll_sketch_agg("value").as("hllsketch")
+        )
+      val i2 = df2.groupBy("id")
+        .agg(
+          hll_sketch_agg("value", 20).as("hllsketch")
+        )
+      val res = i1.union(i2).groupBy("id")
+        .agg(
+          hll_union_agg("hllsketch")
+        )
+      checkAnswer(res, Nil)
+    }
+    assertWrappedSparkThrowable(
+      unionAggError,
+      "HLL_UNION_DIFFERENT_LG_K",
+      Seq("`hll_union_agg`", "12", "20"))
+  }
+
   Seq(true, false).foreach {
     value =>
       testGluten(s"SPARK-31620: agg with subquery (whole-stage-codegen = $value)") {

gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenVariantSuite.scala

@@ -16,4 +16,19 @@
  */
 package org.apache.spark.sql
 
-class GlutenVariantSuite extends VariantSuite with GlutenSQLTestsTrait {}
+class GlutenVariantSuite extends VariantSuite with GlutenSQLTestsTrait {
+  testGluten("non-literal variant_get") {
+    val df = spark.sql(
+      """
+        |SELECT variant_get(parse_json(value), path, 'string')
+        |FROM VALUES
+        |  ('{"a":"x"}', '$.a'),
+        |  ('{"b":"y"}', '$.a'),
+        |  (NULL, '$.a'),
+        |  ('{"a":null}', '$.a')
+        |AS t(value, path)
+        |""".stripMargin)
+
+    checkAnswer(df, Seq(Row("x"), Row(null), Row(null), Row(null)))
+  }
+}