Enable VARCHAR ↔ timestamp_ntz cast native execution in Velox backend

Author: Mariam-Almesfer

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

@@ -599,12 +599,14 @@ class DateFunctionsValidateSuite extends FunctionsValidateSuite {
           checkGlutenPlan[BatchScanExecTransformer]
         }
 
-        // Ensures the fallback of unsupported function works.
-        runQueryAndCompare("select hour(ts) from view") {
-          df =>
-            assert(collect(df.queryExecution.executedPlan) {
-              case p if p.isInstanceOf[ProjectExec] => p
-            }.nonEmpty)
+        // cast(string as timestamp_ntz) runs natively.
+        // Must use a file-backed scan so Gluten has a native leaf to offload the Project above.
+        val strPath = dir.getAbsolutePath + "/str_data"
+        spark.createDataset(inputs).toDF("str")
+          .coalesce(1).write.mode("overwrite").parquet(strPath)
+        spark.read.parquet(strPath).createOrReplaceTempView("str_view")
+        runQueryAndCompare("select cast(str as timestamp_ntz) from str_view") {
+          checkGlutenPlan[ProjectExecTransformer]
         }
     }
   }

ep/build-velox/src/get-velox.sh

@@ -25,7 +25,7 @@ RUN_SETUP_SCRIPT=ON
 ENABLE_ENHANCED_FEATURES=OFF
 
 # Developer use only for testing Velox PR.
-UPSTREAM_VELOX_PR_ID=""
+UPSTREAM_VELOX_PR_ID="17498"
 
 OS=`uname -s`
 

gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/validator/Validators.scala

@@ -25,6 +25,7 @@ import org.apache.gluten.extension.columnar.offload.OffloadSingleNode
 import org.apache.gluten.sql.shims.SparkShimLoader
 
 import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.expressions.Cast
 import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.aggregate.{HashAggregateExec, ObjectHashAggregateExec, SortAggregateExec}
 import org.apache.spark.sql.execution.datasources.WriteFilesExec
@@ -33,7 +34,7 @@ import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ShuffleEx
 import org.apache.spark.sql.execution.joins._
 import org.apache.spark.sql.execution.window.WindowExec
 import org.apache.spark.sql.hive.HiveTableScanExecTransformer
-import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StringType, StructType}
 
 object Validators {
   implicit class ValidatorBuilderImplicits(builder: Validator.Builder) {
@@ -262,9 +263,23 @@ object Validators {
           case p if HiveTableScanExecTransformer.isHiveTableScan(p) => true
           case _ => false
         }
-        val hasNTZ = plan.output.exists(a => containsNTZ(a.dataType)) ||
-          plan.children.exists(_.output.exists(a => containsNTZ(a.dataType)))
-        if (isScan || !hasNTZ) {
+        // Allow: scans, NTZ-consuming ops (e.g. hour(timestamp_ntz)->int),
+        // and explicit cast(varchar as timestamp_ntz). Fall back for everything else
+        // that produces NTZ output (literals, pass-through, unknown conversions).
+        val inputHasNTZ = plan.children.exists(_.output.exists(a => containsNTZ(a.dataType)))
+        val outputHasNTZ = plan.output.exists(a => containsNTZ(a.dataType))
+        val isVarcharToNtzCast = !inputHasNTZ && outputHasNTZ && (plan match {
+          case p: ProjectExec =>
+            p.projectList.forall {
+              expr =>
+                !containsNTZ(expr.dataType) || (expr match {
+                  case c: Cast => c.child.dataType == StringType
+                  case _ => false
+                })
+            }
+          case _ => false
+        })
+        if (isScan || !outputHasNTZ || isVarcharToNtzCast) {
           return pass()
         }
       }