Spark version specific behaviour for cast string to. timestamp

Author: parthchandra

native/core/src/execution/planner.rs

@@ -406,7 +406,12 @@ impl PhysicalPlanner {
                 Ok(Arc::new(Cast::new(
                     child,
                     datatype,
-                    SparkCastOptions::new(eval_mode, &expr.timezone, expr.allow_incompat),
+                    {
+                        let mut opts =
+                            SparkCastOptions::new(eval_mode, &expr.timezone, expr.allow_incompat);
+                        opts.is_spark4_plus = expr.is_spark4_plus;
+                        opts
+                    },
                     spark_expr.expr_id,
                     query_context,
                 )))

native/proto/src/proto/expr.proto

@@ -266,6 +266,9 @@ message Cast {
   string timezone = 3;
   EvalMode eval_mode = 4;
   bool allow_incompat = 5;
+  // True when running against Spark 4.0+. Controls version-specific cast behaviour
+  // such as the handling of leading whitespace before T-prefixed time-only strings.
+  bool is_spark4_plus = 6;
 }
 
 message BinaryExpr {

native/spark-expr/src/conversion_funcs/cast.rs

@@ -131,6 +131,9 @@ pub struct SparkCastOptions {
     pub timezone: String,
     /// Allow casts that are supported but not guaranteed to be 100% compatible
     pub allow_incompat: bool,
+    /// True when running against Spark 4.0+. Enables version-specific cast behaviour
+    /// such as the handling of leading whitespace before T-prefixed time-only strings.
+    pub is_spark4_plus: bool,
     /// Support casting unsigned ints to signed ints (used by Parquet SchemaAdapter)
     pub allow_cast_unsigned_ints: bool,
     /// We also use the cast logic for adapting Parquet schemas, so this flag is used
@@ -148,6 +151,7 @@ impl SparkCastOptions {
             eval_mode,
             timezone: timezone.to_string(),
             allow_incompat,
+            is_spark4_plus: false,
             allow_cast_unsigned_ints: false,
             is_adapting_schema: false,
             null_string: "null".to_string(),
@@ -160,6 +164,7 @@ impl SparkCastOptions {
             eval_mode,
             timezone: "".to_string(),
             allow_incompat,
+            is_spark4_plus: false,
             allow_cast_unsigned_ints: false,
             is_adapting_schema: false,
             null_string: "null".to_string(),
@@ -297,7 +302,13 @@ pub(crate) fn cast_array(
         (Utf8, Boolean) => spark_cast_utf8_to_boolean::<i32>(&array, eval_mode),
         (LargeUtf8, Boolean) => spark_cast_utf8_to_boolean::<i64>(&array, eval_mode),
         (Utf8, Timestamp(_, _)) => {
-            cast_string_to_timestamp(&array, to_type, eval_mode, &cast_options.timezone)
+            cast_string_to_timestamp(
+                &array,
+                to_type,
+                eval_mode,
+                &cast_options.timezone,
+                cast_options.is_spark4_plus,
+            )
         }
         (Utf8, Date32) => cast_string_to_date(&array, to_type, eval_mode),
         (Date32, Int32) => {