[GLUTEN-3924][CORE] Match hive UDF name in case-insensitive mode duri…

…ng expression transformation (#3925)

* case-insensitive matching for hive udfs

* add uts

* fix failed uts

* fix failed uts

backends-clickhouse/src/test/scala/io/glutenproject/execution/GlutenClickHouseHiveTableSuite.scala

@@ -97,6 +97,7 @@ class GlutenClickHouseHiveTableSuite()
         "spark.sql.warehouse.dir",
         getClass.getResource("/").getPath + "unit-tests-working-home/spark-warehouse")
       .set("spark.hive.exec.dynamic.partition.mode", "nonstrict")
+      .set("spark.gluten.supported.hive.udfs", "my_add")
       .setMaster("local[*]")
   }
 
@@ -1060,4 +1061,14 @@ class GlutenClickHouseHiveTableSuite()
     compareResultsAgainstVanillaSpark(select_sql, compareResult = true, _ => {})
     spark.sql("DROP TABLE test_tbl_3548")
   }
+
+  test("test 'hive udf'") {
+    val jarPath = "src/test/resources/udfs/hive-test-udfs.jar"
+    val jarUrl = s"file://${System.getProperty("user.dir")}/$jarPath"
+    spark.sql(
+      s"CREATE FUNCTION my_add as " +
+        s"'org.apache.hadoop.hive.contrib.udf.example.UDFExampleAdd2' USING JAR '$jarUrl'")
+    runQueryAndCompare("select MY_ADD(id, id+1) from range(10)")(
+      checkOperatorMatch[ProjectExecTransformer])
+  }
 }

backends-clickhouse/src/test/scala/io/glutenproject/execution/GlutenClickHouseTPCHParquetSuite.scala

@@ -48,7 +48,6 @@ class GlutenClickHouseTPCHParquetSuite extends GlutenClickHouseTPCHAbstractSuite
       .set("spark.sql.autoBroadcastJoinThreshold", "10MB")
       .set("spark.gluten.sql.columnar.backend.ch.use.v2", "false")
       .set("spark.gluten.supported.scala.udfs", "my_add")
-      .set("spark.gluten.supported.hive.udfs", "my_add")
   }
 
   override protected val createNullableTables = true
@@ -1319,16 +1318,6 @@ class GlutenClickHouseTPCHParquetSuite extends GlutenClickHouseTPCHAbstractSuite
       checkOperatorMatch[ProjectExecTransformer])
   }
 
-  ignore("test 'hive udf'") {
-    val jarPath = "backends-clickhouse/src/test/resources/udfs/hive-test-udfs.jar"
-    val jarUrl = s"file://${System.getProperty("user.dir")}/$jarPath"
-    spark.sql(
-      s"CREATE FUNCTION my_add as " +
-        "'org.apache.hadoop.hive.contrib.udf.example.UDFExampleAdd2' USING JAR '$jarUrl'")
-    runQueryAndCompare("select my_add(id, id+1) from range(10)")(
-      checkOperatorMatch[ProjectExecTransformer])
-  }
-
   override protected def runTPCHQuery(
       queryNum: Int,
       tpchQueries: String = tpchQueries,

cpp-ch/local-engine/Parser/FunctionParser.h

@@ -76,7 +76,7 @@ class FunctionParser
     {
         return plan_parser->toFunctionNode(action_dag, func_name, args);
     }
-    
+
     const DB::ActionsDAG::Node *
     toFunctionNode(DB::ActionsDAGPtr & action_dag, const String & func_name, const String & result_name, const DB::ActionsDAG::NodeRawConstPtrs & args) const
     {

cpp-ch/local-engine/tests/gtest_parquet_write.cpp

@@ -199,7 +199,7 @@ TEST(ParquetWrite, ComplexTypes)
     ch2arrow.chChunkToArrowTable(arrow_table, input_chunks, header.columns());
 
     /// Convert Arrow Table to CH Block
-    ArrowColumnToCHColumn arrow2ch(header, "Parquet", true, true, true);
+    ArrowColumnToCHColumn arrow2ch(header, "Parquet", true, true, FormatSettings::DateTimeOverflowBehavior::Ignore);
     Chunk output_chunk;
     arrow2ch.arrowTableToCHChunk(output_chunk, arrow_table, arrow_table->num_rows());
 

gluten-core/src/main/scala/io/glutenproject/expression/UDFMappings.scala

@@ -23,6 +23,8 @@ import org.apache.spark.internal.Logging
 
 import org.apache.commons.lang3.StringUtils
 
+import java.util.Locale
+
 import scala.collection.mutable.Map
 
 object UDFMappings extends Logging {
@@ -41,7 +43,7 @@ object UDFMappings extends Logging {
           s"will be replaced by value:$value")
     }
 
-    res.put(key, value)
+    res.put(key.toLowerCase(Locale.ROOT), value)
   }
 
   private def parseStringToMap(input: String, res: Map[String, String]) {

gluten-core/src/main/scala/org/apache/spark/sql/hive/HiveSimpleUDFTransformer.scala

@@ -20,6 +20,8 @@ import io.glutenproject.expression.{ExpressionConverter, ExpressionTransformer,
 
 import org.apache.spark.sql.catalyst.expressions._
 
+import java.util.Locale
+
 object HiveSimpleUDFTransformer {
   def isHiveSimpleUDF(expr: Expression): Boolean = {
     expr match {
@@ -36,15 +38,18 @@ object HiveSimpleUDFTransformer {
     }
 
     val udf = expr.asInstanceOf[HiveSimpleUDF]
-    val substraitExprName = UDFMappings.hiveUDFMap.get(udf.name.stripPrefix("default."))
+    val substraitExprName =
+      UDFMappings.hiveUDFMap.get(udf.name.stripPrefix("default.").toLowerCase(Locale.ROOT))
     substraitExprName match {
       case Some(name) =>
         GenericExpressionTransformer(
           name,
           udf.children.map(ExpressionConverter.replaceWithExpressionTransformer(_, attributeSeq)),
           udf)
       case _ =>
-        throw new UnsupportedOperationException(s"Not supported hive simple udf: $udf.")
+        throw new UnsupportedOperationException(
+          s"Not supported hive simple udf:$udf"
+            + s" name:${udf.name} hiveUDFMap:${UDFMappings.hiveUDFMap}")
     }
   }
 }