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 6f8d0aaab792 [SPARK-60493][SQL] Disallow Hive incompatible partition 
column types in Hive serde tables
6f8d0aaab792 is described below

commit 6f8d0aaab7925225679e11d929a7c501eeb41941
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]>
    (cherry picked from commit 1a9f6574ebc9a21e41ff670591bb61deeebecc25)
    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]

Reply via email to