[GLUTEN] Route bitmap_or_agg to native Velox execution

backends-clickhouse/src/main/scala/org/apache/gluten/utils/CHExpressionUtil.scala

@@ -201,6 +201,7 @@ object CHExpressionUtil {
     REGR_INTERCEPT -> DefaultValidator(),
     REGR_SXY -> DefaultValidator(),
     BITMAP_CONSTRUCT_AGG -> DefaultValidator(),
+    BITMAP_OR_AGG -> DefaultValidator(),
     TO_UTC_TIMESTAMP -> UtcTimestampValidator(),
     FROM_UTC_TIMESTAMP -> UtcTimestampValidator(),
     STACK -> DefaultValidator(),

cpp/velox/substrait/SubstraitToVeloxPlanValidator.cc

@@ -1324,7 +1324,8 @@ bool SubstraitToVeloxPlanValidator::validate(const ::substrait::AggregateRel& ag
       "regr_intercept",
       "regr_sxy",
       "regr_replacement",
-      "bitmap_construct_agg"};
+      "bitmap_construct_agg",
+      "bitmap_or_agg"};
 
   auto udafFuncs = UdfLoader::getInstance()->getRegisteredUdafNames();
 

gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala

@@ -92,6 +92,7 @@ class ClickHouseTestSettings extends BackendTestSettings {
   enableSuite[GlutenBitmapExpressionsQuerySuite]
     // bitmap_construct_agg is not supported natively in CH backend.
     .excludeCH("bitmap_construct_agg routes to native")
+    .excludeCH("bitmap_or_agg routes to native")
   enableSuite[GlutenBitwiseExpressionsSuite]
   enableSuite[GlutenBloomFilterAggregateQuerySuite]
     .excludeCH("Test bloom_filter_agg and might_contain")

gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenBitmapExpressionsQuerySuite.scala

@@ -37,4 +37,19 @@ class GlutenBitmapExpressionsQuerySuite
       "Expected native HashAggregateExecBaseTransformer in plan"
     )
   }
+
+  test("bitmap_or_agg routes to native") {
+    val df = spark.sql(
+      "SELECT bitmap_or_agg(bm) FROM (" +
+        "SELECT bitmap_construct_agg(bitmap_bit_position(col)) AS bm " +
+        "FROM values (1L), (2L), (3L) AS t(col)" +
+        ") sub")
+    df.collect()
+    assert(
+      collectWithSubqueries(df.queryExecution.executedPlan) {
+        case h: HashAggregateExecBaseTransformer => h
+      }.nonEmpty,
+      "Expected native HashAggregateExecBaseTransformer in plan"
+    )
+  }
 }

gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala

@@ -91,6 +91,7 @@ class ClickHouseTestSettings extends BackendTestSettings {
   enableSuite[GlutenBitmapExpressionsQuerySuite]
     // bitmap_construct_agg is not supported natively in CH backend.
     .excludeCH("bitmap_construct_agg routes to native")
+    .excludeCH("bitmap_or_agg routes to native")
   enableSuite[GlutenBitwiseExpressionsSuite]
   enableSuite[GlutenBloomFilterAggregateQuerySuite]
     .excludeCH("Test bloom_filter_agg and might_contain")

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

@@ -37,4 +37,19 @@ class GlutenBitmapExpressionsQuerySuite
       "Expected native HashAggregateExecBaseTransformer in plan"
     )
   }
+
+  test("bitmap_or_agg routes to native") {
+    val df = spark.sql(
+      "SELECT bitmap_or_agg(bm) FROM (" +
+        "SELECT bitmap_construct_agg(bitmap_bit_position(col)) AS bm " +
+        "FROM values (1L), (2L), (3L) AS t(col)" +
+        ") sub")
+    df.collect()
+    assert(
+      collectWithSubqueries(df.queryExecution.executedPlan) {
+        case h: HashAggregateExecBaseTransformer => h
+      }.nonEmpty,
+      "Expected native HashAggregateExecBaseTransformer in plan"
+    )
+  }
 }

gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/clickhouse/ClickHouseTestSettings.scala

@@ -91,6 +91,7 @@ class ClickHouseTestSettings extends BackendTestSettings {
   enableSuite[GlutenBitmapExpressionsQuerySuite]
     // bitmap_construct_agg is not supported natively in CH backend.
     .excludeCH("bitmap_construct_agg routes to native")
+    .excludeCH("bitmap_or_agg routes to native")
   enableSuite[GlutenBitwiseExpressionsSuite]
   enableSuite[GlutenBloomFilterAggregateQuerySuite]
     .excludeCH("Test bloom_filter_agg and might_contain")

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

@@ -37,4 +37,19 @@ class GlutenBitmapExpressionsQuerySuite
       "Expected native HashAggregateExecBaseTransformer in plan"
     )
   }
+
+  test("bitmap_or_agg routes to native") {
+    val df = spark.sql(
+      "SELECT bitmap_or_agg(bm) FROM (" +
+        "SELECT bitmap_construct_agg(bitmap_bit_position(col)) AS bm " +
+        "FROM values (1L), (2L), (3L) AS t(col)" +
+        ") sub")
+    df.collect()
+    assert(
+      collectWithSubqueries(df.queryExecution.executedPlan) {
+        case h: HashAggregateExecBaseTransformer => h
+      }.nonEmpty,
+      "Expected native HashAggregateExecBaseTransformer in plan"
+    )
+  }
 }

shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala

@@ -33,6 +33,7 @@ object ExpressionNames {
   final val COLLECT_SET = "collect_set"
   final val BLOOM_FILTER_AGG = "bloom_filter_agg"
   final val BITMAP_CONSTRUCT_AGG = "bitmap_construct_agg"
+  final val BITMAP_OR_AGG = "bitmap_or_agg"
   final val VAR_SAMP = "var_samp"
   final val VAR_POP = "var_pop"
   final val BIT_AND_AGG = "bit_and"

shims/spark35/src/main/scala/org/apache/gluten/sql/shims/spark35/Spark35Shims.scala

@@ -81,7 +81,8 @@ class Spark35Shims extends SparkShims {
       Sig[RegrIntercept](ExpressionNames.REGR_INTERCEPT),
       Sig[RegrSXY](ExpressionNames.REGR_SXY),
       Sig[RegrReplacement](ExpressionNames.REGR_REPLACEMENT),
-      Sig[BitmapConstructAgg](ExpressionNames.BITMAP_CONSTRUCT_AGG)
+      Sig[BitmapConstructAgg](ExpressionNames.BITMAP_CONSTRUCT_AGG),
+      Sig[BitmapOrAgg](ExpressionNames.BITMAP_OR_AGG)
     )
   }
 

shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala

@@ -86,7 +86,8 @@ class Spark40Shims extends SparkShims {
       Sig[RegrIntercept](ExpressionNames.REGR_INTERCEPT),
       Sig[RegrSXY](ExpressionNames.REGR_SXY),
       Sig[RegrReplacement](ExpressionNames.REGR_REPLACEMENT),
-      Sig[BitmapConstructAgg](ExpressionNames.BITMAP_CONSTRUCT_AGG)
+      Sig[BitmapConstructAgg](ExpressionNames.BITMAP_CONSTRUCT_AGG),
+      Sig[BitmapOrAgg](ExpressionNames.BITMAP_OR_AGG)
     )
   }
 

shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala

@@ -85,7 +85,8 @@ class Spark41Shims extends SparkShims {
       Sig[RegrIntercept](ExpressionNames.REGR_INTERCEPT),
       Sig[RegrSXY](ExpressionNames.REGR_SXY),
       Sig[RegrReplacement](ExpressionNames.REGR_REPLACEMENT),
-      Sig[BitmapConstructAgg](ExpressionNames.BITMAP_CONSTRUCT_AGG)
+      Sig[BitmapConstructAgg](ExpressionNames.BITMAP_CONSTRUCT_AGG),
+      Sig[BitmapOrAgg](ExpressionNames.BITMAP_OR_AGG)
     )
   }