diff --git a/backends-velox/src/test/scala/org/apache/gluten/functions/DateFunctionsValidateSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/functions/DateFunctionsValidateSuite.scala
index e5e25c88b6f3..097d2873cb87 100644
--- a/backends-velox/src/test/scala/org/apache/gluten/functions/DateFunctionsValidateSuite.scala
+++ b/backends-velox/src/test/scala/org/apache/gluten/functions/DateFunctionsValidateSuite.scala
@@ -599,12 +599,15 @@ 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(timestamp_ntz as string) runs natively.
+        runQueryAndCompare("select cast(ts as string) from view") {
+          checkGlutenPlan[ProjectExecTransformer]
+        }
+
+        // cast(string as timestamp_ntz) runs natively.
+        spark.createDataset(inputs).toDF("str").createOrReplaceTempView("str_view")
+        runQueryAndCompare("select cast(str as timestamp_ntz) from str_view") {
+          checkGlutenPlan[ProjectExecTransformer]
         }
     }
   }
diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/validator/Validators.scala b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/validator/Validators.scala
index 3218e045f543..7706cd214663 100644
--- a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/validator/Validators.scala
+++ b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/validator/Validators.scala
@@ -262,9 +262,12 @@ 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 nodes that either consume NTZ (e.g. hour(timestamp_ntz) -> int)
+        // or produce NTZ from non-NTZ input (e.g. cast(string as timestamp_ntz)).
+        // Only fall back when NTZ propagates unchanged through both input and output.
+        val inputHasNTZ = plan.children.exists(_.output.exists(a => containsNTZ(a.dataType)))
+        val outputHasNTZ = plan.output.exists(a => containsNTZ(a.dataType))
+        if (isScan || !(inputHasNTZ && outputHasNTZ)) {
           return pass()
         }
       }