Add validation test for localtimestamp

Author: Mariam-Almesfer

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

@@ -104,11 +104,18 @@ class VeloxValidatorApi extends ValidatorApi {
 
 object VeloxValidatorApi {
   private def isPrimitiveType(dataType: DataType): Boolean = {
+    val enableTimestampNtzValidation = VeloxConfig.get.enableTimestampNtzValidation
     dataType match {
       case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType |
           StringType | BinaryType | _: DecimalType | DateType | TimestampType |
           YearMonthIntervalType.DEFAULT | NullType =>
         true
+      case dt
+          if !enableTimestampNtzValidation &&
+            dt.getClass.getSimpleName == "TimestampNTZType" =>
+        // 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

@@ -99,6 +99,8 @@ class VeloxConfig(conf: SQLConf) extends GlutenConfig(conf) {
 
   def valueStreamDynamicFilterEnabled: Boolean =
     getConf(VALUE_STREAM_DYNAMIC_FILTER_ENABLED)
+
+  def enableTimestampNtzValidation: Boolean = getConf(ENABLE_TIMESTAMP_NTZ_VALIDATION)
 }
 
 object VeloxConfig extends ConfigRegistry {
@@ -751,4 +753,13 @@ object VeloxConfig extends ConfigRegistry {
       .doc("Maps table field names to file field names using names, not indices for Parquet files.")
       .booleanConf
       .createWithDefault(true)
+
+  val ENABLE_TIMESTAMP_NTZ_VALIDATION =
+    buildConf("spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation")
+      .doc(
+        "Enable validation fallback for TimestampNTZ type. When true (default), any plan " +
+          "containing TimestampNTZ will fall back to Spark execution. Set to false during " +
+          "development/testing of TimestampNTZ support to allow native execution.")
+      .booleanConf
+      .createWithDefault(true)
 }

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

@@ -1507,4 +1507,15 @@ abstract class ScalarFunctionsValidateSuite extends FunctionsValidateSuite {
         }
     }
   }
+  test("test_current_timestamp") {
+    val df = spark.sql("SELECT l_orderkey, current_timestamp() from lineitem limit 1")
+    val optimizedPlan = df.queryExecution.optimizedPlan.toString()
+    assert(
+      !optimizedPlan.contains("CurrentTimestamp"),
+      s"Expected CurrentTimestamp to be folded to a literal, but got: $optimizedPlan"
+    )
+    checkGlutenPlan[ProjectExecTransformer](df)
+    checkFallbackOperators(df, 0)
+    df.collect()
+  }
 }

docs/velox-configuration.md

@@ -27,6 +27,7 @@ nav_order: 16
 | spark.gluten.sql.columnar.backend.velox.cudf.memoryResource                      | async             | GPU RMM memory resource.                                                                                                                                                                                                                                                                                                                                                                                                                              |
 | spark.gluten.sql.columnar.backend.velox.cudf.shuffleMaxPrefetchBytes             | 1028MB            | Maximum bytes to prefetch in CPU memory during GPU shuffle read while waitingfor GPU available.                                                                                                                                                                                                                                                                                                                                                       |
 | spark.gluten.sql.columnar.backend.velox.directorySizeGuess                       | 32KB              | Deprecated, rename to spark.gluten.sql.columnar.backend.velox.footerEstimatedSize                                                                                                                                                                                                                                                                                                                                                                     |
+| spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation             | true              | Enable validation fallback for TimestampNTZ type. When true (default), any plan containing TimestampNTZ will fall back to Spark execution. Set to false during development/testing of TimestampNTZ support to allow native execution.                                                                                                                                                                                                                 |
 | spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled                   | false             | Disables caching if false. File handle cache should be disabled if files are mutable, i.e. file content may change while file path stays the same.                                                                                                                                                                                                                                                                                                    |
 | spark.gluten.sql.columnar.backend.velox.filePreloadThreshold                     | 1MB               | Set the file preload threshold for velox file scan, refer to Velox's file-preload-threshold                                                                                                                                                                                                                                                                                                                                                           |
 | spark.gluten.sql.columnar.backend.velox.floatingPointMode                        | loose             | Config used to control the tolerance of floating point operations alignment with Spark. When the mode is set to strict, flushing is disabled for sum(float/double)and avg(float/double). When set to loose, flushing will be enabled.                                                                                                                                                                                                                 |

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

@@ -40,6 +40,18 @@ object Validators {
     private val conf = GlutenConfig.get
     private val settings = BackendsApiManager.getSettings
 
+    // Get VeloxConfig
+    private val veloxConf: Option[Any] = {
+      try {
+        val veloxConfigClass =
+          org.apache.spark.util.Utils.classForName("org.apache.gluten.config.VeloxConfig")
+        val getMethod = veloxConfigClass.getMethod("get")
+        Some(getMethod.invoke(null))
+      } catch {
+        case _: Exception => None
+      }
+    }
+
     /** Fails validation if a plan node was already tagged with TRANSFORM_UNSUPPORTED. */
     def fallbackByHint(): Validator.Builder = {
       builder.add(FallbackByHint)
@@ -81,7 +93,7 @@ object Validators {
 
     /** Fails validation if a plan node's input or output schema contains TimestampNTZType. */
     def fallbackByTimestampNTZ(): Validator.Builder = {
-      builder.add(new FallbackByTimestampNTZ())
+      builder.add(new FallbackByTimestampNTZ(veloxConf))
     }
 
     /**
@@ -218,8 +230,25 @@ object Validators {
     }
   }
 
-  private class FallbackByTimestampNTZ() extends Validator {
+  private class FallbackByTimestampNTZ(veloxConf: Option[Any]) extends Validator {
+    // Check if TimestampNTZ validation is enabled via VeloxConfig
+    private val enableValidation: Boolean = veloxConf match {
+      case Some(config) =>
+        try {
+          val enableMethod = config.getClass.getMethod("enableTimestampNtzValidation")
+          enableMethod.invoke(config).asInstanceOf[Boolean]
+        } catch {
+          case _: Exception => true
+        }
+      case None => true
+    }
+
     override def validate(plan: SparkPlan): Validator.OutCome = {
+      if (!enableValidation) {
+        // Validation is disabled, allow TimestampNTZ
+        return pass()
+      }
+
       def containsNTZ(dataType: DataType): Boolean = dataType match {
         case dt if dt.catalogString == "timestamp_ntz" => true
         case st: StructType => st.exists(f => containsNTZ(f.dataType))