[GLUTEN-9106][VL] Add support for staticInvoke CharVarcharCodegenUtils. #9107

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

@@ -207,6 +207,9 @@ object CHExpressionUtil {
     MAKE_DATE -> DefaultValidator(),
     ARRAY_APPEND -> DefaultValidator(),
     JSON_OBJECT_KEYS -> DefaultValidator(),
-    LUHN_CHECK -> DefaultValidator()
+    LUHN_CHECK -> DefaultValidator(),
+    VARCHAR_TYPE_WRITE_SIDE_CHECK -> DefaultValidator(),
+    CHAR_TYPE_WRITE_SIDE_CHECK -> DefaultValidator(),
+    READ_SIDE_PADDING -> DefaultValidator()
   )
 }

backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala

@@ -622,6 +622,47 @@ abstract class ScalarFunctionsValidateSuite extends FunctionsValidateSuite {
     }
   }
 
+  // Add test suite for CharVarcharCodegenUtils functions.
+  // A ProjectExecTransformer is expected to be constructed after expr support.
+  // We currently test below functions with Spark v3.4
+  testWithMinSparkVersion("charTypeWriteSideCheck", "3.4") {
+    withTable("src", "dest") {
+
+      sql("create table src(id string) USING PARQUET")
+      sql("insert into src values('s')")
+      sql("create table dest(id char(3)) USING PARQUET")
+      // check whether the executed plan of a dataframe contains the expected plan.
+      runQueryAndCompare("insert into dest select id from src") {
+        checkGlutenOperatorMatch[ProjectExecTransformer]
+      }
+    }
+  }
+
+  testWithMinSparkVersion("varcharTypeWriteSideCheck", "3.4") {
+    withTable("src", "dest") {
+
+      sql("create table src(id string) USING PARQUET")
+      sql("insert into src values('abc')")
+      sql("create table dest(id varchar(10)) USING PARQUET")
+      // check whether the executed plan of a dataframe contains the expected plan.
+      runQueryAndCompare("insert into dest select id from src") {
+        checkGlutenOperatorMatch[ProjectExecTransformer]
+      }
+    }
+  }
+
+  testWithMinSparkVersion("readSidePadding", "3.4") {
+    withTable("src", "dest") {
+
+      sql("create table tgt(id char(3)) USING PARQUET")
+      sql("insert into tgt values('p')")
+      // check whether the executed plan of a dataframe contains the expected plan.
+      runQueryAndCompare("select id from tgt") {
+        checkGlutenOperatorMatch[ProjectExecTransformer]
+      }
+    }
+  }
+
   test("soundex") {
     runQueryAndCompare("select soundex(c_comment) from customer limit 50") {
       checkGlutenOperatorMatch[ProjectExecTransformer]

gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala

@@ -144,6 +144,66 @@ object ExpressionConverter extends SQLConfHelper with Logging {
     DecimalArithmeticExpressionTransformer(substraitName, leftChild, rightChild, resultType, b)
   }
 
+  private def replaceStaticInvokeWithExpressionTransformer(
+      i: StaticInvoke,
+      attributeSeq: Seq[Attribute],
+      expressionsMap: Map[Class[_], String]): ExpressionTransformer = {
+    def validateAndTransform(
+        exprName: String,
+        childTransformers: => Seq[ExpressionTransformer]): ExpressionTransformer = {
+      if (!BackendsApiManager.getValidatorApiInstance.doExprValidate(exprName, i)) {
+        throw new GlutenNotSupportException(
+          s"Not supported to map current ${i.getClass} call on function: ${i.functionName}.")
+      }
+      GenericExpressionTransformer(exprName, childTransformers, i)
+    }
+
+    i.functionName match {
+      case "encode" | "decode" if i.objectName.endsWith("UrlCodec") =>
+        validateAndTransform(
+          "url_" + i.functionName,
+          Seq(replaceWithExpressionTransformer0(i.arguments.head, attributeSeq, expressionsMap))
+        )
+
+      case "isLuhnNumber" =>
+        validateAndTransform(
+          ExpressionNames.LUHN_CHECK,
+          Seq(replaceWithExpressionTransformer0(i.arguments.head, attributeSeq, expressionsMap))
+        )
+
+      case "encode" | "decode" if i.objectName.endsWith("Base64") =>
+        if (!BackendsApiManager.getValidatorApiInstance.doExprValidate(ExpressionNames.BASE64, i)) {
+          throw new GlutenNotSupportException(
+            s"Not supported to map current ${i.getClass} call on function: ${i.functionName}.")
+        }
+        BackendsApiManager.getSparkPlanExecApiInstance.genBase64StaticInvokeTransformer(
+          ExpressionNames.BASE64,
+          replaceWithExpressionTransformer0(i.arguments.head, attributeSeq, expressionsMap),
+          i
+        )
+
+      case fn
+          if i.objectName.endsWith("CharVarcharCodegenUtils") && Set(
+            "varcharTypeWriteSideCheck",
+            "charTypeWriteSideCheck",
+            "readSidePadding").contains(fn) =>
+        val exprName = fn match {
+          case "varcharTypeWriteSideCheck" => ExpressionNames.VARCHAR_TYPE_WRITE_SIDE_CHECK
+          case "charTypeWriteSideCheck" => ExpressionNames.CHAR_TYPE_WRITE_SIDE_CHECK
+          case "readSidePadding" => ExpressionNames.READ_SIDE_PADDING
+        }
+        validateAndTransform(
+          exprName,
+          i.arguments.map(replaceWithExpressionTransformer0(_, attributeSeq, expressionsMap))
+        )
+
+      case _ =>
+        throw new GlutenNotSupportException(
+          s"Not supported to transform StaticInvoke with object: ${i.staticObject.getName}, " +
+            s"function: ${i.functionName}")
+    }
+  }
+
   private def replaceIcebergStaticInvoke(
       s: StaticInvoke,
       attributeSeq: Seq[Attribute],
@@ -186,33 +246,12 @@ object ExpressionConverter extends SQLConfHelper with Logging {
         return BackendsApiManager.getSparkPlanExecApiInstance.genHiveUDFTransformer(
           expr,
           attributeSeq)
-      case i: StaticInvoke
-          if Seq("encode", "decode").contains(i.functionName) && i.objectName.endsWith(
-            "UrlCodec") =>
-        return GenericExpressionTransformer(
-          "url_" + i.functionName,
-          replaceWithExpressionTransformer0(i.arguments.head, attributeSeq, expressionsMap),
-          i)
-      case i: StaticInvoke if i.functionName.equals("isLuhnNumber") =>
-        return GenericExpressionTransformer(
-          ExpressionNames.LUHN_CHECK,
-          replaceWithExpressionTransformer0(i.arguments.head, attributeSeq, expressionsMap),
-          i)
-      case i: StaticInvoke
-          if Seq("encode", "decode").contains(i.functionName) && i.objectName.endsWith("Base64") =>
-        return BackendsApiManager.getSparkPlanExecApiInstance.genBase64StaticInvokeTransformer(
-          ExpressionNames.BASE64,
-          replaceWithExpressionTransformer0(i.arguments.head, attributeSeq, expressionsMap),
-          i
-        )
       case i: StaticInvoke
           if i.functionName == "invoke" && i.staticObject.getName.startsWith(
             "org.apache.iceberg.spark.functions.") =>
         return replaceIcebergStaticInvoke(i, attributeSeq, expressionsMap)
       case i: StaticInvoke =>
-        throw new GlutenNotSupportException(
-          s"Not supported to transform StaticInvoke with object: ${i.staticObject.getName}, " +
-            s"function: ${i.functionName}")
+        return replaceStaticInvokeWithExpressionTransformer(i, attributeSeq, expressionsMap)
       case _ =>
     }
 

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

@@ -135,6 +135,11 @@ class ClickHouseTestSettings extends BackendTestSettings {
   enableSuite[GlutenDSV2CharVarcharTestSuite]
     // failed on spark32 UT, see https://github.com/oap-project/gluten/issues/4043
     .exclude("SPARK-34833: right-padding applied correctly for correlated subqueries - other preds")
+    // Excluded. The Gluten tests for char/varchar validation were rewritten for Velox.
+    // ClickHouse backend doesn't support this feature and falls back to vanilla Spark,
+    // causing mismatches in error messages.
+    .excludeGlutenTest("length check for input string values: nested in struct")
+    .excludeGlutenTest("length check for input string values: nested in struct of array")
   enableSuite[GlutenDSV2SQLInsertTestSuite]
   enableSuite[GlutenDataFrameAggregateSuite]
     .exclude("average")
@@ -354,6 +359,11 @@ class ClickHouseTestSettings extends BackendTestSettings {
     .exclude("varchar type values length check and trim: partitioned columns")
     .exclude("char/varchar type values length check: partitioned columns of other types")
     .exclude("char type comparison: partitioned columns")
+    // Excluded. The Gluten tests for char/varchar validation were rewritten for Velox.
+    // ClickHouse backend doesn't support this feature and falls back to vanilla Spark,
+    // causing mismatches in error messages.
+    .excludeGlutenTest("length check for input string values: nested in struct")
+    .excludeGlutenTest("length check for input string values: nested in struct of array")
   enableSuite[GlutenFileSourceSQLInsertTestSuite]
     .exclude("SPARK-33474: Support typed literals as partition spec values")
     .exclude(

gluten-ut/spark32/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -422,7 +422,17 @@ class VeloxTestSettings extends BackendTestSettings {
     .excludeByPrefix("SPARK-24705")
     .excludeByPrefix("determining the number of reducers")
   enableSuite[GlutenFileSourceCharVarcharTestSuite]
+    // Following tests are excluded as these are overridden in Gluten test suite..
+    // The overridden tests assert against Velox-specific error messages for char/varchar
+    // length validation, which differ from the original vanilla Spark tests.
+    .exclude("length check for input string values: nested in struct")
+    .exclude("length check for input string values: nested in struct of array")
   enableSuite[GlutenDSV2CharVarcharTestSuite]
+    // Following tests are excluded as these are overridden in Gluten test suite..
+    // The overridden tests assert against Velox-specific error messages for char/varchar
+    // length validation, which differ from the original vanilla Spark tests.
+    .exclude("length check for input string values: nested in struct")
+    .exclude("length check for input string values: nested in struct of array")
   enableSuite[GlutenFileScanSuite]
   enableSuite[GlutenNestedDataSourceV1Suite]
   enableSuite[GlutenNestedDataSourceV2Suite]

gluten-ut/spark32/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala

@@ -16,8 +16,73 @@
  */
 package org.apache.spark.sql
 
+import org.apache.spark.SparkException
+
 class GlutenFileSourceCharVarcharTestSuite
   extends FileSourceCharVarcharTestSuite
-  with GlutenSQLTestsTrait {}
+  with GlutenSQLTestsTrait {
+
+  private def testTableWrite(f: String => Unit): Unit = {
+    withTable("t")(f("char"))
+    withTable("t")(f("varchar"))
+  }
+
+  private val VELOX_ERROR_MESSAGE = "Exceeds allowed length limitation: 5"
+
+  testGluten("length check for input string values: nested in struct") {
+    testTableWrite {
+      typeName =>
+        sql(s"CREATE TABLE t(c STRUCT<c: $typeName(5)>) USING $format")
+        sql("INSERT INTO t SELECT struct(null)")
+        checkAnswer(spark.table("t"), Row(Row(null)))
+        val e = intercept[SparkException] {
+          sql("INSERT INTO t SELECT struct('123456')")
+        }
+        assert(e.getCause.getMessage.contains(VELOX_ERROR_MESSAGE))
+    }
+  }
+
+  testGluten("length check for input string values: nested in struct of array") {
+    testTableWrite {
+      typeName =>
+        sql(s"CREATE TABLE t(c STRUCT<c: ARRAY<$typeName(5)>>) USING $format")
+        sql("INSERT INTO t SELECT struct(array(null))")
+        checkAnswer(spark.table("t"), Row(Row(Seq(null))))
+        val e = intercept[SparkException](sql("INSERT INTO t SELECT struct(array('123456'))"))
+        assert(e.getCause.getMessage.contains(VELOX_ERROR_MESSAGE))
+    }
+  }
+}
+
+class GlutenDSV2CharVarcharTestSuite extends DSV2CharVarcharTestSuite with GlutenSQLTestsTrait {
+  private def testTableWrite(f: String => Unit): Unit = {
+    withTable("t")(f("char"))
+    withTable("t")(f("varchar"))
+  }
+
+  private val VELOX_ERROR_MESSAGE = "Exceeds allowed length limitation: 5"
+
+  testGluten("length check for input string values: nested in struct") {
+    testTableWrite {
+      typeName =>
+        sql(s"CREATE TABLE t(c STRUCT<c: $typeName(5)>) USING $format")
+        sql("INSERT INTO t SELECT struct(null)")
+        checkAnswer(spark.table("t"), Row(Row(null)))
+        val e = intercept[SparkException] {
+          sql("INSERT INTO t SELECT struct('123456')")
+        }
+        assert(e.getCause.getMessage.contains(VELOX_ERROR_MESSAGE))
+    }
+  }
 
-class GlutenDSV2CharVarcharTestSuite extends DSV2CharVarcharTestSuite with GlutenSQLTestsTrait {}
+  testGluten("length check for input string values: nested in struct of array") {
+    testTableWrite {
+      typeName =>
+        sql(s"CREATE TABLE t(c STRUCT<c: ARRAY<$typeName(5)>>) USING $format")
+        sql("INSERT INTO t SELECT struct(array(null))")
+        checkAnswer(spark.table("t"), Row(Row(Seq(null))))
+        val e = intercept[SparkException](sql("INSERT INTO t SELECT struct(array('123456'))"))
+        assert(e.getCause.getMessage.contains(VELOX_ERROR_MESSAGE))
+    }
+  }
+}

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

@@ -152,6 +152,10 @@ class ClickHouseTestSettings extends BackendTestSettings {
   enableSuite[GlutenCountMinSketchAggQuerySuite]
   enableSuite[GlutenCsvFunctionsSuite]
   enableSuite[GlutenDSV2CharVarcharTestSuite]
+    // Excluded. The Gluten tests for char/varchar validation were rewritten for Velox.
+    // ClickHouse backend doesn't support this feature and falls back to vanilla Spark,
+    // causing mismatches in error messages.
+    .excludeGlutenTest("length check for input string values: nested in struct of array")
   enableSuite[GlutenDSV2SQLInsertTestSuite]
   enableSuite[GlutenDataFrameAggregateSuite]
     .exclude("average")
@@ -367,6 +371,10 @@ class ClickHouseTestSettings extends BackendTestSettings {
     .exclude("varchar type values length check and trim: partitioned columns")
     .exclude("char/varchar type values length check: partitioned columns of other types")
     .exclude("char type comparison: partitioned columns")
+    // Excluded. The Gluten tests for char/varchar validation were rewritten for Velox.
+    // ClickHouse backend doesn't support this feature and falls back to vanilla Spark,
+    // causing mismatches in error messages.
+    .excludeGlutenTest("length check for input string values: nested in struct of array")
   enableSuite[GlutenFileSourceSQLInsertTestSuite]
     .exclude("SPARK-33474: Support typed literals as partition spec values")
     .exclude(

gluten-ut/spark33/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -616,7 +616,15 @@ class VeloxTestSettings extends BackendTestSettings {
     // Extra ColumnarToRow is needed to transform vanilla columnar data to gluten columnar data.
     .exclude("SPARK-37369: Avoid redundant ColumnarToRow transition on InMemoryTableScan")
   enableSuite[GlutenFileSourceCharVarcharTestSuite]
+    // Following test is excluded as it is overridden in Gluten test suite..
+    // The overridden tests assert against Velox-specific error messages for char/varchar
+    // length validation, which differ from the original vanilla Spark tests.
+    .exclude("length check for input string values: nested in struct of array")
   enableSuite[GlutenDSV2CharVarcharTestSuite]
+    // Following test is excluded as it is overridden in Gluten test suite..
+    // The overridden tests assert against Velox-specific error messages for char/varchar
+    // length validation, which differ from the original vanilla Spark tests.
+    .exclude("length check for input string values: nested in struct of array")
   enableSuite[GlutenColumnExpressionSuite]
     // Velox raise_error('errMsg') throws a velox_user_error exception with the message 'errMsg'.
     // The final caught Spark exception's getCause().getMessage() contains 'errMsg' but does not

gluten-ut/spark33/src/test/scala/org/apache/spark/sql/GlutenCharVarcharTestSuite.scala

@@ -16,8 +16,48 @@
  */
 package org.apache.spark.sql
 
+import org.apache.spark.SparkException
+
 class GlutenFileSourceCharVarcharTestSuite
   extends FileSourceCharVarcharTestSuite
-  with GlutenSQLTestsTrait {}
+  with GlutenSQLTestsTrait {
+
+  private val VELOX_ERROR_MESSAGE = "Exceeds allowed length limitation: 5"
+
+  private def testTableWrite(f: String => Unit): Unit = {
+    withTable("t")(f("char"))
+    withTable("t")(f("varchar"))
+  }
+
+  testGluten("length check for input string values: nested in struct of array") {
+    testTableWrite {
+      typeName =>
+        sql(s"CREATE TABLE t(c STRUCT<c: ARRAY<$typeName(5)>>) USING $format")
+        sql("INSERT INTO t SELECT struct(array(null))")
+        checkAnswer(spark.table("t"), Row(Row(Seq(null))))
+        val e = intercept[SparkException](sql("INSERT INTO t SELECT struct(array('123456'))"))
+        assert(e.getCause.getMessage.contains(VELOX_ERROR_MESSAGE))
+    }
+  }
+}
+
+class GlutenDSV2CharVarcharTestSuite extends DSV2CharVarcharTestSuite with GlutenSQLTestsTrait {
+
+  private val VELOX_ERROR_MESSAGE = "Exceeds allowed length limitation: 5"
+
+  private def testTableWrite(f: String => Unit): Unit = {
+    withTable("t")(f("char"))
+    withTable("t")(f("varchar"))
+  }
 
-class GlutenDSV2CharVarcharTestSuite extends DSV2CharVarcharTestSuite with GlutenSQLTestsTrait {}
+  testGluten("length check for input string values: nested in struct of array") {
+    testTableWrite {
+      typeName =>
+        sql(s"CREATE TABLE t(c STRUCT<c: ARRAY<$typeName(5)>>) USING $format")
+        sql("INSERT INTO t SELECT struct(array(null))")
+        checkAnswer(spark.table("t"), Row(Row(Seq(null))))
+        val e = intercept[SparkException](sql("INSERT INTO t SELECT struct(array('123456'))"))
+        assert(e.getCause.getMessage.contains(VELOX_ERROR_MESSAGE))
+    }
+  }
+}