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 1a9f6574ebc9 [SPARK-60493][SQL] Disallow Hive incompatible partition
column types in Hive serde tables
1a9f6574ebc9 is described below
commit 1a9f6574ebc9a21e41ff670591bb61deeebecc25
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Mon Oct 26 08:10:03 2026 -0700
[SPARK-60493][SQL] Disallow Hive incompatible partition column types in
Hive serde tables
### What changes were proposed in this pull request?
This PR aims to disallow Hive incompatible partition column types in Hive
serde tables with a clear error, `UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE`. The
affected types are `TIMESTAMP_NTZ`, ANSI intervals, the nanosecond timestamp
types (`TIMESTAMP_NTZ(p)` and `TIMESTAMP_LTZ(p)`), and `TIME`.
Since SPARK-51418 (#50182), `HiveClientImpl.toHiveTable` stores Hive
incompatible partition column types with a placeholder type, `<derived from
deserializer>`, in Hive metastore. This works for data source tables because
Spark restores their schema from the table properties. However, it applies to
Hive serde tables too. `HiveClientImpl` drops the placeholder columns when it
reads a table, so `HiveExternalCatalog.restoreHiveSerdeTable` sees a schema
mismatch with the table properties [...]
This PR adds a check to the Hive serde table branch of
`HiveExternalCatalog.createTable`. It uses `isHiveCompatibleDataType`, the same
condition as the placeholder. Since all table creation paths go through it, the
check covers the following:
- `CREATE TABLE ... STORED AS` and `CREATE TABLE ... USING HIVE`
- `CREATE TABLE ... AS SELECT`
- `CREATE TABLE ... LIKE` with `STORED AS` or `USING HIVE`
- `DataFrameWriter.saveAsTable` and `DataFrameWriterV2.create` with the
`hive` provider
In addition, `CreateHiveTableAsSelectCommand` runs the same check before it
drops the existing table for `DataFrameWriter.saveAsTable` with
`SaveMode.Overwrite`. So, the existing table is kept when the check fails, like
the other checks (SPARK-60387).
Data source tables are not affected.
### Why are the changes needed?
Since Apache Spark 4.0.0, the following statement succeeds, but it creates
a broken table without the partition column `p`.
```sql
CREATE TABLE t (i INT) PARTITIONED BY (p TIMESTAMP_NTZ) STORED AS PARQUET;
```
```
spark-sql> INSERT INTO t PARTITION (p = TIMESTAMP_NTZ'2026-01-01 00:00:00')
VALUES (1);
[PARTITIONS_NOT_FOUND] The partition(s) `p` cannot be found in table
`spark_catalog`.`default`.`t`.
...
spark-sql> SHOW PARTITIONS t;
[INVALID_PARTITION_OPERATION.PARTITION_SCHEMA_IS_EMPTY] The partition
command is invalid. Table `spark_catalog`.`default`.`t` is not partitioned.
SQLSTATE: 42601
```
`CREATE TABLE AS SELECT` creates the table without the partition column, so
the values of `p` are silently lost.
```sql
CREATE TABLE t PARTITIONED BY (p) STORED AS PARQUET
AS SELECT 1 AS i, TIMESTAMP_NTZ'2026-01-01 00:00:00' AS p;
SELECT * FROM t; -- returns only `i`
```
With `spark.sql.timestampType=TIMESTAMP_NTZ`, `PARTITIONED BY (p
TIMESTAMP)` has the same issue. `TIME` partition columns have it too since
SPARK-60482 (#59720), which is not released yet. Moreover, Hive cannot read
these tables (`Unknown type <derived from deserializer>`), and `ALTER TABLE ADD
COLUMNS` corrupts the table metadata so that even `DROP TABLE` fails.
```
The metadata is corrupted. Unable to find the partition column names from
the schema. schema: struct<i:int,j:int>. Partition columns: [p]
```
Before SPARK-51418, Hive metastore rejected these tables at the table
creation, e.g., `InvalidObjectException(message:Invalid partition column type:
timestamp_ntz)`.
### Does this PR introduce _any_ user-facing change?
Yes. Creating a Hive serde table with these partition column types fails
with the following error instead of creating a broken table. Data source tables
are not affected.
```
[UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE] The Hive datasource doesn't support
the column `p` of the type "TIMESTAMP_NTZ". SQLSTATE: 0A000
```
Previously, some CTAS statements with ANSI interval partition columns,
e.g., `STORED AS PARQUET`, failed with
`UNSUPPORTED_FEATURE.HIVE_WITH_ANSI_INTERVALS` after creating and dropping the
table. Now, they fail with the above error before creating the table.
This PR doesn't change the existing tables.
### How was this patch tested?
Pass the CIs with the newly added test cases in `HiveDDLSuite`. Both test
cases fail without this fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59734 from dongjoon-hyun/SPARK-60493.
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/HiveDDLSuite.scala | 70 ++++++++++++++++++++++
3 files changed, 88 insertions(+)
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 c936d1633991..484bde04aeb3 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
@@ -44,6 +44,7 @@ import
org.apache.spark.sql.catalyst.catalog.ExternalCatalogUtils._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap,
CharVarcharUtils, QuotingUtils}
+import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.command.DDLUtils
import org.apache.spark.sql.execution.datasources.{PartitioningUtils,
SourceOptions}
import org.apache.spark.sql.hive.client.HiveClient
@@ -270,6 +271,7 @@ private[spark] class HiveExternalCatalog(conf: SparkConf,
hadoopConf: Configurat
hiveTable.withNewStorage(locationUri = tableLocation),
ignoreIfExists)
} else {
+ verifyHiveSerdePartitionColumnTypes(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
@@ -1570,4 +1572,19 @@ object HiveExternalCatalog {
case st: StringType => st.isUTF8BinaryCollation
case _ => true
}
+
+ /**
+ * Hive incompatible partition column types are stored with a placeholder
type in Hive
+ * metastore (see `HiveClientImpl.toHiveTable`), which only data source
tables can restore.
+ * So, this throws an error if the given Hive serde table has such a
partition column.
+ */
+ private[hive] def verifyHiveSerdePartitionColumnTypes(table: CatalogTable):
Unit = {
+ // Like `createTable`, ignore the collations, which Hive metastore doesn't
store.
+
SchemaUtils.replaceCollatedStringWithString(table.schema).asInstanceOf[StructType]
+ .filter(f => table.partitionColumnNames.contains(f.name))
+ .find(f => !isHiveCompatibleDataType(f.dataType))
+ .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 7e7e94da1e4b..a01831553b7e 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
@@ -75,6 +75,7 @@ case class CreateHiveTableAsSelectCommand(
// `SaveAsV1TableCommand` runs this command with `SaveMode.Overwrite`
to replace the
// 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)
val qualifiedNewTableDesc =
newTableDesc.copy(identifier =
catalog.qualifyIdentifier(tableIdentifier))
sparkSession.sharedState.externalCatalog.unwrapped.asInstanceOf[HiveExternalCatalog]
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 6b7987097916..ff07dd6b7020 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
@@ -3383,6 +3383,76 @@ class HiveDDLSuite
}
}
+ test("SPARK-60493: Disallow Hive incompatible partition column types in Hive
serde tables") {
+ val tbl = "tbl_with_incompatible_partition"
+ def checkUnsupported(columnType: String)(createTable: => Unit): Unit = {
+ checkError(
+ exception = intercept[AnalysisException](createTable),
+ condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ parameters = Map(
+ "columnName" -> "`pt`",
+ "columnType" -> s""""$columnType"""",
+ "format" -> "Hive"))
+ assert(!spark.catalog.tableExists(tbl))
+ }
+ withTable(tbl, "src") {
+ Seq(
+ "TIMESTAMP_NTZ",
+ "INTERVAL YEAR",
+ "INTERVAL DAY TO SECOND",
+ "TIMESTAMP_NTZ(9)",
+ "TIMESTAMP_LTZ(9)",
+ "TIME(6)").foreach { dataType =>
+ checkUnsupported(dataType) {
+ sql(s"CREATE TABLE $tbl (i INT) PARTITIONED BY (pt $dataType) STORED
AS PARQUET")
+ }
+ checkUnsupported(dataType) {
+ sql(s"CREATE TABLE $tbl (i INT, pt $dataType) USING HIVE PARTITIONED
BY (pt)")
+ }
+ checkUnsupported(dataType) {
+ sql(s"CREATE TABLE $tbl PARTITIONED BY (pt) STORED AS TEXTFILE " +
+ s"AS SELECT 1 AS i, CAST(NULL AS $dataType) AS pt")
+ }
+ checkUnsupported(dataType) {
+ sql(s"SELECT 1 AS i, CAST(NULL AS $dataType) AS pt")
+ .write.format("hive").partitionBy("pt").saveAsTable(tbl)
+ }
+ sql(s"CREATE TABLE src (i INT, pt $dataType) USING PARQUET PARTITIONED
BY (pt)")
+ checkUnsupported(dataType) {
+ sql(s"CREATE TABLE $tbl LIKE src STORED AS ORC")
+ }
+ sql("DROP TABLE src")
+ }
+ withSQLConf(SQLConf.TIMESTAMP_TYPE.key -> "TIMESTAMP_NTZ") {
+ checkUnsupported("TIMESTAMP_NTZ") {
+ sql(s"CREATE TABLE $tbl (i INT) PARTITIONED BY (pt TIMESTAMP) STORED
AS AVRO")
+ }
+ }
+ }
+ }
+
+ test("SPARK-60493: saveAsTable with mode Overwrite keeps the existing table
when the new " +
+ "Hive serde table has a Hive incompatible partition column type") {
+ val tbl = "tbl_with_incompatible_partition"
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (i INT, pt INT) USING HIVE PARTITIONED BY (pt)")
+ sql(s"INSERT INTO $tbl PARTITION (pt = 2) VALUES (1)")
+ Seq("TIMESTAMP_NTZ", "INTERVAL YEAR", "TIMESTAMP_NTZ(9)",
"TIME(6)").foreach { dataType =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"SELECT 1 AS i, CAST(NULL AS $dataType) AS pt")
+
.write.format("hive").mode(SaveMode.Overwrite).partitionBy("pt").saveAsTable(tbl)
+ },
+ condition = "UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE",
+ parameters = Map(
+ "columnName" -> "`pt`",
+ "columnType" -> s""""$dataType"""",
+ "format" -> "Hive"))
+ checkAnswer(spark.table(tbl), Row(1, 2))
+ }
+ }
+ }
+
test("SPARK-52272: V2SessionCatalog does not alter schema on Hive Catalog") {
val spyCatalog = spy(spark.sessionState.catalog.externalCatalog)
val v1SessionCatalog = new SessionCatalog(spyCatalog)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
