feat: Enable min/max for boolean type (apache#165)

snmvaughan · Mar 5, 2024 · 852b8cf · 852b8cf
1 parent 72398a6
commit 852b8cf
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 1 deletion.
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -181,7 +181,7 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde {
 
   private def minMaxDataTypeSupported(dt: DataType): Boolean = {
     dt match {
-      case _: NumericType | DateType | TimestampType => true
+      case _: NumericType | DateType | TimestampType | BooleanType => true
       case _ => false
     }
   }

diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
@@ -886,6 +886,32 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("test bool_and/bool_or") {
+    withSQLConf(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true") {
+      Seq(true, false).foreach { bosonColumnShuffleEnabled =>
+        withSQLConf(
+          CometConf.COMET_COLUMNAR_SHUFFLE_ENABLED.key -> bosonColumnShuffleEnabled.toString) {
+          Seq(true, false).foreach { dictionary =>
+            withSQLConf("parquet.enable.dictionary" -> dictionary.toString) {
+              val table = "test"
+              withTable(table) {
+                sql(s"create table $table(a boolean, b int) using parquet")
+                sql(s"insert into $table values(true, 1)")
+                sql(s"insert into $table values(false, 2)")
+                sql(s"insert into $table values(true, 3)")
+                sql(s"insert into $table values(true, 3)")
+                // Spark maps BOOL_AND to MIN and BOOL_OR to MAX
+                checkSparkAnswerAndNumOfAggregates(
+                  s"SELECT MIN(a), MAX(a), BOOL_AND(a), BOOL_OR(a) FROM $table",
+                  2)
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+
   protected def checkSparkAnswerAndNumOfAggregates(query: String, numAggregates: Int): Unit = {
     val df = sql(query)
     checkSparkAnswer(df)