@@ -31,6 +31,7 @@ license: |
 - Since Spark 4.4, when a file source reads a partition column with a user-specified `DECIMAL` type from directory names, it converts each value to the precision and scale of that type, rounding half up as `CAST` does. Previously, the vectorized Parquet and ORC readers applied the declared scale to the digits in the directory name without an error, for example `d=1.5` read as `DECIMAL(10,2)` returned `0.15`. Data derived from affected query results may need to be recomputed. A value that does not fit the declared precision now fails with `INVALID_PARTITION_VALUE` regardless of `spark.sql.ansi.enabled`. Previously, those readers returned a wrong value and the other readers returned `NULL`. To get `NULL` for such a value, set `spark.sql.sources.validatePartitionColumns` to `false`. Partitions tracked in the catalog are not affected.
 - Since Spark 4.4, when a file source reads a partition column with a user-specified `TINYINT` or `SMALLINT` type from directory names, a value out of the range of that type fails with `INVALID_PARTITION_VALUE` regardless of `spark.sql.ansi.enabled`, as it does for `INT`. Previously, the value wrapped around without an error, for example `b=300` read as `TINYINT` returned `44`. Data derived from affected query results may need to be recomputed. To get `NULL` for such a value, set `spark.sql.sources.validatePartitionColumns` to `false`. Reading partitions tracked in the catalog is not affected, but writing to a table whose catalog tracks such a partition, for example one added by `MSCK REPAIR TABLE`, can now fail with `NumberFormatException` unless `spark.sql.sources.validatePartitionColumns` is `false`.
 - Since Spark 4.4, Data Source V2 file scans correctly evaluate deterministic Python UDF filter predicates that reference only partition columns. Previously, these predicates could be dropped, returning rows that did not satisfy the filter (SPARK-60486). Data derived from affected query results may need to be recomputed.
+- Since Spark 4.4, when a file source reads a partition column with a user-specified datetime type such as `DATE` or `TIMESTAMP` from directory names, a value that cannot be cast to that type fails with `INVALID_PARTITION_VALUE` regardless of `spark.sql.ansi.enabled`. Previously, such a value became `NULL` when `spark.sql.ansi.enabled` was `false`. When it was `true`, a `TIMESTAMP_NTZ` value with a valid date and an invalid time, such as `2020-01-01 25:00:00`, became the start of that day. To get `NULL` for such a value, set `spark.sql.sources.validatePartitionColumns` to `false`. Partitions tracked in the catalog are not affected.
 - Since Spark 4.4, quoted Variant path keys decode `\\`, `\/`, `\b`, `\f`, `\n`, `\r`, `\t`, `\uXXXX` (including valid surrogate pairs), and escapes of the enclosing quote (`\'` or `\"`). Previously, backslashes were literal, so existing paths can match different keys or become invalid. For example, use `$['a\\']` instead of `$['a\']`. Unknown escapes and unpaired surrogate escapes stay literal. Dot-notation keys are unchanged. To keep the old behavior, set `spark.sql.legacy.variantPathBackslashAsLiteral` to `true` before creating the query.
 
 - Since Spark 4.4, an auto-filled generated column of a Data Source V2 table reads a field of another column, such as `s.f`, from the value stored in that column. Previously, it read the value of the query before the store assignment cast, so it could silently store a value that does not match the stored column, and a later write of that row failed with `CHECK_CONSTRAINT_VIOLATION` (SPARK-60803). Rows that are already stored keep their values, and data derived from them may need to be recomputed.
 
@@ -588,21 +588,18 @@ object PartitioningUtils extends SQLConfHelper {
     case LongType => JLong.parseLong(value)
     case FloatType => JDouble.parseDouble(value).toFloat
     case DoubleType => JDouble.parseDouble(value)
-    // Cast in ANSI mode to fail on a value that the type cannot hold in any session, so that
-    // `spark.sql.sources.validatePartitionColumns` alone decides between an error and null.
+    // Cast in ANSI mode, here and in the datetime and interval cases below, so that a value that
+    // the type cannot hold fails in any session and `spark.sql.sources.validatePartitionColumns`
+    // alone decides between an error and null.
     case dt: DecimalType => Cast(Literal(value), dt, ansiEnabled = true).eval()
     case DateType =>
-      Cast(Literal(value), DateType, Some(zoneId.getId)).eval()
-    case tt: TimeType => Cast(Literal(unescapePathName(value)), tt).eval()
+      Cast(Literal(value), DateType, Some(zoneId.getId), ansiEnabled = true).eval()
+    case tt: TimeType => Cast(Literal(unescapePathName(value)), tt, ansiEnabled = true).eval()
     // Timestamp types
     case dt if AnyTimestampType.acceptsType(dt) || AnyTimestampNanoType.acceptsType(dt) =>
-      Try {
-        Cast(Literal(unescapePathName(value)), dt, Some(zoneId.getId)).eval()
-      }.getOrElse {
-        Cast(Cast(Literal(value), DateType, Some(zoneId.getId)), dt).eval()
-      }
+      Cast(Literal(unescapePathName(value)), dt, Some(zoneId.getId), ansiEnabled = true).eval()
     case it: AnsiIntervalType =>
-      Cast(Literal(unescapePathName(value)), it).eval()
+      Cast(Literal(unescapePathName(value)), it, ansiEnabled = true).eval()
     case BinaryType => value.getBytes()
     case BooleanType => value.toBoolean
     case dt => throw SparkException.internalError(s"Unsupported partition type: $dt")
 
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.datasources
 
 import java.io.{File, FileNotFoundException}
 import java.net.URI
+import java.time.LocalDateTime
 
 import scala.collection.mutable
 import scala.concurrent.duration._
@@ -37,7 +38,7 @@ import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
 import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{ByteType, DecimalType, IntegerType, ShortType, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{ByteType, DateType, DecimalType, IntegerType, ShortType, StringType, StructField, StructType, TimestampNTZType, TimestampType, TimeType}
 import org.apache.spark.util.{HadoopFSUtils, KnownSizeEstimation}
 
 class FileIndexSuite extends SharedSparkSession {
@@ -215,6 +216,63 @@ class FileIndexSuite extends SharedSparkSession {
     }
   }
 
+  test("SPARK-60825: validate datetime partition values regardless of ANSI mode") {
+    // A TIMESTAMP_NTZ value with a valid date and an invalid time must not become the start of
+    // that day.
+    val cases = Seq(
+      DateType -> "foo",
+      TimeType(6) -> "foo",
+      TimestampType -> "foo",
+      TimestampNTZType -> "2020-01-01 25%3A00%3A00")
+    for ((dataType, value) <- cases; ansiEnabled <- Seq("true", "false")) {
+      withTempDir { dir =>
+        val partitionDirectory = new File(dir, s"a=$value")
+        partitionDirectory.mkdir()
+        stringToFile(new File(partitionDirectory, "text.txt"), "text")
+        val path = new Path(dir.getCanonicalPath)
+        val schema = StructType(Seq(StructField("a", dataType, false)))
+        withSQLConf(
+          SQLConf.ANSI_ENABLED.key -> ansiEnabled,
+          SQLConf.VALIDATE_PARTITION_COLUMNS.key -> "true") {
+          val fileIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty, Some(schema))
+          checkError(
+            exception = intercept[SparkRuntimeException] {
+              fileIndex.partitionSpec()
+            },
+            condition = "INVALID_PARTITION_VALUE",
+            parameters = Map(
+              "value" -> s"'$value'",
+              "dataType" -> s"\"${dataType.sql}\"",
+              "columnName" -> "`a`"))
+        }
+        withSQLConf(
+          SQLConf.ANSI_ENABLED.key -> ansiEnabled,
+          SQLConf.VALIDATE_PARTITION_COLUMNS.key -> "false") {
+          val fileIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty, Some(schema))
+          val partitionValues = fileIndex.partitionSpec().partitions.map(_.values)
+          assert(partitionValues.length == 1 && partitionValues(0).isNullAt(0))
+        }
+      }
+    }
+
+    // A date without a time is a valid timestamp.
+    withTempDir { dir =>
+      val partitionDirectory = new File(dir, "a=2020-01-01")
+      partitionDirectory.mkdir()
+      stringToFile(new File(partitionDirectory, "text.txt"), "text")
+      val path = new Path(dir.getCanonicalPath)
+      val schema = StructType(Seq(StructField("a", TimestampNTZType, false)))
+      val midnight = DateTimeUtils.localDateTimeToMicros(LocalDateTime.of(2020, 1, 1, 0, 0))
+      for (ansiEnabled <- Seq("true", "false")) {
+        withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled) {
+          val fileIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty, Some(schema))
+          val partitionValues = fileIndex.partitionSpec().partitions.map(_.values)
+          assert(partitionValues.length == 1 && partitionValues(0).getLong(0) == midnight)
+        }
+      }
+    }
+  }
+
   test("InMemoryFileIndex: input paths are converted to qualified paths") {
     withTempDir { dir =>
       val file = new File(dir, "text.txt")