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 b5ad05b3d8e7 [SPARK-60565][SQL] Fix reading and writing collated
strings in Hive serde tables
b5ad05b3d8e7 is described below
commit b5ad05b3d8e79d7b4dc6dbab7446b3c5e5fc2e39
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Mon Oct 26 16:32:32 2026 -0700
[SPARK-60565][SQL] Fix reading and writing collated strings in Hive serde
tables
### What changes were proposed in this pull request?
This PR aims to remove the collations from the Hive column types in
`HiveClientImpl.toHiveColumn`, which converts Spark columns into Hive columns.
- Collated `STRING` types, including the nested ones in `ARRAY`, `STRUCT`,
and `MAP`, become the regular `string` type via
`SchemaUtils.replaceCollatedStringWithString`, like `HiveExternalCatalog` does
before calling Hive metastore (SPARK-47873).
- The raw `CHAR`/`VARCHAR` type string in the field metadata, e.g. `char(3)
collate UTF8_LCASE`, loses its ` collate <name>` clauses. The type string is
not parsed again, because the parsing depends on the session configurations.
For example, `struct<a:char(3),b:timestamp>` would become
`struct<a:char(3),b:timestamp_ntz>` with
`spark.sql.timestampType=TIMESTAMP_NTZ`.
### Why are the changes needed?
Hive has no collations. `HiveExternalCatalog` removes the collations before
calling Hive metastore, so a Hive serde table with collated string columns is
created with `string` columns in Hive metastore, and Spark restores the
collations from the table properties. However, `InsertIntoHiveTable`,
`InsertIntoHiveDirCommand`, and `HiveTableScanExec` build the Hive table from
the Spark schema via `HiveClientImpl.toHiveTable`, and `toHiveColumn` uses
`catalogString`, e.g. `string collate UN [...]
```sql
CREATE TABLE t (i INT, c STRING COLLATE UNICODE_CI) STORED AS TEXTFILE;
INSERT INTO t VALUES (1, 'a');
```
```
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0
in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0
(TID 0) (localhost executor driver): java.lang.IllegalArgumentException: Error:
type expected at the position 4 of 'int:string collate UNICODE_CI' but 'string
collate UNICODE_CI' is found.
```
Hive SerDe fails in the same way with
- `SELECT`, e.g. of the rows that Hive wrote,
- `CREATE TABLE ... AS SELECT` and `DataFrameWriter.saveAsTable` with
`format("hive")`, which also drop the new table,
- `INSERT OVERWRITE [LOCAL] DIRECTORY` with Hive SerDe, e.g. `ROW FORMAT
DELIMITED` or `STORED AS TEXTFILE`.
It affects the `TEXTFILE`, `SEQUENCEFILE`, `RCFILE`, and `AVRO` tables, and
the `ORC` and `PARQUET` tables when `spark.sql.hive.convertMetastoreOrc` and
`spark.sql.hive.convertMetastoreParquet` are `false`. An explicit `COLLATE
UTF8_BINARY` also fails because its `catalogString` is `string collate
UTF8_BINARY`. This has been the behavior since such tables can be created
(SPARK-47873, Apache Spark 4.0.0).
For collated `CHAR`/`VARCHAR` types (SPARK-54870), the raw type string in
the field metadata keeps the collations even after `HiveExternalCatalog`
removes them from the data types. So the Hive metastore calls fail too, e.g.
`ALTER TABLE ... ADD COLUMNS` on a table with a `CHAR(3) COLLATE UTF8_LCASE`
column, and `CREATE TABLE ... STORED AS PARQUET AS SELECT * FROM` such a table.
### Does this PR introduce _any_ user-facing change?
Yes. Spark reads and writes Hive serde tables with collated string columns,
and `ALTER TABLE ... ADD COLUMNS` works on tables with collated
`CHAR`/`VARCHAR` columns. Spark keeps the collations of the columns as before,
and Hive reads the tables with the regular string types.
Hive ORC and Parquet tables that Spark converts to the native data sources
(`spark.sql.hive.convertMetastoreOrc` and
`spark.sql.hive.convertMetastoreParquet`, both `true` by default) don't use
Hive SerDe, so this PR doesn't change them.
### How was this patch tested?
Pass the CIs with the newly added test cases.
- `HiveSerDeReadWriteSuite`: reads and writes collated `STRING`, `ARRAY`,
`STRUCT`, `MAP`, `CHAR`, and `VARCHAR` columns via Hive and Spark, and CTAS for
all Hive serde formats.
- `InsertSuite`: `INSERT OVERWRITE LOCAL DIRECTORY` with collated strings.
- `AlterTableAddColumnsSuite`: `ALTER TABLE ... ADD COLUMNS` on a table
with collated `CHAR` and `VARCHAR` columns.
- `HiveClientImplSuite`: `toHiveColumn` removes the collations, and keeps
`timestamp` with `spark.sql.timestampType=TIMESTAMP_NTZ`.
The new tests in `HiveSerDeReadWriteSuite` and `InsertSuite` fail without
this fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59799 from dongjoon-hyun/SPARK-60565.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/sql/hive/client/HiveClientImpl.scala | 15 +++++++--
.../org/apache/spark/sql/hive/InsertSuite.scala | 12 ++++++++
.../sql/hive/client/HiveClientImplSuite.scala | 28 ++++++++++++++++-
.../hive/execution/HiveSerDeReadWriteSuite.scala | 36 ++++++++++++++++++++++
.../command/AlterTableAddColumnsSuite.scala | 12 +++++++-
5 files changed, 99 insertions(+), 4 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 eac12045fece..23fdfd07c366 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
@@ -69,6 +69,7 @@ import org.apache.spark.sql.hive.{HiveExternalCatalog,
HiveUtils}
import org.apache.spark.sql.hive.HiveExternalCatalog.DATASOURCE_SCHEMA
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
+import org.apache.spark.sql.util.SchemaUtils
import org.apache.spark.util.{CircularBuffer, Utils}
/**
@@ -1133,15 +1134,25 @@ private[hive] object HiveClientImpl extends Logging {
// See org.apache.hadoop.hive.metastore.MetaStoreUtils#validateColumnType
for more details.
lazy val INCOMPATIBLE_PARTITION_TYPE_PLACEHOLDER = "<derived from
deserializer>"
+ // The collation clause of a type string, e.g. ` collate UTF8_LCASE` of
+ // `char(3) collate UTF8_LCASE`.
+ private val COLLATION_CLAUSE = " collate \\w+".r
/** Converts the native StructField to Hive's FieldSchema. */
def toHiveColumn(c: StructField): FieldSchema = {
+ // Hive has no collations, so collated strings are passed to Hive as
regular strings.
+ val dataType = SchemaUtils.replaceCollatedStringWithString(c.dataType)
// For Hive Serde, we still need to to restore the raw type for char and
varchar type.
// When reading data in parquet, orc, or avro file format with string type
for char,
// the tailing spaces may lost if we are not going to pad it.
val typeString = if (SQLConf.get.charVarcharAsString) {
- c.dataType.catalogString
+ dataType.catalogString
} else {
-
CharVarcharUtils.getRawTypeString(c.metadata).getOrElse(c.dataType.catalogString)
+ // The raw type string keeps the collations of char and varchar types.
Remove them without
+ // parsing the string again because the parsing depends on the session
configurations,
+ // e.g. `timestamp` becomes TIMESTAMP_NTZ with
spark.sql.timestampType=TIMESTAMP_NTZ.
+ CharVarcharUtils.getRawTypeString(c.metadata)
+ .map(COLLATION_CLAUSE.replaceAllIn(_, ""))
+ .getOrElse(dataType.catalogString)
}
new FieldSchema(c.name, typeString, c.getComment().orNull)
}
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 b68a4c7d1db3..6c260825ece8 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
@@ -833,6 +833,18 @@ class InsertSuite extends QueryTest with TestHiveSingleton
with BeforeAndAfter {
}
}
+ test("SPARK-60565: Write collated strings to a Hive serde directory") {
+ withTempDir { dir =>
+ sql(
+ s"""
+ |INSERT OVERWRITE LOCAL DIRECTORY '${dir.toURI.getPath}'
+ |ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' COLLECTION ITEMS
TERMINATED BY ':'
+ |SELECT 'a' COLLATE UNICODE_CI AS c1, ARRAY('b' COLLATE UNICODE_CI,
'c') AS c2
+ """.stripMargin)
+ checkAnswer(spark.read.text(dir.getCanonicalPath), Row("a,b:c"))
+ }
+ }
+
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/client/HiveClientImplSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/HiveClientImplSuite.scala
index bea93a6695c7..e0d6de4f4ff6 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
@@ -22,9 +22,10 @@ import org.apache.hadoop.hive.metastore.api.FieldSchema
import org.apache.spark.{SparkFunSuite, SparkUnsupportedOperationException}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType}
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.hive.{HiveUtils, StaticInitFlags,
StaticInitInputFormat, StaticInitOutputFormat}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{ArrayType, MapType, StructType,
TimestampType}
+import org.apache.spark.sql.types.{ArrayType, CharType, MapType, StringType,
StructField, StructType, TimestampType, VarcharType}
class HiveClientImplSuite extends SparkFunSuite {
@@ -117,4 +118,29 @@ class HiveClientImplSuite extends SparkFunSuite {
.add("c", MapType(TimestampType, TimestampType)))
}
}
+
+ test("SPARK-60565: toHiveColumn removes collations") {
+ Seq(
+ StringType("UNICODE_CI") -> "string",
+ StringType("UTF8_BINARY") -> "string",
+ ArrayType(StringType("UNICODE_CI")) -> "array<string>",
+ new StructType().add("f", StringType("UTF8_LCASE")) ->
"struct<f:string>",
+ MapType(StringType, StringType("UNICODE_CI")) -> "map<string,string>",
+ CharType(3, "UTF8_LCASE") -> "char(3)",
+ VarcharType(3, "UNICODE_CI") -> "varchar(3)"
+ ).foreach { case (dataType, typeString) =>
+ assert(HiveClientImpl.toHiveColumn(StructField("c", dataType)).getType
=== typeString)
+ }
+
+ // The raw CHAR/VARCHAR type string in the field metadata keeps the
collations. Removing them
+ // must not depend on spark.sql.timestampType.
+ val field = CharVarcharUtils.replaceCharVarcharWithStringInSchema(new
StructType()
+ .add("c", new StructType().add("a", CharType(3, "UTF8_LCASE")).add("b",
TimestampType)))
+ .head
+ val conf = new SQLConf()
+ conf.setConf(SQLConf.TIMESTAMP_TYPE, SQLConf.TimestampTypes.TIMESTAMP_NTZ)
+ SQLConf.withExistingConf(conf) {
+ assert(HiveClientImpl.toHiveColumn(field).getType ===
"struct<a:char(3),b:timestamp>")
+ }
+ }
}
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 f442841f0152..c0618217c547 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
@@ -340,4 +340,40 @@ class HiveSerDeReadWriteSuite extends QueryTest with
TestHiveSingleton {
}
}
}
+
+ test("SPARK-60565: Read and write collated strings in Hive serde tables") {
+ Seq("SEQUENCEFILE", "TEXTFILE", "RCFILE", "ORC", "PARQUET",
"AVRO").foreach { fileFormat =>
+ withTable("hive_serde", "hive_serde_ctas") {
+ spark.sql(
+ s"""
+ |CREATE TABLE hive_serde (
+ | c1 STRING COLLATE UNICODE_CI,
+ | c2 ARRAY<STRING COLLATE UNICODE_CI>,
+ | c3 STRUCT<f: STRING COLLATE UTF8_LCASE>,
+ | c4 MAP<STRING, STRING COLLATE UNICODE_CI>,
+ | c5 CHAR(3) COLLATE UTF8_LCASE,
+ | c6 VARCHAR(3) COLLATE UNICODE_CI)
+ |STORED AS $fileFormat
+ |""".stripMargin)
+ hiveClient.runSqlHive("INSERT INTO TABLE hive_serde SELECT " +
+ "'a', ARRAY('b'), NAMED_STRUCT('f', 'c'), MAP('k', 'd'), 'e', 'f'
FROM (SELECT 1) t")
+ spark.sql("INSERT INTO TABLE hive_serde " +
+ "SELECT 'A', ARRAY('B'), NAMED_STRUCT('f', 'C'), MAP('k', 'D'), 'E',
'F'")
+ val expected = Seq(
+ Row("a", Seq("b"), Row("c"), Map("k" -> "d"), "e ", "f"),
+ Row("A", Seq("B"), Row("C"), Map("k" -> "D"), "E ", "F"))
+ checkAnswer(spark.table("hive_serde"), expected)
+ // The collations are kept when Spark reads the table.
+ checkAnswer(
+ spark.sql("SELECT count(*) FROM hive_serde WHERE c1 = 'a' AND c2[0]
= 'b' " +
+ "AND c3.f = 'c' AND c4['k'] = 'd' AND c5 = 'e' AND c6 = 'f'"),
+ Row(2))
+ assert(hiveClient.runSqlHive("SELECT c1 FROM hive_serde ORDER BY c1")
=== Seq("A", "a"))
+
+ spark.sql(s"CREATE TABLE hive_serde_ctas STORED AS $fileFormat AS
SELECT * FROM hive_serde")
+ assert(spark.table("hive_serde_ctas").schema ===
spark.table("hive_serde").schema)
+ checkAnswer(spark.table("hive_serde_ctas"), expected)
+ }
+ }
+ }
}
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 3ce225941fa4..efb6fad1f861 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,7 +18,7 @@
package org.apache.spark.sql.hive.execution.command
import org.apache.spark.SparkUnsupportedOperationException
-import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId
import org.apache.spark.sql.execution.command.v1
import org.apache.spark.sql.internal.SQLConf
@@ -72,4 +72,14 @@ class AlterTableAddColumnsSuite
}
}
}
+
+ test("SPARK-60565: Add columns to a table with collated CHAR and VARCHAR
columns") {
+ withNamespaceAndTable("ns", "tbl") { tbl =>
+ sql(s"CREATE TABLE $tbl (c CHAR(3) COLLATE UTF8_LCASE, v VARCHAR(3)
COLLATE UNICODE_CI) " +
+ defaultUsing)
+ sql(s"ALTER TABLE $tbl ADD COLUMNS (i INT)")
+ sql(s"INSERT INTO $tbl VALUES ('a', 'b', 1)")
+ checkAnswer(sql(s"SELECT * FROM $tbl WHERE c = 'A' AND v = 'B'"), Row("a
", "b", 1))
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
