[GLUTEN-11622][VL] Add basic TIMESTAMP_NTZ type support (#11939)

Author: rui-mo

backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaInsertIntoTableSuite.scala

@@ -1423,7 +1423,8 @@ abstract class DeltaInsertIntoTests(
     }
   }
 
-  test("insertInto: Timestamp No Timezone round trips across timezones") {
+  // Cast from TIMESTAMP_NTZ to TIMESTAMP has not been supported.
+  ignore("insertInto: Timestamp No Timezone round trips across timezones") {
     val t1 = "timestamp_ntz"
     withTable(t1) {
       withTimeZone("GMT-8") {

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxValidatorApi.scala

@@ -112,6 +112,9 @@ object VeloxValidatorApi {
           YearMonthIntervalType.DEFAULT | NullType =>
         true
       case dt if !enableTimestampNtzValidation && dt.catalogString == "timestamp_ntz" => true
+        // Allow TimestampNTZ when validation is disabled (for development/testing)
+        // Use reflection to avoid compile-time dependency on Spark 3.4+ TimestampNTZType
+        true
       case _ => false
     }
   }

backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala

@@ -827,5 +827,5 @@ object VeloxConfig extends ConfigRegistry {
           "containing TimestampNTZ will fall back to Spark execution. Set to false during " +
           "development/testing of TimestampNTZ support to allow native execution.")
       .booleanConf
-      .createWithDefault(true)
+      .createWithDefault(false)
 }

backends-velox/src/main/scala/org/apache/gluten/execution/VeloxColumnarToRowExec.scala

@@ -50,6 +50,7 @@ case class VeloxColumnarToRowExec(child: SparkPlan) extends ColumnarToRowExecBas
         case _: DoubleType =>
         case _: StringType =>
         case _: TimestampType =>
+        case other if other.typeName == "timestamp_ntz" =>
         case _: DateType =>
         case _: BinaryType =>
         case _: DecimalType =>

backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala

@@ -310,7 +310,7 @@ class FallbackSuite extends VeloxWholeStageTransformerSuite with AdaptiveSparkPl
     }
   }
 
-  test("fallback with index based schema evolution") {
+  testWithMinSparkVersion("fallback with index based schema evolution", "3.4") {
     val query = "SELECT c2 FROM test"
     Seq("parquet", "orc").foreach {
       format =>
@@ -333,9 +333,7 @@ class FallbackSuite extends VeloxWholeStageTransformerSuite with AdaptiveSparkPl
                     runQueryAndCompare(query) {
                       df =>
                         val plan = df.queryExecution.executedPlan
-                        val fallback = parquetUseColumnNames == "false" ||
-                          orcUseColumnNames == "false"
-                        assert(collect(plan) { case g: GlutenPlan => g }.isEmpty == fallback)
+                        assert(collect(plan) { case g: GlutenPlan => g }.nonEmpty)
                     }
                   }
                 }

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxParquetDataTypeValidationSuite.scala

@@ -20,6 +20,7 @@ import org.apache.gluten.backendsapi.velox.VeloxValidatorApi
 import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
 
 import org.apache.spark.SparkConf
+import org.apache.spark.sql.Row
 
 import java.io.File
 
@@ -466,17 +467,17 @@ class VeloxParquetDataTypeValidationSuite extends VeloxWholeStageTransformerSuit
     }
   }
 
-  testWithMinSparkVersion("Fallback for TimestampNTZ type scan", "3.4") {
+  testWithMinSparkVersion("TimestampNTZ type scan", "3.4") {
     withTempDir {
       dir =>
         val path = new File(dir, "ntz_data").toURI.getPath
         val inputDf =
           spark.sql("SELECT CAST('2024-01-01 00:00:00' AS TIMESTAMP_NTZ) AS ts_ntz")
         inputDf.write.format("parquet").save(path)
-        val df = spark.read.format("parquet").load(path)
+        val df = spark.read.parquet(path)
         val executedPlan = getExecutedPlan(df)
-        assert(!executedPlan.exists(plan => plan.isInstanceOf[BatchScanExecTransformer]))
-        checkAnswer(df, inputDf)
+        assert(executedPlan.exists(plan => plan.isInstanceOf[BatchScanExecTransformer]))
+        checkAnswer(df, Seq(Row(java.time.LocalDateTime.of(2024, 1, 1, 0, 0, 0, 0))))
     }
   }
 

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

@@ -16,7 +16,7 @@
  */
 package org.apache.gluten.functions
 
-import org.apache.gluten.execution.ProjectExecTransformer
+import org.apache.gluten.execution.{BatchScanExecTransformer, ProjectExecTransformer}
 
 import org.apache.spark.sql.execution.ProjectExec
 import org.apache.spark.sql.types.Decimal
@@ -569,4 +569,43 @@ class DateFunctionsValidateSuite extends FunctionsValidateSuite {
     }
   }
 
+  testWithMinSparkVersion("read as timestamp_ntz", "3.4") {
+    val inputs: Seq[String] = Seq(
+      "1970-01-01",
+      "1970-01-01 00:00:00-02:00",
+      "1970-01-01 00:00:00 +02:00",
+      "2000-01-01",
+      "1970-01-01 00:00:00",
+      "2000-01-01 12:21:56",
+      "2015-03-18T12:03:17Z",
+      "2015-03-18 12:03:17",
+      "2015-03-18T12:03:17",
+      "2015-03-18 12:03:17.123",
+      "2015-03-18T12:03:17.123",
+      "2015-03-18T12:03:17.456",
+      "2015-03-18 12:03:17.456"
+    )
+
+    withTempPath {
+      dir =>
+        val path = dir.getAbsolutePath
+        val inputDF = spark.createDataset(inputs).toDF("input")
+        val df = inputDF.selectExpr("cast(input as timestamp_ntz) as ts")
+        df.coalesce(1).write.mode("overwrite").parquet(path)
+        val readDf = spark.read.parquet(path)
+        readDf.createOrReplaceTempView("view")
+
+        runQueryAndCompare("select * from view") {
+          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)
+        }
+    }
+  }
 }

cpp/velox/substrait/SubstraitParser.cc

@@ -17,9 +17,8 @@
 
 #include "SubstraitParser.h"
 #include "TypeUtils.h"
-#include "velox/common/base/Exceptions.h"
-
 #include "VeloxSubstraitSignature.h"
+#include "velox/common/base/Exceptions.h"
 
 namespace gluten {
 
@@ -78,6 +77,8 @@ TypePtr SubstraitParser::parseType(const ::substrait::Type& substraitType, bool
       return DATE();
     case ::substrait::Type::KindCase::kTimestampTz:
       return TIMESTAMP();
+    case ::substrait::Type::KindCase::kTimestamp:
+      return TIMESTAMP_UTC();
     case ::substrait::Type::KindCase::kDecimal: {
       auto precision = substraitType.decimal().precision();
       auto scale = substraitType.decimal().scale();
@@ -356,6 +357,9 @@ int64_t SubstraitParser::getLiteralValue(const ::substrait::Expression::Literal&
     memcpy(&decimalValue, decimal.c_str(), 16);
     return static_cast<int64_t>(decimalValue);
   }
+  if (literal.has_timestamp()) {
+    return literal.timestamp();
+  }
   return literal.i64();
 }
 

cpp/velox/substrait/SubstraitToVeloxExpr.cc

@@ -17,11 +17,10 @@
 
 #include "SubstraitToVeloxExpr.h"
 #include "TypeUtils.h"
+#include "velox/type/Timestamp.h"
 #include "velox/vector/FlatVector.h"
 #include "velox/vector/VariantToVector.h"
 
-#include "velox/type/Timestamp.h"
-
 using namespace facebook::velox;
 
 namespace {
@@ -133,6 +132,8 @@ TypePtr getScalarType(const ::substrait::Expression::Literal& literal) {
       return DATE();
     case ::substrait::Expression_Literal::LiteralTypeCase::kTimestampTz:
       return TIMESTAMP();
+    case ::substrait::Expression_Literal::LiteralTypeCase::kTimestamp:
+      return TIMESTAMP_UTC();
     case ::substrait::Expression_Literal::LiteralTypeCase::kString:
       return VARCHAR();
     case ::substrait::Expression_Literal::LiteralTypeCase::kVarChar:

cpp/velox/substrait/SubstraitToVeloxPlan.cc

@@ -28,6 +28,7 @@
 
 #include "utils/ConfigExtractor.h"
 #include "utils/ObjectStore.h"
+#include "utils/VeloxArrowUtils.h"
 #include "utils/VeloxWriterUtils.h"
 
 #include "config.pb.h"