[GLUTEN-5391][CH] Fix equalTo NaN issue (#5391) (#5392)

[GLUTEN-5391][CH] Fix equalTo NaN issue

backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHSparkPlanExecApi.scala

@@ -35,7 +35,7 @@ import org.apache.spark.serializer.Serializer
 import org.apache.spark.shuffle.{GenShuffleWriterParameters, GlutenShuffleWriterWrapper, HashPartitioningWrapper}
 import org.apache.spark.shuffle.utils.CHShuffleUtil
 import org.apache.spark.sql.{SparkSession, Strategy}
-import org.apache.spark.sql.catalyst.CHAggregateFunctionRewriteRule
+import org.apache.spark.sql.catalyst.{CHAggregateFunctionRewriteRule, EqualToRewrite}
 import org.apache.spark.sql.catalyst.catalog.BucketSpec
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions._
@@ -585,7 +585,8 @@ class CHSparkPlanExecApi extends SparkPlanExecApi {
     List(
       spark => new CommonSubexpressionEliminateRule(spark, spark.sessionState.conf),
       spark => CHAggregateFunctionRewriteRule(spark),
-      _ => CountDistinctWithoutExpand
+      _ => CountDistinctWithoutExpand,
+      _ => EqualToRewrite
     )
   }
 

backends-clickhouse/src/main/scala/org/apache/spark/sql/catalyst/EqualToRewrite.scala

@@ -0,0 +1,31 @@
+/*
+ * 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.spark.sql.catalyst
+
+import org.apache.spark.sql.catalyst.expressions.{DoubleLiteral, EqualTo, FloatLiteral, IsNaN}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+
+/** The result of 'equal to NaN' and isNaN is different in CH. */
+object EqualToRewrite extends Rule[LogicalPlan] {
+  def apply(plan: LogicalPlan): LogicalPlan = plan.transformExpressions {
+    case EqualTo(left, FloatLiteral(f)) if f.isNaN => IsNaN(left)
+    case EqualTo(left, DoubleLiteral(d)) if d.isNaN => IsNaN(left)
+    case EqualTo(FloatLiteral(f), right) if f.isNaN => IsNaN(right)
+    case EqualTo(DoubleLiteral(d), right) if d.isNaN => IsNaN(right)
+  }
+}

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

@@ -655,4 +655,16 @@ class GlutenFunctionValidateSuite extends GlutenClickHouseWholeStageTransformerS
     }
   }
 
+  test("equalTo rewrite to isNaN") {
+    withTable("tb_scrt") {
+      sql("create table tb_scrt(id int) using parquet")
+      sql("""
+            |insert into tb_scrt values (-2147483648),(-2147483648)
+            |""".stripMargin)
+      val q = "select sqrt(id),sqrt(id)='NaN' from tb_scrt"
+      runQueryAndCompare(q)(checkGlutenOperatorMatch[ProjectExecTransformer])
+    }
+
+  }
+
 }