You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/21 22:16:31 UTC
(spark) branch branch-4.x updated: [SPARK-60424][SQL] Fix Hive `timestamp` type parsing with `spark.sql.timestampType=TIMESTAMP_NTZ`
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 5712b93d9151 [SPARK-60424][SQL] Fix Hive `timestamp` type parsing with `spark.sql.timestampType=TIMESTAMP_NTZ`
5712b93d9151 is described below
commit 5712b93d915160c15ec7e757da014969de7e5120
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Wed Oct 21 15:16:09 2026 -0700
[SPARK-60424][SQL] Fix Hive `timestamp` type parsing with `spark.sql.timestampType=TIMESTAMP_NTZ`
### What changes were proposed in this pull request?
This PR aims to map the Hive `timestamp` type to `TIMESTAMP_LTZ` regardless of `spark.sql.timestampType`. It affects Hive tables of all file formats, e.g. `PARQUET`, `ORC` and `TEXTFILE`.
`sql/hive` parses Hive type strings with `CatalystSqlParser`, which resolves `timestamp` by `spark.sql.timestampType`, in two places:
- `HiveClientImpl.getSparkSQLDataType` for the column types in Hive metastore
- `OrcFileOperator.readSchema` and `OrcFileOperator.readOrcSchemasInParallel` for the ORC file schema with `spark.sql.orc.impl=hive`
However, Spark reads and writes Hive `timestamp` as `TIMESTAMP_LTZ`, and Hive has no `timestamp_ntz` type. This PR adds `HiveClientImpl.parseHiveType`, which maps `TIMESTAMP_NTZ` back to `TIMESTAMP_LTZ` after parsing, and uses it in both places.
### Why are the changes needed?
With `spark.sql.timestampType=TIMESTAMP_NTZ`, Spark writes wrong data into Hive tables, returns wrong query results, or fails.
The `timestamp` columns in Hive metastore become `TIMESTAMP_NTZ`. The Hive tables created by Spark are also affected because Spark falls back to the Hive metastore schema when it differs from the schema in the table properties. For example, `INSERT` writes `TIMESTAMP_NTZ` values into a Hive `timestamp` column.
```scala
$ bin/spark-shell -c spark.sql.session.timeZone=America/Los_Angeles
scala> sql("CREATE TABLE t (ts TIMESTAMP) STORED AS PARQUET")
scala> spark.conf.set("spark.sql.timestampType", "TIMESTAMP_NTZ")
scala> spark.table("t").schema
val res2: org.apache.spark.sql.types.StructType = StructType(StructField(ts,TimestampNTZType,true))
scala> sql("INSERT INTO t VALUES (TIMESTAMP_LTZ'1970-01-01 00:00:01')")
scala> spark.conf.unset("spark.sql.timestampType")
scala> spark.table("t").show()
+-------------------+
| ts|
+-------------------+
|1969-12-31 16:00:01|
+-------------------+
```
- For `STORED AS ORC`, the native ORC writer writes ORC `bigint` values, and reading the table fails with `UNSUPPORTED_FEATURE.ORC_TYPE_CAST` later.
- Reading Hive tables fails with `IllegalArgumentException: Error: type expected at the position 0 of 'timestamp_ntz' but 'timestamp_ntz' is found.` in Hive serde readers, e.g. `STORED AS TEXTFILE`, and with `UNSUPPORTED_FEATURE.ORC_TYPE_CAST` in the native ORC reader.
- The native Parquet reader and `spark.sql.orc.impl=hive` return wrong values, e.g. `1970-01-01T08:00:01` for `TIMESTAMP_LTZ'1970-01-01 00:00:01'` in `America/Los_Angeles`.
`spark.read.orc` with `spark.sql.orc.impl=hive` also infers ORC `timestamp` columns as `TIMESTAMP_NTZ` and returns the same wrong values.
### Does this PR introduce _any_ user-facing change?
Yes, this fixes the wrong data and query results since Apache Spark 3.4.0 ([v3.4.0](https://github.com/apache/spark/releases/tag/v3.4.0) (2023-04-07)). With `spark.sql.timestampType=TIMESTAMP_NTZ`, Hive `timestamp` columns are `TIMESTAMP_LTZ` as with the default `spark.sql.timestampType`.
### How was this patch tested?
Pass the CIs with the newly added test cases in `HiveClientImplSuite`, `HiveOrcSourceSuite` and `HiveSerDeReadWriteSuite`. The test cases fail without this fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59674 from dongjoon-hyun/SPARK-60424.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 537f8a115d752d89f44a44eef903f03ff4c5ed51)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../spark/sql/hive/client/HiveClientImpl.scala | 13 +++++++++-
.../spark/sql/hive/orc/OrcFileOperator.scala | 6 ++---
.../sql/hive/client/HiveClientImplSuite.scala | 16 +++++++++++-
.../hive/execution/HiveSerDeReadWriteSuite.scala | 29 +++++++++++++++++++++-
.../spark/sql/hive/orc/HiveOrcSourceSuite.scala | 18 ++++++++++++++
5 files changed, 76 insertions(+), 6 deletions(-)
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index 6c265ce6a43f..a37b33ab49c0 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -1122,6 +1122,17 @@ private[hive] object HiveClientImpl extends Logging {
new FieldSchema(c.name, typeString, c.getComment().orNull)
}
+ /**
+ * Parses a Hive type string. The parser resolves `timestamp` by spark.sql.timestampType, but
+ * Hive `timestamp` is always TIMESTAMP_LTZ in Spark and Hive has no TIMESTAMP_NTZ type, so map
+ * TIMESTAMP_NTZ back to TIMESTAMP_LTZ.
+ */
+ def parseHiveType(typeStr: String): DataType = {
+ CatalystSqlParser.parseDataType(typeStr).transformRecursively {
+ case TimestampNTZType => TimestampType
+ }
+ }
+
/** Get the Spark SQL native DataType from Hive's FieldSchema. */
private def getSparkSQLDataType(hc: FieldSchema): DataType = {
// For struct types, Hive metastore API uses unquoted element names, so does the spark catalyst
@@ -1136,7 +1147,7 @@ private[hive] object HiveClientImpl extends Logging {
// map<string,struct<x:int,y.z:int>> -> map<string,struct<`x`:int,`y.z`:int>>
val typeStr = hc.getType.replaceAll("(?<=struct<|,)([^,<:]+)(?=:)", "`$1`")
try {
- CatalystSqlParser.parseDataType(typeStr)
+ parseHiveType(typeStr)
} catch {
case e: ParseException =>
// Hive's union type (uniontype<...>) is not supported by Spark SQL and makes the parser
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileOperator.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileOperator.scala
index a46b435d7e89..48c0afe7f36e 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileOperator.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileOperator.scala
@@ -28,8 +28,8 @@ import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.PATH
-import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.hive.client.HiveClientImpl
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.ThreadUtils
@@ -105,7 +105,7 @@ private[hive] object OrcFileOperator extends Logging {
val readerInspector = reader.getObjectInspector.asInstanceOf[StructObjectInspector]
val schema = readerInspector.getTypeName
logDebug(s"Reading schema from file $paths, got Hive schema string: $schema")
- CatalystSqlParser.parseDataType(schema).asInstanceOf[StructType]
+ HiveClientImpl.parseHiveType(schema).asInstanceOf[StructType]
}
}
@@ -123,7 +123,7 @@ private[hive] object OrcFileOperator extends Logging {
val readerInspector = reader.getObjectInspector.asInstanceOf[StructObjectInspector]
val schema = readerInspector.getTypeName
logDebug(s"Reading schema from file $file., got Hive schema string: $schema")
- CatalystSqlParser.parseDataType(schema).asInstanceOf[StructType]
+ HiveClientImpl.parseHiveType(schema).asInstanceOf[StructType]
})
}.flatten
}
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/HiveClientImplSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/HiveClientImplSuite.scala
index 6dd3b16637b3..bea93a6695c7 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/HiveClientImplSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/HiveClientImplSuite.scala
@@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
import org.apache.spark.sql.hive.{HiveUtils, StaticInitFlags, StaticInitInputFormat, StaticInitOutputFormat}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{ArrayType, MapType, StructType, TimestampType}
class HiveClientImplSuite extends SparkFunSuite {
@@ -103,4 +103,18 @@ class HiveClientImplSuite extends SparkFunSuite {
"fieldType" -> "\"STRUCT<A:UNIONTYPE<INT,STRING>>\"",
"fieldName" -> "`c`"))
}
+
+ test("SPARK-60424: Hive timestamp type is TIMESTAMP_LTZ regardless of " +
+ "spark.sql.timestampType") {
+ val column = new FieldSchema(
+ "c", "struct<a:timestamp,b:array<timestamp>,c:map<timestamp,timestamp>>", null)
+ val conf = new SQLConf()
+ conf.setConf(SQLConf.TIMESTAMP_TYPE, SQLConf.TimestampTypes.TIMESTAMP_NTZ)
+ SQLConf.withExistingConf(conf) {
+ assert(HiveClientImpl.fromHiveColumn(column).dataType === new StructType()
+ .add("a", TimestampType)
+ .add("b", ArrayType(TimestampType))
+ .add("c", MapType(TimestampType, TimestampType)))
+ }
+ }
}
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveSerDeReadWriteSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveSerDeReadWriteSuite.scala
index f64750e3d1a8..f442841f0152 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveSerDeReadWriteSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveSerDeReadWriteSuite.scala
@@ -29,7 +29,8 @@ import org.apache.spark.internal.config.HADOOP_RDD_IGNORE_EMPTY_SPLITS
import org.apache.spark.sql.{QueryTest, Row}
import org.apache.spark.sql.hive.HiveUtils.{CONVERT_METASTORE_ORC, CONVERT_METASTORE_PARQUET, USE_DELEGATE_FOR_SYMLINK_TEXT_INPUT_FORMAT}
import org.apache.spark.sql.hive.test.TestHiveSingleton
-import org.apache.spark.sql.internal.SQLConf.{ORC_IMPLEMENTATION}
+import org.apache.spark.sql.internal.SQLConf.{ORC_IMPLEMENTATION, TIMESTAMP_TYPE}
+import org.apache.spark.sql.types.{StructType, TimestampType}
import org.apache.spark.tags.SlowHiveTest
@SlowHiveTest
@@ -313,4 +314,30 @@ class HiveSerDeReadWriteSuite extends QueryTest with TestHiveSingleton {
}
}
}
+
+ test("SPARK-60424: Hive TIMESTAMP is TIMESTAMP_LTZ regardless of spark.sql.timestampType") {
+ Seq(
+ "TEXTFILE" -> Nil,
+ "ORC" -> Nil,
+ "ORC" -> Seq(CONVERT_METASTORE_ORC.key -> "true"),
+ "ORC" -> Seq(CONVERT_METASTORE_ORC.key -> "true", ORC_IMPLEMENTATION.key -> "native"),
+ "PARQUET" -> Nil,
+ "PARQUET" -> Seq(CONVERT_METASTORE_PARQUET.key -> "true")
+ ).foreach { case (fileFormat, confs) =>
+ withTable("hive_serde") {
+ spark.sql(s"CREATE TABLE hive_serde (c1 TIMESTAMP) STORED AS $fileFormat")
+ hiveClient.runSqlHive("INSERT INTO TABLE hive_serde values('2019-04-11 15:50:00')")
+ withSQLConf(confs :+ (TIMESTAMP_TYPE.key -> "TIMESTAMP_NTZ"): _*) {
+ assert(spark.table("hive_serde").schema === new StructType().add("c1", TimestampType))
+ spark.sql("INSERT INTO TABLE hive_serde VALUES (TIMESTAMP_LTZ'2019-04-12 15:50:00')")
+ checkAnswer(
+ spark.table("hive_serde"),
+ Seq(Row(Timestamp.valueOf("2019-04-11 15:50:00")),
+ Row(Timestamp.valueOf("2019-04-12 15:50:00"))))
+ }
+ assert(hiveClient.runSqlHive("SELECT c1 FROM hive_serde ORDER BY c1") ===
+ Seq("2019-04-11 15:50:00", "2019-04-12 15:50:00"))
+ }
+ }
+ }
}
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
index 782c5c0bc934..3f6825cbf39a 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/HiveOrcSourceSuite.scala
@@ -569,4 +569,22 @@ class HiveOrcSourceSuite extends OrcSuite with TestHiveSingleton {
}
}
}
+
+ test("SPARK-60424: ORC TIMESTAMP is inferred as TIMESTAMP_LTZ regardless of " +
+ "spark.sql.timestampType") {
+ withSQLConf(SQLConf.ORC_IMPLEMENTATION.key -> "hive") {
+ withTempPath { dir =>
+ val df = sql("SELECT ts, named_struct('ts', ts) AS s, array(ts) AS a, map('k', ts) AS m " +
+ "FROM VALUES (TIMESTAMP_LTZ'1970-01-01 00:00:01') AS t(ts)")
+ df.write.orc(dir.getCanonicalPath)
+ withSQLConf(SQLConf.TIMESTAMP_TYPE.key -> "TIMESTAMP_NTZ") {
+ Seq("false", "true").foreach { mergeSchema =>
+ val readDf = spark.read.option("mergeSchema", mergeSchema).orc(dir.getCanonicalPath)
+ assert(readDf.schema === df.schema.asNullable)
+ checkAnswer(readDf, df)
+ }
+ }
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org