fix ci

Author: KevinyhZou

backends-clickhouse/src/main/scala/org/apache/gluten/execution/CHFilterExecTransformer.scala

@@ -16,7 +16,7 @@
  */
 package org.apache.gluten.execution
 
-import org.apache.spark.sql.catalyst.expressions.{And, Expression}
+import org.apache.spark.sql.catalyst.expressions.{And, Expression, ExprId, IsNotNull, ScalaUDF}
 import org.apache.spark.sql.execution.SparkPlan
 
 case class CHFilterExecTransformer(condition: Expression, child: SparkPlan)
@@ -48,4 +48,13 @@ case class FilterExecTransformer(condition: Expression, child: SparkPlan)
   override protected def getRemainingCondition: Expression = condition
   override protected def withNewChildInternal(newChild: SparkPlan): FilterExecTransformer =
     copy(child = newChild)
+  override protected val notNullAttributes: Seq[ExprId] = condition match {
+    case s: ScalaUDF =>
+      val (notNullPreds, _) = s.children.partition {
+        case IsNotNull(a) => isNullIntolerant(a) && a.references.subsetOf(child.outputSet)
+        case _ => false
+      }
+      notNullPreds.flatMap(_.references).distinct.map(_.exprId)
+    case _ => notNullPreds.flatMap(_.references).distinct.map(_.exprId)
+  }
 }

backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenFunctionValidateSuite.scala

@@ -63,7 +63,6 @@ class GlutenFunctionValidateSuite extends GlutenClickHouseWholeStageTransformerS
       .set("spark.io.compression.codec", "snappy")
       .set("spark.sql.shuffle.partitions", "5")
       .set("spark.sql.autoBroadcastJoinThreshold", "10MB")
-      .set("spark.gluten.sql.supported.collapseNestedFunctions", "and,or")
   }
 
   override def beforeAll(): Unit = {
@@ -208,10 +207,6 @@ class GlutenFunctionValidateSuite extends GlutenClickHouseWholeStageTransformerS
     spark.catalog.createTable("url_table", urlFilePath, fileFormat)
   }
 
-  override def afterAll(): Unit = {
-    sparkConf.set("spark.gluten.sql.supported.collapseNestedFunctions", "")
-  }
-
   test("Test get_json_object 1") {
     runQueryAndCompare("SELECT get_json_object(string_field1, '$.a') from json_test") {
       checkGlutenOperatorMatch[ProjectExecTransformer]
@@ -399,21 +394,15 @@ class GlutenFunctionValidateSuite extends GlutenClickHouseWholeStageTransformerS
       }
       false
     }
-    try {
-      withSQLConf(("spark.gluten.sql.supported.collapseNestedFunctions", "and,or")) {
-        runQueryAndCompare(
-          "SELECT count(1) from json_test where int_field1 = 5 and double_field1 > 1.0" +
-            " and string_field1 is not null") {
-          x => assert(checkCollapsedFunctions(x.queryExecution.executedPlan, "and", 5))
-        }
-        runQueryAndCompare(
-          "SELECT count(1) from json_test where int_field1 = 5 or double_field1 > 1.0" +
-            " or string_field1 is not null") {
-          x => assert(checkCollapsedFunctions(x.queryExecution.executedPlan, "or", 3))
-        }
-      }
-    } finally {
-      sparkConf.set("spark.gluten.sql.supported.collapseNestedFunctions", "")
+    runQueryAndCompare(
+      "SELECT count(1) from json_test where int_field1 = 5 and double_field1 > 1.0" +
+        " and string_field1 is not null") {
+      x => assert(checkCollapsedFunctions(x.queryExecution.executedPlan, "and", 5))
+    }
+    runQueryAndCompare(
+      "SELECT count(1) from json_test where int_field1 = 5 or double_field1 > 1.0" +
+        " or string_field1 is not null") {
+      x => assert(checkCollapsedFunctions(x.queryExecution.executedPlan, "or", 3))
     }
   }
 

shims/common/src/main/scala/org/apache/gluten/config/GlutenConfig.scala

@@ -696,7 +696,7 @@ object GlutenConfig {
       .internal()
       .doc("Collapse nested functions as one for optimization.")
       .stringConf
-      .createWithDefault("");
+      .createWithDefault("and,or");
 
   val GLUTEN_SOFT_AFFINITY_ENABLED =
     buildConf("spark.gluten.soft-affinity.enabled")