This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new b13b8135f15e [SPARK-60545][SQL] Raise a clear error for
`TIMESTAMP_NTZ` columns in Hive serde tables
b13b8135f15e is described below
commit b13b8135f15e6dac6ad1be5a27613f9a2f6be376
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Mon Oct 26 09:42:26 2026 -0700
[SPARK-60545][SQL] Raise a clear error for `TIMESTAMP_NTZ` columns in Hive
serde tables
### What changes were proposed in this pull request?
This PR aims to raise a clear error for `TIMESTAMP_NTZ` columns in Hive
serde tables.
Hive has no `TIMESTAMP_NTZ` type, so Hive serde tables cannot store
`TIMESTAMP_NTZ` columns. However, Spark passes `timestamp_ntz` to Hive as it is
and fails with a Hive internal error.
Like SPARK-60493 for the partition columns, this PR adds
`HiveExternalCatalog.verifyHiveSerdeDataColumnTypes`, which rejects the
`TIMESTAMP_NTZ` data columns of a Hive serde table, including the nested types
(`STRUCT`, `ARRAY` and `MAP`). It raises the existing
`UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE` error with the `Hive` format. Spark
calls it where it creates or alters a Hive serde table in Hive metastore:
- `HiveExternalCatalog.createTable` for `CREATE TABLE`, `CREATE TABLE AS
SELECT`, `CREATE TABLE LIKE`, `DataFrameWriter.saveAsTable` and
`Catalog.createTable`
- `CreateHiveTableAsSelectCommand` before it drops the existing table for
`DataFrameWriter.saveAsTable` with `SaveMode.Overwrite`
- `HiveExternalCatalog.alterTableSchema` for `ALTER TABLE ADD COLUMNS`
`CREATE TABLE IF NOT EXISTS` is still a no-op for an existing table because
`HiveExternalCatalog.createTable` is called only when the table doesn't exist.
In addition, like the `TIME` type (SPARK-57556), the nanosecond timestamp
types (SPARK-57815) and the geospatial types (SPARK-60519),
`HiveFileFormat.supportDataType` rejects `TIMESTAMP_NTZ` for `INSERT OVERWRITE
DIRECTORY ... STORED AS` through the Hive serde.
### Why are the changes needed?
With `spark.sql.timestampType=TIMESTAMP_NTZ`, the `TIMESTAMP` keyword in
DDL statements is `TIMESTAMP_NTZ` by design. So, the simple DDL statement below
fails with a Hive internal error, which doesn't tell the users which column is
wrong or why.
```sql
SET spark.sql.timestampType=TIMESTAMP_NTZ;
CREATE TABLE t (ts TIMESTAMP) STORED AS PARQUET;
```
```
org.apache.hadoop.hive.ql.metadata.HiveException:
java.lang.IllegalArgumentException: Error: type expected at the position 0 of
'timestamp_ntz' but 'timestamp_ntz' is found.
```
This issue was found during the review of #59674.
### Does this PR introduce _any_ user-facing change?
Yes, only the error changes. These statements already failed before this
PR. For the above example, the error becomes the following.
```
[UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE] The Hive datasource doesn't support
the column `ts` of the type "TIMESTAMP_NTZ". SQLSTATE: 0A000
```
For `INSERT OVERWRITE DIRECTORY ... STORED AS`, this error is the cause of
the `SparkException` like the other unsupported types.
### How was this patch tested?
Pass the CIs with the newly added test cases in `HiveDDLSuite`,
`AlterTableAddColumnsSuite` and `InsertSuite`. The test cases fail without this
fix, except the `CREATE TABLE IF NOT EXISTS` test case, which checks that the
existing no-op behavior is kept.
In addition, `TIMESTAMP_NTZ` is removed from the SPARK-60515 test case in
`MetastoreDataSourcesSuite`. That test case uses the types that the Hive SerDe
rejects, and this PR now rejects `TIMESTAMP_NTZ` before Hive does.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59783 from dongjoon-hyun/SPARK-60545.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/sql/hive/HiveExternalCatalog.scala | 17 +++++
.../execution/CreateHiveTableAsSelectCommand.scala | 1 +
.../spark/sql/hive/execution/HiveFileFormat.scala | 5 +-
.../org/apache/spark/sql/hive/InsertSuite.scala | 33 +++++++++
.../spark/sql/hive/MetastoreDataSourcesSuite.scala | 2 +-
.../spark/sql/hive/execution/HiveDDLSuite.scala | 84 ++++++++++++++++++++++
.../command/AlterTableAddColumnsSuite.scala | 24 +++++++
7 files changed, 164 insertions(+), 2 deletions(-)
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
index 484bde04aeb3..0379fbffebaf 100644
---
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
+++
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
@@ -272,6 +272,9 @@ private[spark] class HiveExternalCatalog(conf: SparkConf,
hadoopConf: Configurat
ignoreIfExists)
} else {
verifyHiveSerdePartitionColumnTypes(tableDefinition)
+ if (!tableDefinition.isViewLike) {
+ verifyHiveSerdeDataColumnTypes(tableDefinition)
+ }
val hiveCompatibleSchema = schemaWithNoCollation match {
// Spark-created views do not have to be Hive compatible. If the data
type is not
// Hive compatible, we can set schema to empty so that Spark can still
read this
@@ -802,6 +805,7 @@ private[spark] class HiveExternalCatalog(conf: SparkConf,
hadoopConf: Configurat
client.alterTableDataSchema(db, table, EMPTY_DATA_SCHEMA,
schemaProps)
}
} else {
+ verifyHiveSerdeDataColumnTypes(oldTable.copy(schema = newSchema))
client.alterTableDataSchema(db, table, hiveSchema, schemaProps)
}
}
@@ -1587,4 +1591,17 @@ object HiveExternalCatalog {
throw
QueryCompilationErrors.dataTypeUnsupportedByDataSourceError("Hive", f)
}
}
+
+ /**
+ * Hive has no TIMESTAMP_NTZ type. So, this throws an error if the given
Hive serde table has a
+ * TIMESTAMP_NTZ data column, including the nested types.
+ */
+ private[hive] def verifyHiveSerdeDataColumnTypes(table: CatalogTable): Unit
= {
+ table.schema
+ .filterNot(f => table.partitionColumnNames.contains(f.name))
+ .find(_.dataType.existsRecursively(_.isInstanceOf[TimestampNTZType]))
+ .foreach { f =>
+ throw
QueryCompilationErrors.dataTypeUnsupportedByDataSourceError("Hive", f)
+ }
+ }
}
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/CreateHiveTableAsSelectCommand.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/CreateHiveTableAsSelectCommand.scala
index a01831553b7e..57e884de31c1 100644
---
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/CreateHiveTableAsSelectCommand.scala
+++
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/CreateHiveTableAsSelectCommand.scala
@@ -76,6 +76,7 @@ case class CreateHiveTableAsSelectCommand(
// table. Drop it here, after the analysis and the checks of the new
table, because the
// new table is created before the write.
HiveExternalCatalog.verifyHiveSerdePartitionColumnTypes(newTableDesc)
+ HiveExternalCatalog.verifyHiveSerdeDataColumnTypes(newTableDesc)
val qualifiedNewTableDesc =
newTableDesc.copy(identifier =
catalog.qualifyIdentifier(tableIdentifier))
sparkSession.sharedState.externalCatalog.unwrapped.asInstanceOf[HiveExternalCatalog]
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
index 189d90235bc1..55cc0c9cf93e 100644
---
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
+++
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
@@ -41,7 +41,7 @@ import
org.apache.spark.sql.execution.datasources.{FileFormat, OutputWriter, Out
import org.apache.spark.sql.hive.{HiveInspectors, HiveTableUtil}
import org.apache.spark.sql.internal.SessionStateHelper
import org.apache.spark.sql.sources.DataSourceRegister
-import org.apache.spark.sql.types.{AnyTimestampNanoType, ArrayType, DataType,
GeographyType, GeometryType, MapType, StructType, TimeType, UserDefinedType}
+import org.apache.spark.sql.types.{AnyTimestampNanoType, ArrayType, DataType,
GeographyType, GeometryType, MapType, StructType, TimestampNTZType, TimeType,
UserDefinedType}
import org.apache.spark.util.SerializableJobConf
/**
@@ -127,6 +127,9 @@ case class HiveFileFormat(fileSinkConf: FileSinkDesc)
// Hive has no geospatial types, so they cannot be stored in a Hive serde
table.
case _: GeometryType | _: GeographyType => false
+ // Hive has no TIMESTAMP_NTZ type, so it cannot be stored in a Hive serde
table.
+ case _: TimestampNTZType => false
+
case st: StructType => st.forall { f => supportDataType(f.dataType) }
case ArrayType(elementType, _) => supportDataType(elementType)
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
index 1367041b18fd..b68a4c7d1db3 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala
@@ -800,6 +800,39 @@ class InsertSuite extends QueryTest with TestHiveSingleton
with BeforeAndAfter {
}
}
+ test("SPARK-60545: TIMESTAMP_NTZ type is unsupported when writing to a Hive
serde directory") {
+ // Disable native data source conversion so that PARQUET and ORC also go
through the Hive
+ // serde path (HiveFileFormat) like TEXTFILE.
+ withSQLConf(HiveUtils.CONVERT_METASTORE_INSERT_DIR.key -> "false") {
+ Seq("TEXTFILE", "PARQUET", "ORC").foreach { fileFormat =>
+ Seq(
+ "TIMESTAMP_NTZ'2026-01-01 00:00:00'" -> "TIMESTAMP_NTZ",
+ "ARRAY(TIMESTAMP_NTZ'2026-01-01 00:00:00')" -> "ARRAY<TIMESTAMP_NTZ>"
+ ).foreach { case (expr, columnType) =>
+ withTempDir { dir =>
+ // InsertIntoHiveDirCommand wraps the failure in a SparkException,
so assert on the
+ // cause.
+ val e = intercept[SparkException] {
+ sql(
+ s"""
+ |INSERT OVERWRITE LOCAL DIRECTORY '${dir.toURI.getPath}'
+ |STORED AS $fileFormat
+ |SELECT $expr AS c
+ """.stripMargin)
+ }
+ checkError(
+ exception = e.getCause.asInstanceOf[AnalysisException],
+ condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ parameters = Map(
+ "columnName" -> "`c`",
+ "columnType" -> s""""$columnType"""",
+ "format" -> "Hive"))
+ }
+ }
+ }
+ }
+ }
+
test("insert overwrite to dir from temp table") {
withTempView("test_insert_table") {
spark.range(10).selectExpr("id", "id AS
str").createOrReplaceTempView("test_insert_table")
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
index 71fd3e9b28fc..476bd133aaf5 100644
---
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
+++
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
@@ -1343,7 +1343,7 @@ class MetastoreDataSourcesSuite extends QueryTest
}
}
// The Hive SerDe can't parse these types.
- Seq("TIME(6)", "TIMESTAMP_NTZ", "TIMESTAMP_NTZ(9)", "VARIANT").foreach {
dataType =>
+ Seq("TIME(6)", "TIMESTAMP_NTZ(9)", "VARIANT").foreach { dataType =>
checkKeepsTable("textfile", "id", s"CAST(NULL AS $dataType) AS c")
}
// The Parquet Hive SerDe doesn't support the void type.
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index ff07dd6b7020..9e0c1b988ac4 100644
---
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -3295,6 +3295,90 @@ class HiveDDLSuite
}
}
+ test("SPARK-60545: Disallow TIMESTAMP_NTZ columns when the provider is
Hive") {
+ val tbl = "tbl_with_timestamp_ntz"
+ def checkUnsupported(columnType: String)(createTable: => Unit): Unit = {
+ checkError(
+ exception = intercept[AnalysisException](createTable),
+ condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ parameters = Map(
+ "columnName" -> "`c`",
+ "columnType" -> s""""$columnType"""",
+ "format" -> "Hive"))
+ assert(!spark.catalog.tableExists(tbl))
+ }
+ withTable(tbl, "src") {
+ withSQLConf(SQLConf.TIMESTAMP_TYPE.key -> "TIMESTAMP_NTZ") {
+ checkUnsupported("TIMESTAMP_NTZ") {
+ sql(s"CREATE TABLE $tbl (c TIMESTAMP) STORED AS PARQUET")
+ }
+ checkUnsupported("TIMESTAMP_NTZ") {
+ sql(s"CREATE TABLE $tbl STORED AS ORC AS SELECT TIMESTAMP'2026-01-01
00:00:00' AS c")
+ }
+ }
+ Seq(
+ "TIMESTAMP_NTZ",
+ "STRUCT<a: TIMESTAMP_NTZ>",
+ "ARRAY<TIMESTAMP_NTZ>",
+ "MAP<STRING, TIMESTAMP_NTZ>").foreach { dataType =>
+ checkUnsupported(dataType) {
+ sql(s"CREATE TABLE $tbl (c $dataType) STORED AS TEXTFILE")
+ }
+ checkUnsupported(dataType) {
+ sql(s"CREATE TABLE $tbl (pt INT, c $dataType) USING HIVE PARTITIONED
BY (pt)")
+ }
+ checkUnsupported(dataType) {
+ sql(s"CREATE TABLE $tbl STORED AS PARQUET AS SELECT CAST(NULL AS
$dataType) AS c")
+ }
+ checkUnsupported(dataType) {
+ sql(s"SELECT CAST(NULL AS $dataType) AS
c").write.format("hive").saveAsTable(tbl)
+ }
+ sql(s"CREATE TABLE src (c $dataType) USING PARQUET")
+ checkUnsupported(dataType) {
+ sql(s"CREATE TABLE $tbl LIKE src STORED AS ORC")
+ }
+ sql("DROP TABLE src")
+ }
+ }
+ }
+
+ test("SPARK-60545: CREATE TABLE IF NOT EXISTS with TIMESTAMP_NTZ columns is
a no-op for " +
+ "an existing Hive serde table") {
+ val tbl = "tbl_with_timestamp_ntz"
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (c TIMESTAMP) STORED AS PARQUET")
+ sql(s"INSERT INTO $tbl VALUES (TIMESTAMP'2026-01-01 00:00:00')")
+ withSQLConf(SQLConf.TIMESTAMP_TYPE.key -> "TIMESTAMP_NTZ") {
+ sql(s"CREATE TABLE IF NOT EXISTS $tbl (c TIMESTAMP) STORED AS PARQUET")
+ sql(s"CREATE TABLE IF NOT EXISTS $tbl STORED AS PARQUET " +
+ "AS SELECT TIMESTAMP'2026-01-01 00:00:00' AS c")
+ }
+ sql(s"CREATE TABLE IF NOT EXISTS $tbl (c TIMESTAMP_NTZ) STORED AS
PARQUET")
+ assert(spark.table(tbl).schema === new StructType().add("c",
TimestampType))
+ assert(spark.table(tbl).count() === 1)
+ }
+ }
+
+ test("SPARK-60545: saveAsTable with mode Overwrite keeps the existing table
when the new " +
+ "Hive serde table has a TIMESTAMP_NTZ column") {
+ val tbl = "tbl_with_timestamp_ntz"
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (c INT) STORED AS PARQUET")
+ sql(s"INSERT INTO $tbl VALUES (1)")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SELECT TIMESTAMP_NTZ'2026-01-01 00:00:00' AS c")
+ .write.format("hive").mode(SaveMode.Overwrite).saveAsTable(tbl)
+ },
+ condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ parameters = Map(
+ "columnName" -> "`c`",
+ "columnType" -> "\"TIMESTAMP_NTZ\"",
+ "format" -> "Hive"))
+ checkAnswer(spark.table(tbl), Row(1))
+ }
+ }
+
test("SPARK-37046: Alter view should preserve column case with view
definition change") {
withView("v") {
// Changing view definition should preserve column case
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableAddColumnsSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableAddColumnsSuite.scala
index 521ad759c302..3ce225941fa4 100644
---
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableAddColumnsSuite.scala
+++
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/command/AlterTableAddColumnsSuite.scala
@@ -18,8 +18,10 @@
package org.apache.spark.sql.hive.execution.command
import org.apache.spark.SparkUnsupportedOperationException
+import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId
import org.apache.spark.sql.execution.command.v1
+import org.apache.spark.sql.internal.SQLConf
/**
* The class contains tests for the `ALTER TABLE .. ADD COLUMNS` command to
check
@@ -48,4 +50,26 @@ class AlterTableAddColumnsSuite
check(tbl)
}
}
+
+ test("SPARK-60545: Disallow TIMESTAMP_NTZ columns when the provider is
Hive") {
+ def check(tbl: String, dataType: String, columnType: String): Unit = {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"ALTER TABLE $tbl ADD COLUMNS (c $dataType)")
+ },
+ condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ parameters = Map(
+ "columnName" -> "`c`",
+ "columnType" -> s""""$columnType"""",
+ "format" -> "Hive"))
+ }
+ withNamespaceAndTable("ns", "tbl") { tbl =>
+ sql(s"CREATE TABLE $tbl (id INT) $defaultUsing")
+ check(tbl, "TIMESTAMP_NTZ", "TIMESTAMP_NTZ")
+ check(tbl, "ARRAY<TIMESTAMP_NTZ>", "ARRAY<TIMESTAMP_NTZ>")
+ withSQLConf(SQLConf.TIMESTAMP_TYPE.key -> "TIMESTAMP_NTZ") {
+ check(tbl, "TIMESTAMP", "TIMESTAMP_NTZ")
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
