Fix TIMESTAMP_NTZ cast not stripping timezone

crates/executor/src/tests/sql/functions/timestamp/snapshots/to_timestamp/query_timestamp_with_timezone_to_timestamp.snap

@@ -4,10 +4,10 @@ description: "\"SELECT TO_TIMESTAMP(CONVERT_TIMEZONE('UTC', '2024-12-31 10:00:00
 ---
 Ok(
     [
-        "+----------------------+",
-        "| model_tstamp         |",
-        "+----------------------+",
-        "| 2024-12-31T10:00:00Z |",
-        "+----------------------+",
+        "+---------------------+",
+        "| model_tstamp        |",
+        "+---------------------+",
+        "| 2024-12-31T10:00:00 |",
+        "+---------------------+",
     ],
 )

crates/executor/src/utils.rs

@@ -950,7 +950,7 @@ mod tests {
             (
                 TimeUnit::Nanosecond,
                 Some(1_627_846_261_233_222_111),
-                "1627846261.233222111 1440", // 1020-1440 = -420 (PDT)
+                "1627846261.233222111 1440",
                 Some("UTC".to_string()),
                 (1_627_846_261, 233_222_111, 1_627_846_261_233_222_111),
             ),

crates/functions/src/conversion/to_timestamp.rs

@@ -204,15 +204,19 @@ impl ScalarUDFImpl for ToTimestampFunc {
             }
         }
 
-        // If the first argument is already a timestamp type, return it as is (with its timezone).
+        // If the first argument is already a timestamp type, preserve the unit but
+        // apply this function's timezone policy (e.g. to_timestamp_ntz strips timezone).
         if matches!(
             args.arg_fields[0].data_type(),
             DataType::Timestamp(TimeUnit::Microsecond, _)
                 | DataType::Timestamp(TimeUnit::Nanosecond, Some(_))
         ) {
+            let DataType::Timestamp(unit, _) = args.arg_fields[0].data_type() else {
+                unreachable!()
+            };
             return Ok(Arc::new(Field::new(
                 self.name(),
-                args.arg_fields[0].data_type().clone(),
+                DataType::Timestamp(unit.clone(), self.timezone()),
                 true,
             )));
         }
@@ -317,6 +321,20 @@ impl ScalarUDFImpl for ToTimestampFunc {
                 DataType::Timestamp(TimeUnit::Microsecond, _)
                     | DataType::Timestamp(TimeUnit::Nanosecond, Some(_))
             ) {
+                // If the input timezone differs from this function's target timezone
+                // (e.g. to_timestamp_ntz needs to strip tz), cast to apply the policy.
+                let DataType::Timestamp(unit, ref tz) = arr.data_type().clone() else {
+                    unreachable!()
+                };
+                let target_tz = self.timezone();
+                if *tz != target_tz {
+                    let arr = cast_with_options(
+                        &arr,
+                        &DataType::Timestamp(unit, target_tz),
+                        &DEFAULT_CAST_OPTIONS,
+                    )?;
+                    return Ok(ColumnarValue::Array(arr));
+                }
                 return Ok(ColumnarValue::Array(Arc::new(arr)));
             }