You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/09/23 03:21:44 UTC
(spark) branch master updated: [SPARK-59276][SQL] Preserve CHAR/VARCHAR collations in Spark Connect
This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 8e30cf934937 [SPARK-59276][SQL] Preserve CHAR/VARCHAR collations in Spark Connect
8e30cf934937 is described below
commit 8e30cf9349378248d09ac0d618c783b84d07e1b2
Author: Serge Rielau <se...@rielau.com>
AuthorDate: Wed Sep 23 03:21:29 2026 +0000
[SPARK-59276][SQL] Preserve CHAR/VARCHAR collations in Spark Connect
### What changes were proposed in this pull request?
Preserve first-class CHAR/VARCHAR types and collations through Spark Connect schema conversion and row-based local relations.
- Support collated `CharType` and `VarcharType` in Python DDL, JSON, repr, and Connect type conversion.
- Store CHAR/VARCHAR collation restoration in `__CHAR_VARCHAR_COLLATIONS`, allowing preceding readers to parse the uncollated constrained type and retain the unknown metadata.
- Reject caller metadata that already owns the reserved restoration key instead of overwriting or dropping it.
- Encode Java and Python list/row local data with physical STRING leaves while carrying the requested logical schema separately for server-side policy reconciliation.
- Preserve the pre-existing UDT boundary; this PR does not add CHAR/VARCHAR support inside `UDT.sqlType`.
JIRA: https://issues.apache.org/jira/browse/SPARK-59276
### Why are the changes needed?
Spark Connect local-relation transport uses physical Arrow STRING values for CHAR/VARCHAR, but the server needs the logical constrained types and collations to apply standard, legacy-as-string, or rejection policy consistently. Schema conversion also needs a backward-compatible representation that does not cause preceding readers to reject collated CHAR/VARCHAR fields.
### Does this PR introduce _any_ user-facing change?
Yes. Spark Connect now preserves CHAR/VARCHAR types and collations through schema conversion and row-based `createDataFrame`, including populated, empty, and cached local relations.
Pandas, `pyarrow.Table`, CHAR/VARCHAR inside `UDT.sqlType`, and new `StringType` identity semantics are explicitly outside this PR.
### How was this patch tested?
Focused Scala tests passed:
- `StructTypeSuite`: 7 SPARK-59276 tests.
- `ArrowEncoderSuite`: 2 SPARK-59276 tests.
- `ClientE2ETestSuite`: 2 SPARK-59276 tests, including forced cached local-relation transport.
- `SparkConnectPlannerSuite`: 1 SPARK-59276 test.
Focused Python schema-codec tests passed:
- CHAR/VARCHAR type and collation round trips.
- Invalid restoration metadata, including metadata on `decimal(10,2)`.
- Reserved metadata-key collision rejection.
`dev/lint-python --ruff`, Python source compilation, and `git diff --check` passed. Python Connect and Arrow UDF execution tests were skipped because the local Python environment does not have the optional Connect, PyArrow, and Pandas dependencies.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.6
Closes #58581 from srielau/serge-rielau_data/SPARK-59276.
Authored-by: Serge Rielau <se...@rielau.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../src/main/resources/error/error-conditions.json | 23 ++
python/pyspark/errors/error-conditions.json | 22 ++
python/pyspark/sql/connect/proto/types_pb2.py | 52 +--
python/pyspark/sql/connect/proto/types_pb2.pyi | 48 ++-
python/pyspark/sql/connect/session.py | 51 ++-
python/pyspark/sql/connect/types.py | 14 +-
.../sql/tests/connect/test_connect_basic.py | 148 +++++++-
.../connect/test_connect_dataframe_creation.py | 42 ++-
.../pyspark/sql/tests/connect/test_connect_plan.py | 28 ++
python/pyspark/sql/tests/test_types.py | 397 ++++++++++++++++++++-
python/pyspark/sql/types.py | 310 ++++++++++++----
.../org/apache/spark/sql/types/CharType.scala | 4 +
.../org/apache/spark/sql/types/DataType.scala | 178 +++++++--
.../org/apache/spark/sql/types/StructField.scala | 70 +++-
.../org/apache/spark/sql/types/VarcharType.scala | 4 +
.../apache/spark/sql/types/StructTypeSuite.scala | 333 +++++++++++++++++
.../spark/sql/connect/ClientE2ETestSuite.scala | 88 +++++
.../connect/client/arrow/ArrowEncoderSuite.scala | 13 +-
.../src/main/protobuf/spark/connect/types.proto | 6 +
.../apache/spark/sql/connect/SparkSession.scala | 24 +-
.../connect/common/DataTypeProtoConverter.scala | 22 +-
.../sql/connect/planner/SparkConnectPlanner.scala | 89 +++--
.../LiteralExpressionProtoConverterSuite.scala | 33 ++
.../connect/planner/SparkConnectPlannerSuite.scala | 15 +-
24 files changed, 1819 insertions(+), 195 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 3585a10f29b9..5ef335276bc7 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -3875,6 +3875,29 @@
},
"sqlState" : "42K03"
},
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA" : {
+ "message" : [
+ "Invalid CHAR/VARCHAR collation metadata."
+ ],
+ "subClass" : {
+ "INVALID_VALUE" : {
+ "message" : [
+ "The value <value> must be a provider-qualified, non-empty collation name."
+ ]
+ },
+ "RESERVED_METADATA_KEY" : {
+ "message" : [
+ "The metadata key <metadataKey> is reserved and cannot be provided by callers."
+ ]
+ },
+ "UNRECOGNIZED_PATH" : {
+ "message" : [
+ "The field path <fieldPath> does not identify an uncollated CHAR/VARCHAR type."
+ ]
+ }
+ },
+ "sqlState" : "2203G"
+ },
"INVALID_CHECKPOINT_DIRECTORY" : {
"message" : [
"Cannot read the checkpoint directory <path>: expected the partition file <expectedFileName> but found <fileName>. The partition files must be numbered contiguously from part-00000."
diff --git a/python/pyspark/errors/error-conditions.json b/python/pyspark/errors/error-conditions.json
index 66d2ec3e37ac..d60a4c7e3c05 100644
--- a/python/pyspark/errors/error-conditions.json
+++ b/python/pyspark/errors/error-conditions.json
@@ -395,6 +395,28 @@
"Invalid call to `<func_name>` on unresolved object."
]
},
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA": {
+ "message": [
+ "Invalid CHAR/VARCHAR collation metadata."
+ ],
+ "sub_class": {
+ "INVALID_VALUE": {
+ "message": [
+ "The value <value> must be a provider-qualified, non-empty collation name."
+ ]
+ },
+ "RESERVED_METADATA_KEY": {
+ "message": [
+ "The metadata key <metadataKey> is reserved and cannot be provided by callers."
+ ]
+ },
+ "UNRECOGNIZED_PATH": {
+ "message": [
+ "The field path <fieldPath> does not identify an uncollated CHAR/VARCHAR type."
+ ]
+ }
+ }
+ },
"INVALID_CONNECT_URL": {
"message": [
"Invalid URL for Spark Connect: <detail>"
diff --git a/python/pyspark/sql/connect/proto/types_pb2.py b/python/pyspark/sql/connect/proto/types_pb2.py
index 074de0ae2805..2642b6f2da2b 100644
--- a/python/pyspark/sql/connect/proto/types_pb2.py
+++ b/python/pyspark/sql/connect/proto/types_pb2.py
@@ -36,7 +36,7 @@ _sym_db = _symbol_database.Default()
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(
- b"\n\x19spark/connect/types.proto\x12\rspark.connect\"\x92)\n\x08\x44\x61taType\x12\x32\n\x04null\x18\x01 \x01(\x0b\x32\x1c.spark.connect.DataType.NULLH\x00R\x04null\x12\x38\n\x06\x62inary\x18\x02 \x01(\x0b\x32\x1e.spark.connect.DataType.BinaryH\x00R\x06\x62inary\x12;\n\x07\x62oolean\x18\x03 \x01(\x0b\x32\x1f.spark.connect.DataType.BooleanH\x00R\x07\x62oolean\x12\x32\n\x04\x62yte\x18\x04 \x01(\x0b\x32\x1c.spark.connect.DataType.ByteH\x00R\x04\x62yte\x12\x35\n\x05short\x18\x05 \x01(\x [...]
+ b"\n\x19spark/connect/types.proto\x12\rspark.connect\"\xf6)\n\x08\x44\x61taType\x12\x32\n\x04null\x18\x01 \x01(\x0b\x32\x1c.spark.connect.DataType.NULLH\x00R\x04null\x12\x38\n\x06\x62inary\x18\x02 \x01(\x0b\x32\x1e.spark.connect.DataType.BinaryH\x00R\x06\x62inary\x12;\n\x07\x62oolean\x18\x03 \x01(\x0b\x32\x1f.spark.connect.DataType.BooleanH\x00R\x07\x62oolean\x12\x32\n\x04\x62yte\x18\x04 \x01(\x0b\x32\x1c.spark.connect.DataType.ByteH\x00R\x04\x62yte\x12\x35\n\x05short\x18\x05 \x01(\x [...]
)
_globals = globals()
@@ -48,7 +48,7 @@ if not _descriptor._USE_C_DESCRIPTORS:
"DESCRIPTOR"
]._serialized_options = b"\n\036org.apache.spark.connect.protoP\001Z\022internal/generated"
_globals["_DATATYPE"]._serialized_start = 45
- _globals["_DATATYPE"]._serialized_end = 5311
+ _globals["_DATATYPE"]._serialized_end = 5411
_globals["_DATATYPE_BOOLEAN"]._serialized_start = 1964
_globals["_DATATYPE_BOOLEAN"]._serialized_end = 2031
_globals["_DATATYPE_BYTE"]._serialized_start = 2033
@@ -87,28 +87,28 @@ if not _descriptor._USE_C_DESCRIPTORS:
_globals["_DATATYPE_YEARMONTHINTERVAL"]._serialized_end = 3508
_globals["_DATATYPE_DAYTIMEINTERVAL"]._serialized_start = 3511
_globals["_DATATYPE_DAYTIMEINTERVAL"]._serialized_end = 3688
- _globals["_DATATYPE_CHAR"]._serialized_start = 3690
- _globals["_DATATYPE_CHAR"]._serialized_end = 3778
- _globals["_DATATYPE_VARCHAR"]._serialized_start = 3780
- _globals["_DATATYPE_VARCHAR"]._serialized_end = 3871
- _globals["_DATATYPE_DECIMAL"]._serialized_start = 3874
- _globals["_DATATYPE_DECIMAL"]._serialized_end = 4027
- _globals["_DATATYPE_STRUCTFIELD"]._serialized_start = 4030
- _globals["_DATATYPE_STRUCTFIELD"]._serialized_end = 4191
- _globals["_DATATYPE_STRUCT"]._serialized_start = 4193
- _globals["_DATATYPE_STRUCT"]._serialized_end = 4320
- _globals["_DATATYPE_ARRAY"]._serialized_start = 4323
- _globals["_DATATYPE_ARRAY"]._serialized_end = 4485
- _globals["_DATATYPE_MAP"]._serialized_start = 4488
- _globals["_DATATYPE_MAP"]._serialized_end = 4707
- _globals["_DATATYPE_GEOMETRY"]._serialized_start = 4709
- _globals["_DATATYPE_GEOMETRY"]._serialized_end = 4797
- _globals["_DATATYPE_GEOGRAPHY"]._serialized_start = 4799
- _globals["_DATATYPE_GEOGRAPHY"]._serialized_end = 4888
- _globals["_DATATYPE_VARIANT"]._serialized_start = 4890
- _globals["_DATATYPE_VARIANT"]._serialized_end = 4957
- _globals["_DATATYPE_UDT"]._serialized_start = 4960
- _globals["_DATATYPE_UDT"]._serialized_end = 5249
- _globals["_DATATYPE_UNPARSED"]._serialized_start = 5251
- _globals["_DATATYPE_UNPARSED"]._serialized_end = 5303
+ _globals["_DATATYPE_CHAR"]._serialized_start = 3691
+ _globals["_DATATYPE_CHAR"]._serialized_end = 3828
+ _globals["_DATATYPE_VARCHAR"]._serialized_start = 3831
+ _globals["_DATATYPE_VARCHAR"]._serialized_end = 3971
+ _globals["_DATATYPE_DECIMAL"]._serialized_start = 3974
+ _globals["_DATATYPE_DECIMAL"]._serialized_end = 4127
+ _globals["_DATATYPE_STRUCTFIELD"]._serialized_start = 4130
+ _globals["_DATATYPE_STRUCTFIELD"]._serialized_end = 4291
+ _globals["_DATATYPE_STRUCT"]._serialized_start = 4293
+ _globals["_DATATYPE_STRUCT"]._serialized_end = 4420
+ _globals["_DATATYPE_ARRAY"]._serialized_start = 4423
+ _globals["_DATATYPE_ARRAY"]._serialized_end = 4585
+ _globals["_DATATYPE_MAP"]._serialized_start = 4588
+ _globals["_DATATYPE_MAP"]._serialized_end = 4807
+ _globals["_DATATYPE_GEOMETRY"]._serialized_start = 4809
+ _globals["_DATATYPE_GEOMETRY"]._serialized_end = 4897
+ _globals["_DATATYPE_GEOGRAPHY"]._serialized_start = 4899
+ _globals["_DATATYPE_GEOGRAPHY"]._serialized_end = 4988
+ _globals["_DATATYPE_VARIANT"]._serialized_start = 4990
+ _globals["_DATATYPE_VARIANT"]._serialized_end = 5057
+ _globals["_DATATYPE_UDT"]._serialized_start = 5060
+ _globals["_DATATYPE_UDT"]._serialized_end = 5349
+ _globals["_DATATYPE_UNPARSED"]._serialized_start = 5351
+ _globals["_DATATYPE_UNPARSED"]._serialized_end = 5403
# @@protoc_insertion_point(module_scope)
diff --git a/python/pyspark/sql/connect/proto/types_pb2.pyi b/python/pyspark/sql/connect/proto/types_pb2.pyi
index e80ff3399eb7..f1f0cd20d52a 100644
--- a/python/pyspark/sql/connect/proto/types_pb2.pyi
+++ b/python/pyspark/sql/connect/proto/types_pb2.pyi
@@ -518,40 +518,84 @@ class DataType(google.protobuf.message.Message):
LENGTH_FIELD_NUMBER: builtins.int
TYPE_VARIATION_REFERENCE_FIELD_NUMBER: builtins.int
+ COLLATION_FIELD_NUMBER: builtins.int
length: builtins.int
type_variation_reference: builtins.int
+ collation: builtins.str
+ """(Optional) Collation name. When omitted, the collation was not explicitly specified.
+ Older clients ignore this field and decode the historical uncollated CHAR type.
+ """
def __init__(
self,
*,
length: builtins.int = ...,
type_variation_reference: builtins.int = ...,
+ collation: builtins.str | None = ...,
) -> None: ...
+ def HasField(
+ self,
+ field_name: typing_extensions.Literal[
+ "_collation", b"_collation", "collation", b"collation"
+ ],
+ ) -> builtins.bool: ...
def ClearField(
self,
field_name: typing_extensions.Literal[
- "length", b"length", "type_variation_reference", b"type_variation_reference"
+ "_collation",
+ b"_collation",
+ "collation",
+ b"collation",
+ "length",
+ b"length",
+ "type_variation_reference",
+ b"type_variation_reference",
],
) -> None: ...
+ def WhichOneof(
+ self, oneof_group: typing_extensions.Literal["_collation", b"_collation"]
+ ) -> typing_extensions.Literal["collation"] | None: ...
class VarChar(google.protobuf.message.Message):
DESCRIPTOR: google.protobuf.descriptor.Descriptor
LENGTH_FIELD_NUMBER: builtins.int
TYPE_VARIATION_REFERENCE_FIELD_NUMBER: builtins.int
+ COLLATION_FIELD_NUMBER: builtins.int
length: builtins.int
type_variation_reference: builtins.int
+ collation: builtins.str
+ """(Optional) Collation name. When omitted, the collation was not explicitly specified.
+ Older clients ignore this field and decode the historical uncollated VARCHAR type.
+ """
def __init__(
self,
*,
length: builtins.int = ...,
type_variation_reference: builtins.int = ...,
+ collation: builtins.str | None = ...,
) -> None: ...
+ def HasField(
+ self,
+ field_name: typing_extensions.Literal[
+ "_collation", b"_collation", "collation", b"collation"
+ ],
+ ) -> builtins.bool: ...
def ClearField(
self,
field_name: typing_extensions.Literal[
- "length", b"length", "type_variation_reference", b"type_variation_reference"
+ "_collation",
+ b"_collation",
+ "collation",
+ b"collation",
+ "length",
+ b"length",
+ "type_variation_reference",
+ b"type_variation_reference",
],
) -> None: ...
+ def WhichOneof(
+ self, oneof_group: typing_extensions.Literal["_collation", b"_collation"]
+ ) -> typing_extensions.Literal["collation"] | None: ...
class Decimal(google.protobuf.message.Message):
DESCRIPTOR: google.protobuf.descriptor.Descriptor
diff --git a/python/pyspark/sql/connect/session.py b/python/pyspark/sql/connect/session.py
index bd6f32ab5ac0..ddfa8990a68c 100644
--- a/python/pyspark/sql/connect/session.py
+++ b/python/pyspark/sql/connect/session.py
@@ -89,20 +89,59 @@ from pyspark.sql.profiler import Profile
from pyspark.sql.session import SparkSession as PySparkSession
from pyspark.sql.session import classproperty
from pyspark.sql.types import (
+ ArrayType,
AtomicType,
+ CharType,
DataType,
DayTimeIntervalType,
MapType,
Row,
StringType,
+ StructField,
StructType,
TimestampType,
+ VarcharType,
_has_nulltype,
_infer_schema,
_merge_type,
)
from pyspark.sql.utils import to_str
+
+def _replace_char_varchar_with_collation_preserving_string(
+ data_type: DataType,
+) -> DataType:
+ """Lower CHAR/VARCHAR recursively for Arrow storage, preserving explicit collations."""
+ if isinstance(data_type, (CharType, VarcharType)):
+ if data_type.collation is None:
+ return StringType()
+ return StringType(data_type.collation)
+ if isinstance(data_type, ArrayType):
+ return ArrayType(
+ _replace_char_varchar_with_collation_preserving_string(data_type.elementType),
+ data_type.containsNull,
+ )
+ if isinstance(data_type, MapType):
+ return MapType(
+ _replace_char_varchar_with_collation_preserving_string(data_type.keyType),
+ _replace_char_varchar_with_collation_preserving_string(data_type.valueType),
+ data_type.valueContainsNull,
+ )
+ if isinstance(data_type, StructType):
+ return StructType(
+ [
+ StructField(
+ field.name,
+ _replace_char_varchar_with_collation_preserving_string(field.dataType),
+ field.nullable,
+ field.metadata,
+ )
+ for field in data_type.fields
+ ]
+ )
+ return data_type
+
+
if TYPE_CHECKING:
import pyspark.sql.connect.proto as pb2
from pyspark.sql.connect._typing import OptionalPrimitiveType
@@ -744,7 +783,13 @@ class SparkSession:
# Spark Connect will try its best to build the Arrow table with the
# inferred schema in the client side, and then rename the columns and
# cast the datatypes in the server side.
- _table = LocalDataToArrowConversion.convert(_data, _schema, prefers_large_types)
+ arrow_compatible_schema = cast(
+ StructType,
+ _replace_char_varchar_with_collation_preserving_string(_schema),
+ )
+ _table = LocalDataToArrowConversion.convert(
+ _data, arrow_compatible_schema, prefers_large_types
+ )
# TODO: Beside the validation on number of columns, we should also check
# whether the Arrow Schema is compatible with the user provided Schema.
@@ -776,7 +821,9 @@ class SparkSession:
configs["spark.sql.session.localRelationChunkSizeBytes"] # type: ignore[arg-type]
)
max_batch_of_chunks_size_bytes = int(
- configs["spark.sql.session.localRelationBatchOfChunksSizeBytes"] # type: ignore[arg-type]
+ configs[ # type: ignore[arg-type]
+ "spark.sql.session.localRelationBatchOfChunksSizeBytes"
+ ]
)
plan: LogicalPlan = local_relation
if cache_threshold <= _table.nbytes:
diff --git a/python/pyspark/sql/connect/types.py b/python/pyspark/sql/connect/types.py
index 3bc5aeb2cfe9..1f596bdaeeef 100644
--- a/python/pyspark/sql/connect/types.py
+++ b/python/pyspark/sql/connect/types.py
@@ -130,8 +130,12 @@ def pyspark_types_to_proto_types(data_type: DataType) -> pb2.DataType:
ret.null.CopyFrom(pb2.DataType.NULL())
elif isinstance(data_type, CharType):
ret.char.length = data_type.length
+ if data_type.collation is not None:
+ ret.char.collation = data_type.collation
elif isinstance(data_type, VarcharType):
ret.var_char.length = data_type.length
+ if data_type.collation is not None:
+ ret.var_char.collation = data_type.collation
elif isinstance(data_type, StringType):
ret.string.collation = data_type.collation
elif isinstance(data_type, BooleanType):
@@ -246,9 +250,15 @@ def proto_schema_to_pyspark_data_type(schema: pb2.DataType) -> DataType:
collation = schema.string.collation if schema.string.collation != "" else "UTF8_BINARY"
return StringType(collation)
elif schema.HasField("char"):
- return CharType(schema.char.length)
+ return CharType(
+ schema.char.length,
+ schema.char.collation if schema.char.HasField("collation") else None,
+ )
elif schema.HasField("var_char"):
- return VarcharType(schema.var_char.length)
+ return VarcharType(
+ schema.var_char.length,
+ schema.var_char.collation if schema.var_char.HasField("collation") else None,
+ )
elif schema.HasField("date"):
return DateType()
elif schema.HasField("time"):
diff --git a/python/pyspark/sql/tests/connect/test_connect_basic.py b/python/pyspark/sql/tests/connect/test_connect_basic.py
index 6a28f101b9b3..b18c338ec6e2 100755
--- a/python/pyspark/sql/tests/connect/test_connect_basic.py
+++ b/python/pyspark/sql/tests/connect/test_connect_basic.py
@@ -24,7 +24,7 @@ import tempfile
import unittest
from contextlib import redirect_stdout
-from pyspark.errors import PySparkTypeError, PySparkValueError
+from pyspark.errors import AnalysisException, PySparkTypeError, PySparkValueError
from pyspark.sql.types import (
ArrayType,
CharType,
@@ -491,8 +491,14 @@ class SparkConnectBasicTests(SparkConnectSQLTestCase):
self._check_print_schema(query)
def test_char_varchar_result_schema(self):
- # SPARK-58794: Python Connect maps first-class CHAR/VARCHAR the same as classic.
- query = "SELECT CAST('ab' AS CHAR(4)) AS c, CAST('cd' AS VARCHAR(6)) AS v"
+ # SPARK-59276: Python Connect maps first-class CHAR/VARCHAR the same as classic.
+ query = """
+ SELECT
+ CAST('ab' AS CHAR(4)) AS c,
+ CAST('cd' AS VARCHAR(6)) AS v,
+ CAST('ef' AS CHAR(4) COLLATE UTF8_LCASE) AS collated_c,
+ CAST('gh' AS VARCHAR(6) COLLATE UNICODE_CI) AS collated_v
+ """
conf = {"spark.sql.charVarchar.standardSemantics.enabled": "true"}
with self.both_conf(conf):
classic_df = self.spark.sql(query)
@@ -500,8 +506,142 @@ class SparkConnectBasicTests(SparkConnectSQLTestCase):
self.assertEqual(classic_df.schema, connect_df.schema)
self.assertEqual(classic_df.schema["c"].dataType, CharType(4))
self.assertEqual(classic_df.schema["v"].dataType, VarcharType(6))
+ self.assertEqual(classic_df.schema["collated_c"].dataType, CharType(4, "UTF8_LCASE"))
+ self.assertEqual(classic_df.schema["collated_v"].dataType, VarcharType(6, "UNICODE_CI"))
self.assertEqual(classic_df.collect(), connect_df.collect())
- self.assertEqual(connect_df.collect(), [Row(c="ab ", v="cd")])
+ self.assertEqual(
+ connect_df.collect(),
+ [Row(c="ab ", v="cd", collated_c="ef ", collated_v="gh")],
+ )
+
+ def test_create_dataframe_with_char_varchar_schema(self):
+ schema = StructType(
+ [
+ StructField("c", CharType(4)),
+ StructField("explicit_c", CharType(4, "UTF8_BINARY")),
+ StructField("v", VarcharType(3, "UTF8_LCASE")),
+ StructField(
+ "nested",
+ StructType(
+ [
+ StructField("c", CharType(3, "UTF8_BINARY")),
+ StructField(
+ "values",
+ ArrayType(VarcharType(4, "UNICODE_CI"), containsNull=False),
+ ),
+ StructField(
+ "lookup",
+ MapType(
+ CharType(3, "UTF8_LCASE"),
+ VarcharType(4, "UTF8_BINARY"),
+ valueContainsNull=False,
+ ),
+ ),
+ ]
+ ),
+ ),
+ ]
+ )
+ rows = [
+ (
+ "ab",
+ "cd",
+ "ef",
+ Row(c="x", values=["gh", "ij"], lookup={"k": "lm"}),
+ )
+ ]
+ standard_conf = {
+ "spark.sql.charVarchar.standardSemantics.enabled": "true",
+ "spark.sql.legacy.charVarcharAsString": "false",
+ }
+ with self.both_conf(standard_conf):
+ df = self.connect.createDataFrame(rows, schema)
+ self.assertEqual(df.schema, schema)
+ self.assertEqual(
+ df.collect(),
+ [
+ Row(
+ c="ab ",
+ explicit_c="cd ",
+ v="ef",
+ nested=Row(
+ c="x ",
+ values=["gh", "ij"],
+ lookup={"k ": "lm"},
+ ),
+ )
+ ],
+ )
+ empty = self.connect.createDataFrame([], schema)
+ self.assertEqual(empty.schema, schema)
+ self.assertEqual(empty.collect(), [])
+
+ legacy_conf = {
+ "spark.sql.charVarchar.standardSemantics.enabled": "false",
+ "spark.sql.legacy.charVarcharAsString": "true",
+ }
+ with self.both_conf(legacy_conf):
+ expected = StructType(
+ [
+ StructField("c", StringType()),
+ StructField("explicit_c", StringType("UTF8_BINARY")),
+ StructField("v", StringType("UTF8_LCASE")),
+ StructField(
+ "nested",
+ StructType(
+ [
+ StructField("c", StringType("UTF8_BINARY")),
+ StructField(
+ "values",
+ ArrayType(StringType("UNICODE_CI"), containsNull=False),
+ ),
+ StructField(
+ "lookup",
+ MapType(
+ StringType("UTF8_LCASE"),
+ StringType("UTF8_BINARY"),
+ valueContainsNull=False,
+ ),
+ ),
+ ]
+ ),
+ ),
+ ]
+ )
+ df = self.connect.createDataFrame(rows, schema)
+ self.assertEqual(df.schema, expected)
+ self.assertEqual(
+ df.collect(),
+ [
+ Row(
+ c="ab",
+ explicit_c="cd",
+ v="ef",
+ nested=Row(
+ c="x",
+ values=["gh", "ij"],
+ lookup={"k": "lm"},
+ ),
+ )
+ ],
+ )
+ empty = self.connect.createDataFrame([], schema)
+ self.assertEqual(empty.schema, expected)
+ self.assertEqual(empty.collect(), [])
+
+ default_conf = {
+ "spark.sql.charVarchar.standardSemantics.enabled": "false",
+ "spark.sql.legacy.charVarcharAsString": "false",
+ }
+ with self.both_conf(default_conf):
+ for data in (rows, []):
+ with self.assertRaises(AnalysisException) as ctx:
+ self.connect.createDataFrame(data, schema).schema
+ self.check_error(
+ exception=ctx.exception,
+ errorClass="UNSUPPORTED_CHAR_OR_VARCHAR_AS_STRING",
+ messageParameters={},
+ )
def test_to(self):
# SPARK-41464: test DataFrame.to()
diff --git a/python/pyspark/sql/tests/connect/test_connect_dataframe_creation.py b/python/pyspark/sql/tests/connect/test_connect_dataframe_creation.py
index 6660fc8935fb..a38aeab5b43b 100644
--- a/python/pyspark/sql/tests/connect/test_connect_dataframe_creation.py
+++ b/python/pyspark/sql/tests/connect/test_connect_dataframe_creation.py
@@ -32,7 +32,7 @@ from pyspark.sql.types import (
StructType,
)
from pyspark.testing.connectutils import ReusedMixedTestCase, should_test_connect
-from pyspark.testing.objects import MyObject, PythonOnlyUDT
+from pyspark.testing.objects import MyObject, PythonOnlyPoint, PythonOnlyUDT
from pyspark.testing.pandasutils import PandasOnSparkTestUtils
if should_test_connect:
@@ -733,6 +733,46 @@ class SparkConnectDataFrameCreationTests(ReusedMixedTestCase, PandasOnSparkTestU
self.assertEqual(cdf.schema, sdf.schema)
+ def test_python_udt_with_local_data(self):
+ direct_schema = StructType().add("point", PythonOnlyUDT(), nullable=False)
+ nested_schema = StructType().add(
+ "nested",
+ StructType(
+ [
+ StructField(
+ "points", ArrayType(PythonOnlyUDT(), containsNull=False), nullable=False
+ ),
+ StructField(
+ "points_by_id",
+ MapType(IntegerType(), PythonOnlyUDT(), valueContainsNull=False),
+ nullable=False,
+ ),
+ ]
+ ),
+ nullable=False,
+ )
+ test_cases = [
+ (direct_schema, [(PythonOnlyPoint(1.0, 2.0),)]),
+ (
+ nested_schema,
+ [
+ (
+ Row(
+ points=[PythonOnlyPoint(1.0, 2.0)],
+ points_by_id={3: PythonOnlyPoint(4.0, 5.0)},
+ ),
+ )
+ ],
+ ),
+ ]
+
+ for schema, data in test_cases:
+ with self.subTest(schema=schema):
+ cdf = self.connect.createDataFrame(data, schema)
+ sdf = self.spark.createDataFrame(data, schema)
+ self.assertEqual(cdf.schema, sdf.schema)
+ self.assertEqual(cdf.collect(), sdf.collect())
+
if __name__ == "__main__":
from pyspark.testing import main
diff --git a/python/pyspark/sql/tests/connect/test_connect_plan.py b/python/pyspark/sql/tests/connect/test_connect_plan.py
index 6069b703b72f..6ee4a8993dcb 100644
--- a/python/pyspark/sql/tests/connect/test_connect_plan.py
+++ b/python/pyspark/sql/tests/connect/test_connect_plan.py
@@ -60,6 +60,7 @@ if should_test_connect:
)
from pyspark.sql.types import (
ArrayType,
+ CharType,
DoubleType,
IntegerType,
MapType,
@@ -68,6 +69,7 @@ if should_test_connect:
StructType,
TimestampLTZNanosType,
TimestampNTZNanosType,
+ VarcharType,
)
@@ -76,6 +78,32 @@ class SparkConnectPlanTests(PlanOnlyTestFixture):
"""These test cases exercise the interface to the proto plan
generation but do not call Spark."""
+ def test_char_varchar_collation_type_round_trip(self):
+ data_types = [
+ CharType(4),
+ CharType(4, "UTF8_BINARY"),
+ CharType(4, "UTF8_LCASE"),
+ VarcharType(6),
+ VarcharType(6, "UTF8_BINARY"),
+ VarcharType(6, "UNICODE_CI"),
+ StructType(
+ [
+ StructField("c", CharType(4, "UTF8_LCASE")),
+ StructField("v", ArrayType(VarcharType(6, "UNICODE_CI"))),
+ ]
+ ),
+ ]
+
+ for data_type in data_types:
+ proto_type = pyspark_types_to_proto_types(data_type)
+ self.assertEqual(data_type, proto_schema_to_pyspark_data_type(proto_type))
+
+ default_char = pyspark_types_to_proto_types(CharType(4)).char
+ self.assertFalse(default_char.HasField("collation"))
+ explicit_binary_char = pyspark_types_to_proto_types(CharType(4, "UTF8_BINARY")).char
+ self.assertTrue(explicit_binary_char.HasField("collation"))
+ self.assertEqual(explicit_binary_char.collation, "UTF8_BINARY")
+
def test_sql_project(self):
plan = self.connect.sql("SELECT 1")._plan.to_proto(self.connect)
self.assertEqual(plan.root.sql.query, "SELECT 1")
diff --git a/python/pyspark/sql/tests/test_types.py b/python/pyspark/sql/tests/test_types.py
index 0ca38327c9f3..04ca5211247d 100644
--- a/python/pyspark/sql/tests/test_types.py
+++ b/python/pyspark/sql/tests/test_types.py
@@ -18,8 +18,10 @@
import array
import ctypes
import datetime
+import json
import os
import pickle
+import re
import sys
import unittest
from dataclasses import asdict, dataclass
@@ -657,6 +659,150 @@ class TypesTestsMixin:
self.assertEqual(StringType("UTF8_LCASE").simpleString(), "string collate UTF8_LCASE")
self.assertEqual(StringType("UNICODE").simpleString(), "string collate UNICODE")
+ def test_char_varchar_type_collations(self):
+ from pyspark.sql.types import _parse_datatype_json_string
+
+ self.assertEqual(CharType(4).simpleString(), "char(4)")
+ self.assertEqual(CharType(4, "UTF8_LCASE").simpleString(), "char(4) collate UTF8_LCASE")
+ self.assertEqual(VarcharType(6).simpleString(), "varchar(6)")
+ self.assertEqual(
+ VarcharType(6, "UNICODE_CI").simpleString(), "varchar(6) collate UNICODE_CI"
+ )
+
+ data_types = [
+ CharType(4),
+ CharType(4, "UTF8_BINARY"),
+ CharType(4, "UTF8_LCASE"),
+ VarcharType(6),
+ VarcharType(6, "UTF8_BINARY"),
+ VarcharType(6, "UNICODE_CI"),
+ StructType(
+ [
+ StructField("c", CharType(4, "UTF8_LCASE")),
+ StructField("v", ArrayType(VarcharType(6, "UNICODE_CI"))),
+ ]
+ ),
+ StructType(
+ [
+ StructField(
+ "mixed",
+ MapType(
+ CharType(4, "UTF8_BINARY"),
+ VarcharType(6, "UNICODE_CI"),
+ ),
+ )
+ ]
+ ),
+ StructType([StructField("", ArrayType(CharType(4, "UTF8_LCASE")))]),
+ StructType(
+ [
+ StructField(
+ "",
+ MapType(
+ CharType(4, "UTF8_BINARY"),
+ VarcharType(6, "UNICODE_CI"),
+ ),
+ )
+ ]
+ ),
+ ]
+ for data_type in data_types:
+ self.assertEqual(data_type, _parse_datatype_json_string(data_type.json()))
+
+ compatibility_schema = StructType(
+ [
+ StructField("plain", CharType(3)),
+ StructField("binary", CharType(4, "UTF8_BINARY")),
+ StructField(
+ "nested",
+ ArrayType(VarcharType(6, "UNICODE_CI")),
+ ),
+ ]
+ )
+ self.assertEqual(
+ compatibility_schema.jsonValue(),
+ {
+ "type": "struct",
+ "fields": [
+ {
+ "name": "plain",
+ "type": "char(3)",
+ "nullable": True,
+ "metadata": {},
+ },
+ {
+ "name": "binary",
+ "type": "char(4)",
+ "nullable": True,
+ "metadata": {
+ "__CHAR_VARCHAR_COLLATIONS": {"binary": "spark.UTF8_BINARY"},
+ },
+ },
+ {
+ "name": "nested",
+ "type": {
+ "type": "array",
+ "elementType": "varchar(6)",
+ "containsNull": True,
+ },
+ "nullable": True,
+ "metadata": {
+ "__CHAR_VARCHAR_COLLATIONS": {"nested.element": "icu.UNICODE_CI"},
+ },
+ },
+ ],
+ },
+ )
+
+ mixed_field = StructField(
+ "mixed",
+ MapType(StringType("UTF8_LCASE"), CharType(4, "UNICODE_CI")),
+ )
+ self.assertEqual(
+ mixed_field.getCollationMetadata(),
+ {"mixed.key": "spark.UTF8_LCASE"},
+ )
+ self.assertEqual(
+ mixed_field.getCharVarcharCollationMetadata(),
+ {"mixed.value": "icu.UNICODE_CI"},
+ )
+ mixed_metadata = mixed_field.jsonValue()["metadata"]
+ self.assertEqual(
+ mixed_field.getCollationsMap(mixed_metadata),
+ {"mixed.key": "UTF8_LCASE"},
+ )
+ self.assertEqual(
+ mixed_field.getCharVarcharCollationsMap(mixed_metadata),
+ {"mixed.value": "UNICODE_CI"},
+ )
+
+ def test_standalone_collated_char_varchar_json_is_current_reader_only(self):
+ from pyspark.sql.types import _parse_datatype_json_string
+
+ standalone = CharType(4, "UTF8_LCASE")
+ self.assertEqual(standalone.jsonValue(), "char(4) collate UTF8_LCASE")
+ self.assertEqual(_parse_datatype_json_string(standalone.json()), standalone)
+
+ field_json = StructType([StructField("c", standalone)]).jsonValue()["fields"][0]
+ self.assertEqual(field_json["type"], "char(4)")
+ self.assertEqual(
+ field_json["metadata"]["__CHAR_VARCHAR_COLLATIONS"],
+ {"c": "spark.UTF8_LCASE"},
+ )
+
+ def test_char_varchar_json_rejects_trailing_tokens(self):
+ from pyspark.sql.types import _parse_datatype_json_string
+
+ for data_type in [
+ "char(4) collate UTF8_LCASE junk",
+ "varchar(6) collate UNICODE_CI junk",
+ ]:
+ with self.subTest(data_type=data_type):
+ self.assertRaises(
+ PySparkValueError,
+ lambda: _parse_datatype_json_string(json.dumps(data_type)),
+ )
+
def test_schema_with_collations_json_ser_de(self):
from pyspark.sql.types import _parse_datatype_json_string
@@ -763,6 +909,231 @@ class TypesTestsMixin:
assert schema == python_datatype
assert schema == _parse_datatype_json_string(schema.json())
+ def test_schema_rejects_inline_and_metadata_collations(self):
+ from pyspark.sql.types import (
+ _CHAR_VARCHAR_COLLATIONS_METADATA_KEY,
+ _COLLATIONS_METADATA_KEY,
+ _parse_datatype_json_string,
+ )
+
+ for data_type, metadata_collation in [
+ ("char(4) collate UTF8_LCASE", "spark.UTF8_LCASE"),
+ ("varchar(6) collate UTF8_LCASE", "icu.UNICODE_CI"),
+ ]:
+ schema_json = {
+ "type": "struct",
+ "fields": [
+ {
+ "name": "c",
+ "type": data_type,
+ "nullable": True,
+ "metadata": {
+ _CHAR_VARCHAR_COLLATIONS_METADATA_KEY: {"c": metadata_collation},
+ },
+ }
+ ],
+ }
+ self.assertRaises(
+ PySparkTypeError,
+ lambda: _parse_datatype_json_string(json.dumps(schema_json)),
+ )
+
+ collations_on_char_json = {
+ "type": "struct",
+ "fields": [
+ {
+ "name": "c",
+ "type": "char(4)",
+ "nullable": True,
+ "metadata": {
+ _COLLATIONS_METADATA_KEY: {"c": "spark.UTF8_LCASE"},
+ },
+ }
+ ],
+ }
+ self.assertRaises(
+ PySparkTypeError,
+ lambda: _parse_datatype_json_string(json.dumps(collations_on_char_json)),
+ )
+
+ char_varchar_collations_on_decimal_json = {
+ "type": "struct",
+ "fields": [
+ {
+ "name": "d",
+ "type": "decimal(10,2)",
+ "nullable": True,
+ "metadata": {
+ _CHAR_VARCHAR_COLLATIONS_METADATA_KEY: {"d": "spark.UTF8_LCASE"},
+ },
+ }
+ ],
+ }
+ self.assertRaises(
+ PySparkTypeError,
+ lambda: _parse_datatype_json_string(
+ json.dumps(char_varchar_collations_on_decimal_json)
+ ),
+ )
+
+ def test_char_varchar_collation_metadata_key_collision_is_rejected(self):
+ from pyspark.sql.types import _CHAR_VARCHAR_COLLATIONS_METADATA_KEY
+
+ field = StructField(
+ "c",
+ CharType(4, "UTF8_LCASE"),
+ metadata={_CHAR_VARCHAR_COLLATIONS_METADATA_KEY: {"caller": "value"}},
+ )
+ with self.assertRaises(PySparkTypeError) as pe:
+ field.jsonValue()
+ self.check_error(
+ exception=pe.exception,
+ errorClass="INVALID_CHAR_VARCHAR_COLLATION_METADATA.RESERVED_METADATA_KEY",
+ messageParameters={"metadataKey": _CHAR_VARCHAR_COLLATIONS_METADATA_KEY},
+ )
+
+ def test_schema_rejects_malformed_char_varchar_collation_metadata(self):
+ from pyspark.sql.types import (
+ _CHAR_VARCHAR_COLLATIONS_METADATA_KEY,
+ _parse_datatype_json_string,
+ )
+
+ def schema_json(metadata_value):
+ return json.dumps(
+ {
+ "type": "struct",
+ "fields": [
+ {
+ "name": "c",
+ "type": "char(4)",
+ "nullable": True,
+ "metadata": {
+ _CHAR_VARCHAR_COLLATIONS_METADATA_KEY: metadata_value,
+ },
+ }
+ ],
+ }
+ )
+
+ cases = [
+ (
+ "caller",
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ {"value": '"caller"'},
+ ),
+ (
+ {"c": 1},
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ {"value": "1"},
+ ),
+ (
+ {"c": "spark."},
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ {"value": '"spark."'},
+ ),
+ (
+ {"c": ".UTF8_LCASE"},
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ {"value": '".UTF8_LCASE"'},
+ ),
+ (
+ {"c": "spark.UTF8_LCASE", "typo": "spark.UTF8_LCASE"},
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.UNRECOGNIZED_PATH",
+ {"fieldPath": "typo"},
+ ),
+ ]
+ for metadata_value, error_class, message_parameters in cases:
+ with self.assertRaises(PySparkTypeError) as pe:
+ _parse_datatype_json_string(schema_json(metadata_value))
+ self.check_error(
+ exception=pe.exception,
+ errorClass=error_class,
+ messageParameters=message_parameters,
+ )
+
+ def test_preceding_reader_retains_unknown_char_varchar_collation_metadata(self):
+ # Reduced independent reader copied from the relevant parser branches at base
+ # 389de941f002a5c92e22dc3ed0f65af602a174db. It intentionally does not use any
+ # current parsing helper, so unknown metadata exercises the preceding-reader contract.
+ def preceding_read_type(value):
+ if isinstance(value, str):
+ if match := re.fullmatch(r"char\((\d+)\)", value):
+ return CharType(int(match.group(1)))
+ if match := re.fullmatch(r"varchar\((\d+)\)", value):
+ return VarcharType(int(match.group(1)))
+ raise AssertionError(f"unsupported pinned-reader type: {value}")
+
+ if value["type"] == "array":
+ return ArrayType(preceding_read_type(value["elementType"]), value["containsNull"])
+ if value["type"] == "map":
+ return MapType(
+ preceding_read_type(value["keyType"]),
+ preceding_read_type(value["valueType"]),
+ value["valueContainsNull"],
+ )
+ if value["type"] == "struct":
+ return StructType([preceding_read_field(field) for field in value["fields"]])
+ raise AssertionError(f"unsupported pinned-reader type: {value}")
+
+ def preceding_read_field(value):
+ return StructField(
+ value["name"],
+ preceding_read_type(value["type"]),
+ value.get("nullable", True),
+ value.get("metadata"),
+ )
+
+ schema = StructType(
+ [
+ StructField("c", CharType(4, "UTF8_LCASE")),
+ StructField("nested", ArrayType(VarcharType(6, "UNICODE_CI"))),
+ StructField(
+ "mapped",
+ MapType(
+ CharType(3, "UTF8_BINARY"),
+ VarcharType(5, "UTF8_LCASE"),
+ ),
+ ),
+ StructField(
+ "nestedStruct",
+ StructType([StructField("c", CharType(2, "UNICODE_CI"))]),
+ ),
+ ]
+ )
+ preceding_schema = preceding_read_type(json.loads(schema.json()))
+ self.assertEqual(preceding_schema["c"].dataType, CharType(4))
+ self.assertEqual(
+ preceding_schema["nested"].dataType,
+ ArrayType(VarcharType(6)),
+ )
+ self.assertEqual(
+ preceding_schema["mapped"].dataType,
+ MapType(CharType(3), VarcharType(5)),
+ )
+ nested_struct = preceding_schema["nestedStruct"].dataType
+ self.assertEqual(nested_struct["c"].dataType, CharType(2))
+
+ metadata_key = "__CHAR_VARCHAR_COLLATIONS"
+ self.assertEqual(
+ preceding_schema["c"].metadata[metadata_key],
+ {"c": "spark.UTF8_LCASE"},
+ )
+ self.assertEqual(
+ preceding_schema["nested"].metadata[metadata_key],
+ {"nested.element": "icu.UNICODE_CI"},
+ )
+ self.assertEqual(
+ preceding_schema["mapped"].metadata[metadata_key],
+ {
+ "mapped.key": "spark.UTF8_BINARY",
+ "mapped.value": "spark.UTF8_LCASE",
+ },
+ )
+ self.assertEqual(
+ nested_struct["c"].metadata[metadata_key],
+ {"c": "icu.UNICODE_CI"},
+ )
+
def test_schema_with_collations_on_non_string_types(self):
from pyspark.sql.types import _COLLATIONS_METADATA_KEY, _parse_datatype_json_string
@@ -943,8 +1314,8 @@ class TypesTestsMixin:
("geography(SRID:ANY)", GeographyType("ANY")),
("geography(srid:any)", GeographyType("ANY")),
]
- for json, expected in valid_test_cases:
- python_datatype = _parse_datatype_json_value(json)
+ for json_value, expected in valid_test_cases:
+ python_datatype = _parse_datatype_json_value(json_value)
self.assertEqual(python_datatype, expected)
self.assertEqual(expected, _parse_datatype_json_string(expected.json()))
@@ -976,9 +1347,9 @@ class TypesTestsMixin:
"geography(SRID0)",
"geography(SRID:4326, ALG)",
]
- for json in invalid_test_cases:
+ for json_value in invalid_test_cases:
with self.assertRaises(Exception):
- _parse_datatype_json_value(json)
+ _parse_datatype_json_value(json_value)
def test_geometry_json_serde(self):
from pyspark.sql.types import _parse_datatype_json_string, _parse_datatype_json_value
@@ -989,8 +1360,8 @@ class TypesTestsMixin:
("geometry(SRID:ANY)", GeometryType("ANY")),
("geometry(srid:any)", GeometryType("ANY")),
]
- for json, expected in valid_test_cases:
- python_datatype = _parse_datatype_json_value(json)
+ for json_value, expected in valid_test_cases:
+ python_datatype = _parse_datatype_json_value(json_value)
self.assertEqual(python_datatype, expected)
self.assertEqual(expected, _parse_datatype_json_string(expected.json()))
@@ -1020,9 +1391,9 @@ class TypesTestsMixin:
"geometry(SRID:0,)",
"geometry(SRID0)",
]
- for json in invalid_test_cases:
+ for json_value in invalid_test_cases:
with self.assertRaises(Exception):
- _parse_datatype_json_value(json)
+ _parse_datatype_json_value(json_value)
def test_udt(self):
from pyspark.sql.types import _infer_type, _make_type_verifier, _parse_datatype_json_string
@@ -2068,7 +2439,11 @@ class TypesTestsMixin:
StringType("UNICODE"),
StringType("UNICODE_CI"),
CharType(10),
+ CharType(10, "UTF8_BINARY"),
+ CharType(10, "UTF8_LCASE"),
VarcharType(10),
+ VarcharType(10, "UTF8_BINARY"),
+ VarcharType(10, "UNICODE_CI"),
BinaryType(),
BooleanType(),
DateType(),
@@ -2744,9 +3119,9 @@ class TypesTestsMixin:
)
# check parse_json
- for key, json, obj in expected_values:
- self.assertEqual(VariantVal.parseJson(json).toJson(), json)
- self.assertEqual(VariantVal.parseJson(json).toPython(), obj)
+ for key, json_value, obj in expected_values:
+ self.assertEqual(VariantVal.parseJson(json_value).toJson(), json_value)
+ self.assertEqual(VariantVal.parseJson(json_value).toPython(), obj)
# compare the parse_json in Spark vs python. `json_str` contains all of `expected_values`.
parse_json_spark_output = variants[0]
diff --git a/python/pyspark/sql/types.py b/python/pyspark/sql/types.py
index e381d4abd7a8..cabbe4e2b87b 100644
--- a/python/pyspark/sql/types.py
+++ b/python/pyspark/sql/types.py
@@ -39,6 +39,7 @@ from typing import (
Iterator,
List,
Optional,
+ Set,
Tuple,
Type,
TypeVar,
@@ -324,47 +325,84 @@ class StringType(AtomicType):
class CharType(AtomicType):
- """Char data type
+ """Char data type.
+
+ A standalone collated ``CharType`` writes its collation inline in JSON. An older Python
+ reader may accept only the ``char(n)`` prefix and silently drop the collation; a current reader
+ is required to preserve it. Within a ``StructField``, schema JSON stores the collation in field
+ metadata and emits an uncollated ``char(n)`` type so older readers can still read it.
Parameters
----------
length : int
the length limitation.
+ collation : str, optional
+ name of the collation. Defaults to ``None``, which is distinct from explicitly specifying
+ ``UTF8_BINARY``.
"""
- def __init__(self, length: int):
+ def __init__(self, length: int, collation: Optional[str] = None):
self.length = length
+ self.collation = collation
def simpleString(self) -> str:
- return "char(%d)" % (self.length)
+ if self.collation is None:
+ return "char(%d)" % (self.length)
+
+ return "char(%d) collate %s" % (self.length, self.collation)
def jsonValue(self) -> str:
- return "char(%d)" % (self.length)
+ return self.simpleString()
def __repr__(self) -> str:
- return "CharType(%d)" % (self.length)
+ if self.collation is None:
+ return "CharType(%d)" % (self.length)
+
+ return "CharType(%d, '%s')" % (self.length, self.collation)
+
+ def isUTF8BinaryCollation(self) -> bool:
+ return self.collation is None or self.collation == "UTF8_BINARY"
class VarcharType(AtomicType):
- """Varchar data type
+ """Varchar data type.
+
+ A standalone collated ``VarcharType`` writes its collation inline in JSON. An older Python
+ reader may accept only the ``varchar(n)`` prefix and silently drop the collation; a current
+ reader is required to preserve it. Within a ``StructField``, schema JSON stores the collation
+ in field metadata and emits an uncollated ``varchar(n)`` type so older readers can still
+ read it.
Parameters
----------
length : int
the length limitation.
+ collation : str, optional
+ name of the collation. Defaults to ``None``, which is distinct from explicitly specifying
+ ``UTF8_BINARY``.
"""
- def __init__(self, length: int):
+ def __init__(self, length: int, collation: Optional[str] = None):
self.length = length
+ self.collation = collation
def simpleString(self) -> str:
- return "varchar(%d)" % (self.length)
+ if self.collation is None:
+ return "varchar(%d)" % (self.length)
+
+ return "varchar(%d) collate %s" % (self.length, self.collation)
def jsonValue(self) -> str:
- return "varchar(%d)" % (self.length)
+ return self.simpleString()
def __repr__(self) -> str:
- return "VarcharType(%d)" % (self.length)
+ if self.collation is None:
+ return "VarcharType(%d)" % (self.length)
+
+ return "VarcharType(%d, '%s')" % (self.length, self.collation)
+
+ def isUTF8BinaryCollation(self) -> bool:
+ return self.collation is None or self.collation == "UTF8_BINARY"
class BinaryType(AtomicType, metaclass=DataTypeSingleton):
@@ -1175,11 +1213,15 @@ class ArrayType(DataType):
json: Dict[str, Any],
fieldPath: str = "",
collationsMap: Optional[Dict[str, str]] = None,
+ charVarcharCollationsMap: Optional[Dict[str, str]] = None,
+ remainingCharVarcharPaths: Optional[Set[str]] = None,
) -> "ArrayType":
elementType = _parse_datatype_json_value(
json["elementType"],
- "element" if fieldPath == "" else fieldPath + ".element",
+ _append_field_to_path(fieldPath, "element"),
collationsMap,
+ charVarcharCollationsMap,
+ remainingCharVarcharPaths,
)
return ArrayType(elementType, json["containsNull"])
@@ -1317,12 +1359,22 @@ class MapType(DataType):
json: Dict[str, Any],
fieldPath: str = "",
collationsMap: Optional[Dict[str, str]] = None,
+ charVarcharCollationsMap: Optional[Dict[str, str]] = None,
+ remainingCharVarcharPaths: Optional[Set[str]] = None,
) -> "MapType":
keyType = _parse_datatype_json_value(
- json["keyType"], "key" if fieldPath == "" else fieldPath + ".key", collationsMap
+ json["keyType"],
+ _append_field_to_path(fieldPath, "key"),
+ collationsMap,
+ charVarcharCollationsMap,
+ remainingCharVarcharPaths,
)
valueType = _parse_datatype_json_value(
- json["valueType"], "value" if fieldPath == "" else fieldPath + ".value", collationsMap
+ json["valueType"],
+ _append_field_to_path(fieldPath, "value"),
+ collationsMap,
+ charVarcharCollationsMap,
+ remainingCharVarcharPaths,
)
return MapType(
keyType,
@@ -1416,22 +1468,32 @@ class StructField(DataType):
return "StructField('%s', %s, %s)" % (self.name, self.dataType, str(self.nullable))
def jsonValue(self) -> Dict[str, Any]:
- collationMetadata = self.getCollationMetadata()
- metadata = (
- self.metadata
- if not collationMetadata
- else {**self.metadata, _COLLATIONS_METADATA_KEY: collationMetadata}
- )
+ if _CHAR_VARCHAR_COLLATIONS_METADATA_KEY in self.metadata:
+ raise PySparkTypeError(
+ errorClass="INVALID_CHAR_VARCHAR_COLLATION_METADATA.RESERVED_METADATA_KEY",
+ messageParameters={"metadataKey": _CHAR_VARCHAR_COLLATIONS_METADATA_KEY},
+ )
+ string_metadata = self.getCollationMetadata()
+ char_varchar_metadata = self.getCharVarcharCollationMetadata()
+ metadata = dict(self.metadata)
+ if string_metadata:
+ metadata[_COLLATIONS_METADATA_KEY] = string_metadata
+ if char_varchar_metadata:
+ metadata[_CHAR_VARCHAR_COLLATIONS_METADATA_KEY] = char_varchar_metadata
return {
"name": self.name,
- "type": self._dataTypeJsonValue(collationMetadata),
+ "type": self._dataTypeJsonValue(string_metadata, char_varchar_metadata),
"nullable": self.nullable,
"metadata": metadata,
}
- def _dataTypeJsonValue(self, collationMetadata: Dict[str, str]) -> Union[str, Dict[str, Any]]:
- if not collationMetadata:
+ def _dataTypeJsonValue(
+ self,
+ string_metadata: Dict[str, str],
+ char_varchar_metadata: Dict[str, str],
+ ) -> Union[str, Dict[str, Any]]:
+ if not string_metadata and not char_varchar_metadata:
return self.dataType.jsonValue()
def removeCollations(dt: DataType) -> DataType:
@@ -1445,12 +1507,12 @@ class StructField(DataType):
removeCollations(dt.valueType),
dt.valueContainsNull,
)
- elif isinstance(dt, StringType):
- return StringType()
elif isinstance(dt, VarcharType):
return VarcharType(dt.length)
elif isinstance(dt, CharType):
return CharType(dt.length)
+ elif isinstance(dt, StringType):
+ return StringType()
else:
return dt
@@ -1461,55 +1523,66 @@ class StructField(DataType):
@classmethod
def fromJson(cls, json: Dict[str, Any]) -> "StructField":
metadata = json.get("metadata")
- collationsMap = {}
- if metadata and _COLLATIONS_METADATA_KEY in metadata:
- collationsMap = metadata[_COLLATIONS_METADATA_KEY]
- for key, value in collationsMap.items():
- nameParts = value.split(".")
- assert len(nameParts) == 2
- provider, name = nameParts[0], nameParts[1]
- _assert_valid_collation_provider(provider)
- collationsMap[key] = name
-
+ collationsMap = _parse_collation_metadata_map(metadata, _COLLATIONS_METADATA_KEY)
+ charVarcharCollationsMap = _parse_collation_metadata_map(
+ metadata, _CHAR_VARCHAR_COLLATIONS_METADATA_KEY
+ )
+ if metadata:
metadata = {
- key: value for key, value in metadata.items() if key != _COLLATIONS_METADATA_KEY
+ key: value
+ for key, value in metadata.items()
+ if key not in (_COLLATIONS_METADATA_KEY, _CHAR_VARCHAR_COLLATIONS_METADATA_KEY)
}
+ remaining: Set[str] = set(charVarcharCollationsMap)
+ parsed_type = _parse_datatype_json_value(
+ json["type"], json["name"], collationsMap, charVarcharCollationsMap, remaining
+ )
+ if remaining:
+ raise PySparkTypeError(
+ errorClass="INVALID_CHAR_VARCHAR_COLLATION_METADATA.UNRECOGNIZED_PATH",
+ messageParameters={"fieldPath": min(remaining)},
+ )
+
return StructField(
json["name"],
- _parse_datatype_json_value(json["type"], json["name"], collationsMap),
+ parsed_type,
json.get("nullable", True),
metadata,
)
def getCollationsMap(self, metadata: Dict[str, Any]) -> Dict[str, str]:
- if not metadata or _COLLATIONS_METADATA_KEY not in metadata:
- return {}
+ """Return STRING-only collations from ``__COLLATIONS`` in field metadata.
- collationMetadata: Dict[str, str] = metadata[_COLLATIONS_METADATA_KEY]
- collationsMap: Dict[str, str] = {}
-
- for key, value in collationMetadata.items():
- nameParts = value.split(".")
- assert len(nameParts) == 2
- provider, name = nameParts[0], nameParts[1]
- _assert_valid_collation_provider(provider)
- collationsMap[key] = name
+ CHAR/VARCHAR collations are stored separately and are not returned here.
+ Use :meth:`getCharVarcharCollationsMap` for ``__CHAR_VARCHAR_COLLATIONS``.
+ """
+ return _parse_collation_metadata_map(metadata, _COLLATIONS_METADATA_KEY)
- return collationsMap
+ def getCharVarcharCollationsMap(self, metadata: Dict[str, Any]) -> Dict[str, str]:
+ """Return CHAR/VARCHAR collations from ``__CHAR_VARCHAR_COLLATIONS`` metadata."""
+ return _parse_collation_metadata_map(metadata, _CHAR_VARCHAR_COLLATIONS_METADATA_KEY)
def getCollationMetadata(self) -> Dict[str, str]:
+ """Return field paths and collations for plain STRING types."""
+ return self._getCollationMetadata(self._isCollatedPlainString)
+
+ def getCharVarcharCollationMetadata(self) -> Dict[str, str]:
+ """Return field paths and collations for CHAR/VARCHAR types."""
+ return self._getCollationMetadata(self._isCollatedCharVarchar)
+
+ def _getCollationMetadata(self, include: Callable[[DataType], bool]) -> Dict[str, str]:
def visitRecursively(dt: DataType, fieldPath: str) -> None:
if isinstance(dt, ArrayType):
- processDataType(dt.elementType, fieldPath + ".element")
+ processDataType(dt.elementType, _append_field_to_path(fieldPath, "element"))
elif isinstance(dt, MapType):
- processDataType(dt.keyType, fieldPath + ".key")
- processDataType(dt.valueType, fieldPath + ".value")
- elif isinstance(dt, StringType) and self._isCollatedString(dt):
+ processDataType(dt.keyType, _append_field_to_path(fieldPath, "key"))
+ processDataType(dt.valueType, _append_field_to_path(fieldPath, "value"))
+ elif include(dt):
collationMetadata[fieldPath] = self.schemaCollationValue(dt)
def processDataType(dt: DataType, fieldPath: str) -> None:
- if self._isCollatedString(dt):
+ if include(dt):
collationMetadata[fieldPath] = self.schemaCollationValue(dt)
else:
visitRecursively(dt, fieldPath)
@@ -1518,11 +1591,15 @@ class StructField(DataType):
visitRecursively(self.dataType, self.name)
return collationMetadata
- def _isCollatedString(self, dt: DataType) -> bool:
+ def _isCollatedCharVarchar(self, dt: DataType) -> bool:
+ return isinstance(dt, (CharType, VarcharType)) and dt.collation is not None
+
+ def _isCollatedPlainString(self, dt: DataType) -> bool:
return isinstance(dt, StringType) and not dt.isUTF8BinaryCollation()
def schemaCollationValue(self, dt: DataType) -> str:
- assert isinstance(dt, StringType)
+ assert isinstance(dt, (StringType, CharType, VarcharType))
+ assert dt.collation is not None
collationName = dt.collation
provider = StringType.collationProvider(collationName)
return f"{provider}.{collationName}"
@@ -2435,8 +2512,8 @@ _all_mappable_types: Dict[str, Type[DataType]] = {
"interval": CalendarIntervalType,
}
-_LENGTH_CHAR = re.compile(r"char\(\s*(\d+)\s*\)")
-_LENGTH_VARCHAR = re.compile(r"varchar\(\s*(\d+)\s*\)")
+_LENGTH_CHAR = re.compile(r"char\(\s*(\d+)\s*\)(?:\s+collate\s+(\w+))?")
+_LENGTH_VARCHAR = re.compile(r"varchar\(\s*(\d+)\s*\)(?:\s+collate\s+(\w+))?")
_STRING_WITH_COLLATION = re.compile(r"string\s+collate\s+(\w+)")
_FIXED_DECIMAL = re.compile(r"decimal\(\s*(\d+)\s*,\s*(-?\d+)\s*\)")
_INTERVAL_DAYTIME = re.compile(r"interval (day|hour|minute|second)( to (day|hour|minute|second))?")
@@ -2452,6 +2529,7 @@ _GEOGRAPHY_CRS_ALG = re.compile(r"geography\(\s*([\w]+:-?[\w]+)\s*,\s*(\w+)\s*\)
_GEOGRAPHY_ALG = re.compile(r"geography\(\s*(\w+)\s*\)")
_COLLATIONS_METADATA_KEY = "__COLLATIONS"
+_CHAR_VARCHAR_COLLATIONS_METADATA_KEY = "__CHAR_VARCHAR_COLLATIONS"
def _drop_metadata(d: Union[DataType, StructField]) -> Union[DataType, StructField]:
@@ -2588,15 +2666,38 @@ def _parse_parameterized_timestamp_type(precision: int, ntz: bool) -> DataType:
return TimestampNTZNanosType(precision) if ntz else TimestampLTZNanosType(precision)
+def _append_field_to_path(base_path: str, field_name: str) -> str:
+ return field_name if not base_path else f"{base_path}.{field_name}"
+
+
def _parse_datatype_json_value( # type: ignore[return]
json_value: Union[dict, str],
fieldPath: str = "",
collationsMap: Optional[Dict[str, str]] = None,
+ charVarcharCollationsMap: Optional[Dict[str, str]] = None,
+ remainingCharVarcharPaths: Optional[Set[str]] = None,
) -> DataType:
+ in_string = collationsMap is not None and fieldPath in collationsMap
+ in_char_varchar = charVarcharCollationsMap is not None and fieldPath in charVarcharCollationsMap
+ if in_string and in_char_varchar:
+ raise PySparkTypeError(
+ errorClass="INVALID_JSON_DATA_TYPE_FOR_COLLATIONS",
+ messageParameters={"jsonType": str(json_value)},
+ )
+ json_type = json_value["type"] if isinstance(json_value, dict) else json_value
+ if in_string:
+ assert collationsMap is not None
+ _assert_valid_type_for_collation(fieldPath, json_type, collationsMap)
+ if in_char_varchar:
+ assert charVarcharCollationsMap is not None
+ _assert_valid_type_for_char_varchar_collation(
+ fieldPath, json_type, charVarcharCollationsMap
+ )
+
if not isinstance(json_value, dict):
if json_value in _all_mappable_types.keys():
- if collationsMap is not None and fieldPath in collationsMap:
- _assert_valid_type_for_collation(fieldPath, json_value, collationsMap)
+ if in_string:
+ assert collationsMap is not None
collation_name = collationsMap[fieldPath]
return StringType(collation_name)
return _all_mappable_types[json_value]()
@@ -2624,10 +2725,22 @@ def _parse_datatype_json_value( # type: ignore[return]
return YearMonthIntervalType(first_field, second_field)
elif m := _STRING_WITH_COLLATION.match(json_value):
return StringType(m.group(1))
- elif m := _LENGTH_CHAR.match(json_value):
- return CharType(int(m.group(1)))
- elif m := _LENGTH_VARCHAR.match(json_value):
- return VarcharType(int(m.group(1)))
+ elif m := _LENGTH_CHAR.fullmatch(json_value):
+ collation = m.group(2)
+ if in_char_varchar:
+ assert charVarcharCollationsMap is not None
+ collation = charVarcharCollationsMap[fieldPath]
+ if remainingCharVarcharPaths is not None:
+ remainingCharVarcharPaths.discard(fieldPath)
+ return CharType(int(m.group(1)), collation)
+ elif m := _LENGTH_VARCHAR.fullmatch(json_value):
+ collation = m.group(2)
+ if in_char_varchar:
+ assert charVarcharCollationsMap is not None
+ collation = charVarcharCollationsMap[fieldPath]
+ if remainingCharVarcharPaths is not None:
+ remainingCharVarcharPaths.discard(fieldPath)
+ return VarcharType(int(m.group(1)), collation)
elif _GEOMETRY.match(json_value):
return GeometryType._from_crs(GeometryType.DEFAULT_CRS)
elif _GEOMETRY_CRS.match(json_value):
@@ -2656,14 +2769,23 @@ def _parse_datatype_json_value( # type: ignore[return]
else:
tpe = json_value["type"]
if tpe in _all_complex_types:
- if collationsMap is not None and fieldPath in collationsMap:
- _assert_valid_type_for_collation(fieldPath, tpe, collationsMap)
-
complex_type = _all_complex_types[tpe]
if complex_type is ArrayType:
- return ArrayType.fromJson(json_value, fieldPath, collationsMap)
+ return ArrayType.fromJson(
+ json_value,
+ fieldPath,
+ collationsMap,
+ charVarcharCollationsMap,
+ remainingCharVarcharPaths,
+ )
elif complex_type is MapType:
- return MapType.fromJson(json_value, fieldPath, collationsMap)
+ return MapType.fromJson(
+ json_value,
+ fieldPath,
+ collationsMap,
+ charVarcharCollationsMap,
+ remainingCharVarcharPaths,
+ )
return StructType.fromJson(json_value)
elif tpe == "udt":
return UserDefinedType.fromJson(json_value)
@@ -2674,6 +2796,43 @@ def _parse_datatype_json_value( # type: ignore[return]
)
+def _parse_collation_metadata_map(metadata: Optional[Dict[str, Any]], key: str) -> Dict[str, str]:
+ if not metadata or key not in metadata:
+ return {}
+
+ raw = metadata[key]
+ if key == _CHAR_VARCHAR_COLLATIONS_METADATA_KEY:
+ if not isinstance(raw, dict):
+ raise _invalid_char_varchar_collation_metadata(raw)
+ parsed: Dict[str, str] = {}
+ for path, value in raw.items():
+ name_parts = value.split(".") if isinstance(value, str) else None
+ if name_parts is None or len(name_parts) != 2:
+ raise _invalid_char_varchar_collation_metadata(value)
+ provider, name = name_parts
+ if not provider or not name:
+ raise _invalid_char_varchar_collation_metadata(value)
+ _assert_valid_collation_provider(provider)
+ parsed[path] = name
+ return parsed
+
+ parsed = {}
+ for path, value in raw.items():
+ nameParts = value.split(".")
+ assert len(nameParts) == 2
+ provider, name = nameParts[0], nameParts[1]
+ _assert_valid_collation_provider(provider)
+ parsed[path] = name
+ return parsed
+
+
+def _invalid_char_varchar_collation_metadata(invalid: Any) -> PySparkTypeError:
+ return PySparkTypeError(
+ errorClass="INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ messageParameters={"value": json.dumps(invalid, separators=(",", ":"))},
+ )
+
+
def _assert_valid_type_for_collation(
fieldPath: str, fieldType: Any, collationMap: Dict[str, str]
) -> None:
@@ -2684,6 +2843,21 @@ def _assert_valid_type_for_collation(
)
+def _assert_valid_type_for_char_varchar_collation(
+ fieldPath: str, fieldType: Any, collationMap: Dict[str, str]
+) -> None:
+ char_match = _LENGTH_CHAR.fullmatch(fieldType) if isinstance(fieldType, str) else None
+ varchar_match = _LENGTH_VARCHAR.fullmatch(fieldType) if isinstance(fieldType, str) else None
+ is_uncollated_char_varchar = (char_match is not None and char_match.group(2) is None) or (
+ varchar_match is not None and varchar_match.group(2) is None
+ )
+ if fieldPath in collationMap and not is_uncollated_char_varchar:
+ raise PySparkTypeError(
+ errorClass="INVALID_JSON_DATA_TYPE_FOR_COLLATIONS",
+ messageParameters={"jsonType": fieldType},
+ )
+
+
def _assert_valid_collation_provider(provider: str) -> None:
if provider.lower() not in StringType.providers:
raise PySparkValueError(
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/types/CharType.scala b/sql/api/src/main/scala/org/apache/spark/sql/types/CharType.scala
index 71b7216d9a93..669467569c71 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/types/CharType.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/types/CharType.scala
@@ -23,6 +23,10 @@ import org.apache.spark.sql.catalyst.util.CollationFactory
/**
* A data type representing fixed-length character strings with a specified length.
*
+ * A standalone collated `CharType` writes its collation inline in JSON and therefore requires a
+ * current reader. Within a [[StructField]], schema JSON stores the collation in field metadata
+ * and emits an uncollated `char(n)` type so older readers can still read the schema.
+ *
* @param length
* The fixed length of the char string (must be non-negative)
* @param collation
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala b/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala
index 91e6eade4eae..4017264ba515 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.types
import java.util.Locale
+import scala.collection.mutable
import scala.util.control.NonFatal
import com.fasterxml.jackson.databind.annotation.{JsonDeserialize, JsonSerialize}
@@ -126,6 +127,9 @@ object DataType {
private val FIXED_DECIMAL = """decimal\(\s*(\d+)\s*,\s*(\-?\d+)\s*\)""".r
private val CHAR_TYPE = """char\(\s*(\d+)\s*\)""".r
private val VARCHAR_TYPE = """varchar\(\s*(\d+)\s*\)""".r
+ private val CHAR_TYPE_WITH_COLLATION = """char\(\s*(\d+)\s*\)\s+collate\s+(\w+)""".r
+ private val VARCHAR_TYPE_WITH_COLLATION =
+ """varchar\(\s*(\d+)\s*\)\s+collate\s+(\w+)""".r
private val STRING_WITH_COLLATION = """string\s+collate\s+(\w+)""".r
private val TIMESTAMP_LTZ_NANOS_TYPE = """timestamp_ltz\(\s*(\d+)\s*\)""".r
private val TIMESTAMP_NTZ_NANOS_TYPE = """timestamp_ntz\(\s*(\d+)\s*\)""".r
@@ -135,6 +139,21 @@ object DataType {
private val GEOGRAPHY_TYPE_CRS_ALG = """geography\(\s*(\w+:-?\w+)\s*,\s*(\w+)\s*\)""".r
val COLLATIONS_METADATA_KEY = "__COLLATIONS"
+ // CHAR/VARCHAR collations are stored separately so older readers that only understand
+ // __COLLATIONS (and reject it on non-STRING types) still see uncollated char(n)/varchar(n).
+ val CHAR_VARCHAR_COLLATIONS_METADATA_KEY = "__CHAR_VARCHAR_COLLATIONS"
+
+ /**
+ * Lowers CHAR/VARCHAR recursively to the physical STRING types used to encode local data,
+ * preserving explicit collations. This does not apply CHAR/VARCHAR length semantics.
+ */
+ private[spark] def replaceCharVarcharWithCollationPreservingString(
+ dataType: DataType): DataType = {
+ dataType.transformRecursively {
+ case c: CharType => c.toStringType
+ case v: VarcharType => v.toStringType
+ }
+ }
def fromDDL(ddl: String): DataType = {
parseTypeWithFallback(
@@ -259,6 +278,10 @@ object DataType {
CharType(parseIntTypeParameterFromJson(length, "length", "CHAR"))
case VARCHAR_TYPE(length) =>
VarcharType(parseIntTypeParameterFromJson(length, "length", "VARCHAR"))
+ case CHAR_TYPE_WITH_COLLATION(length, collation) =>
+ CharType(parseIntTypeParameterFromJson(length, "length", "CHAR"), collation)
+ case VARCHAR_TYPE_WITH_COLLATION(length, collation) =>
+ VarcharType(parseIntTypeParameterFromJson(length, "length", "VARCHAR"), collation)
case STRING_WITH_COLLATION(collation) => StringType(collation)
// If the coordinate reference system (CRS) value is omitted, Parquet and other storage
// formats (Delta, Iceberg) consider "OGC:CRS84" to be the default value of the crs.
@@ -345,13 +368,46 @@ object DataType {
private[sql] def parseDataType(
json: JValue,
fieldPath: String,
- collationsMap: Map[String, String]): DataType = json match {
+ collationsMap: Map[String, String],
+ charVarcharCollationsMap: Map[String, String] = Map.empty): DataType = {
+ val remainingCharVarcharPaths = mutable.Set.from(charVarcharCollationsMap.keySet)
+ val parsedType = parseDataType(
+ json,
+ fieldPath,
+ collationsMap,
+ charVarcharCollationsMap,
+ remainingCharVarcharPaths)
+ if (remainingCharVarcharPaths.nonEmpty) {
+ throw new SparkIllegalArgumentException(
+ errorClass = "INVALID_CHAR_VARCHAR_COLLATION_METADATA.UNRECOGNIZED_PATH",
+ messageParameters = Map("fieldPath" -> remainingCharVarcharPaths.min))
+ }
+ parsedType
+ }
+
+ private def parseDataType(
+ json: JValue,
+ fieldPath: String,
+ collationsMap: Map[String, String],
+ charVarcharCollationsMap: Map[String, String],
+ remainingCharVarcharPaths: mutable.Set[String]): DataType = json match {
case JString(name) =>
- collationsMap.get(fieldPath) match {
- case Some(collation) =>
+ val stringCollation = collationsMap.get(fieldPath)
+ val charVarcharCollation = charVarcharCollationsMap.get(fieldPath)
+ (stringCollation, charVarcharCollation) match {
+ case (Some(_), Some(_)) =>
+ throw new SparkIllegalArgumentException(
+ errorClass = "INVALID_JSON_DATA_TYPE_FOR_COLLATIONS",
+ messageParameters = Map("jsonType" -> name))
+ case (None, Some(collation)) =>
+ assertValidTypeForCharVarcharCollations(fieldPath, name, charVarcharCollationsMap)
+ remainingCharVarcharPaths.remove(fieldPath)
+ stringTypeWithCollation(name, collation)
+ case (Some(collation), None) =>
assertValidTypeForCollations(fieldPath, name, collationsMap)
- stringTypeWithCollation(collation)
- case _ => nameToType(name)
+ StringType(CollationFactory.collationNameToId(collation))
+ case (None, None) =>
+ nameToType(name)
}
case JSortedObject(
@@ -359,7 +415,13 @@ object DataType {
("elementType", t: JValue),
("type", JString("array"))) =>
assertValidTypeForCollations(fieldPath, "array", collationsMap)
- val elementType = parseDataType(t, appendFieldToPath(fieldPath, "element"), collationsMap)
+ assertValidTypeForCharVarcharCollations(fieldPath, "array", charVarcharCollationsMap)
+ val elementType = parseDataType(
+ t,
+ appendFieldToPath(fieldPath, "element"),
+ collationsMap,
+ charVarcharCollationsMap,
+ remainingCharVarcharPaths)
ArrayType(elementType, n)
case JSortedObject(
@@ -368,12 +430,24 @@ object DataType {
("valueContainsNull", JBool(n)),
("valueType", v: JValue)) =>
assertValidTypeForCollations(fieldPath, "map", collationsMap)
- val keyType = parseDataType(k, appendFieldToPath(fieldPath, "key"), collationsMap)
- val valueType = parseDataType(v, appendFieldToPath(fieldPath, "value"), collationsMap)
+ assertValidTypeForCharVarcharCollations(fieldPath, "map", charVarcharCollationsMap)
+ val keyType = parseDataType(
+ k,
+ appendFieldToPath(fieldPath, "key"),
+ collationsMap,
+ charVarcharCollationsMap,
+ remainingCharVarcharPaths)
+ val valueType = parseDataType(
+ v,
+ appendFieldToPath(fieldPath, "value"),
+ collationsMap,
+ charVarcharCollationsMap,
+ remainingCharVarcharPaths)
MapType(keyType, valueType, n)
case JSortedObject(("fields", JArray(fields)), ("type", JString("struct"))) =>
assertValidTypeForCollations(fieldPath, "struct", collationsMap)
+ assertValidTypeForCharVarcharCollations(fieldPath, "struct", charVarcharCollationsMap)
StructType(fields.map(parseStructField))
// Scala/Java UDT
@@ -382,6 +456,7 @@ object DataType {
("pyClass", _),
("sqlType", _),
("type", JString("udt"))) =>
+ assertValidTypeForCharVarcharCollations(fieldPath, "udt", charVarcharCollationsMap)
if (!SqlApiConf.get.allowCreatingUDTFromString &&
!SqlApiConf.get.allowedDynamicUDTClasses.contains(udtClass)) {
throw DataTypeErrors.udtClassLoadingDisabledError(
@@ -411,6 +486,7 @@ object DataType {
("serializedClass", JString(serialized)),
("sqlType", v: JValue),
("type", JString("udt"))) =>
+ assertValidTypeForCharVarcharCollations(fieldPath, "udt", charVarcharCollationsMap)
new PythonUserDefinedType(parseDataType(v), pyClass, serialized)
case other =>
@@ -425,12 +501,16 @@ object DataType {
("name", JString(name)),
("nullable", JBool(nullable)),
("type", dataType: JValue)) =>
- val collationsMap = getCollationsMap(metadataFields)
- val metadataWithoutCollations =
- JObject(metadataFields.filterNot(_._1 == COLLATIONS_METADATA_KEY))
+ val collationsMap = getCollationsMap(metadataFields, COLLATIONS_METADATA_KEY)
+ val charVarcharCollationsMap =
+ getCollationsMap(metadataFields, CHAR_VARCHAR_COLLATIONS_METADATA_KEY)
+ val metadataWithoutCollations = JObject(metadataFields.filterNot { field =>
+ field._1 == COLLATIONS_METADATA_KEY ||
+ field._1 == CHAR_VARCHAR_COLLATIONS_METADATA_KEY
+ })
StructField(
name,
- parseDataType(dataType, name, collationsMap),
+ parseDataType(dataType, name, collationsMap, charVarcharCollationsMap),
nullable,
Metadata.fromJObject(metadataWithoutCollations))
// Support reading schema when 'metadata' is missing.
@@ -452,7 +532,23 @@ object DataType {
fieldPath: String,
fieldType: String,
collationMap: Map[String, String]): Unit = {
- if (collationMap.contains(fieldPath) && fieldType != "string") {
+ val isStringType = fieldType == "string"
+ if (collationMap.contains(fieldPath) && !isStringType) {
+ throw new SparkIllegalArgumentException(
+ errorClass = "INVALID_JSON_DATA_TYPE_FOR_COLLATIONS",
+ messageParameters = Map("jsonType" -> fieldType))
+ }
+ }
+
+ private def assertValidTypeForCharVarcharCollations(
+ fieldPath: String,
+ fieldType: String,
+ collationMap: Map[String, String]): Unit = {
+ val isUncollatedCharVarchar = fieldType match {
+ case CHAR_TYPE(_) | VARCHAR_TYPE(_) => true
+ case _ => false
+ }
+ if (collationMap.contains(fieldPath) && !isUncollatedCharVarchar) {
throw new SparkIllegalArgumentException(
errorClass = "INVALID_JSON_DATA_TYPE_FOR_COLLATIONS",
messageParameters = Map("jsonType" -> fieldType))
@@ -462,31 +558,59 @@ object DataType {
/**
* Appends a field name to a given path, using a dot separator if the path is not empty.
*/
- private def appendFieldToPath(basePath: String, fieldName: String): String = {
+ private[sql] def appendFieldToPath(basePath: String, fieldName: String): String = {
if (basePath.isEmpty) fieldName else s"$basePath.$fieldName"
}
/**
* Returns a map of field path to collation name.
*/
- private def getCollationsMap(metadataFields: List[JField]): Map[String, String] = {
- val collationsJsonOpt = metadataFields.find(_._1 == COLLATIONS_METADATA_KEY).map(_._2)
- collationsJsonOpt match {
+ private def getCollationsMap(
+ metadataFields: List[JField],
+ metadataKey: String): Map[String, String] = {
+ val requireCompleteMap = metadataKey == CHAR_VARCHAR_COLLATIONS_METADATA_KEY
+ metadataFields.find(_._1 == metadataKey).map(_._2) match {
case Some(JObject(fields)) =>
- fields.collect { case (fieldPath, JString(collation)) =>
- collation.split("\\.", 2) match {
- case Array(provider: String, collationName: String) =>
- CollationFactory.assertValidProvider(provider)
- fieldPath -> collationName
- }
- }.toMap
-
+ val parsed = Map.newBuilder[String, String]
+ fields.foreach {
+ case (fieldPath, JString(collation)) =>
+ collation.split("\\.", 2) match {
+ case Array(provider, collationName)
+ if requireCompleteMap && (provider.isEmpty || collationName.isEmpty) =>
+ throw invalidCharVarcharCollationMetadata(JString(collation))
+ case Array(provider, collationName) =>
+ CollationFactory.assertValidProvider(provider)
+ parsed += (fieldPath -> collationName)
+ case _ if requireCompleteMap =>
+ throw invalidCharVarcharCollationMetadata(JString(collation))
+ case _ =>
+ }
+ case (_, invalid) if requireCompleteMap =>
+ throw invalidCharVarcharCollationMetadata(invalid)
+ case _ =>
+ }
+ parsed.result()
+ case Some(invalid) if requireCompleteMap =>
+ throw invalidCharVarcharCollationMetadata(invalid)
case _ => Map.empty
}
}
- private def stringTypeWithCollation(collationName: String): StringType = {
- StringType(CollationFactory.collationNameToId(collationName))
+ private def invalidCharVarcharCollationMetadata(
+ invalid: JValue): SparkIllegalArgumentException = {
+ new SparkIllegalArgumentException(
+ errorClass = "INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ messageParameters = Map("value" -> compact(render(invalid))))
+ }
+
+ private def stringTypeWithCollation(typeName: String, collationName: String): StringType = {
+ typeName match {
+ case CHAR_TYPE(length) =>
+ CharType(parseIntTypeParameterFromJson(length, "length", "CHAR"), collationName)
+ case VARCHAR_TYPE(length) =>
+ VarcharType(parseIntTypeParameterFromJson(length, "length", "VARCHAR"), collationName)
+ case _ => StringType(CollationFactory.collationNameToId(collationName))
+ }
}
protected[types] def buildFormattedString(
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/types/StructField.scala b/sql/api/src/main/scala/org/apache/spark/sql/types/StructField.scala
index 255830f1ece6..342025cc2882 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/types/StructField.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/types/StructField.scala
@@ -23,7 +23,7 @@ import org.json4s.{JObject, JString}
import org.json4s.JsonAST.JValue
import org.json4s.JsonDSL._
-import org.apache.spark.SparkException
+import org.apache.spark.{SparkException, SparkIllegalArgumentException}
import org.apache.spark.annotation.Stable
import org.apache.spark.sql.catalyst.util.{CollationFactory, QuotingUtils, StringConcat}
import org.apache.spark.sql.catalyst.util.FieldMetadataUtils.FIELD_ID_METADATA_KEY
@@ -77,7 +77,9 @@ case class StructField(
}
private[sql] def dataTypeJsonValue: JValue = {
- if (collationMetadata.isEmpty) return dataType.jsonValue
+ if (stringCollationMetadata.isEmpty && charVarcharCollationMetadata.isEmpty) {
+ return dataType.jsonValue
+ }
def removeCollations(dt: DataType): DataType = dt match {
// Only recurse into map and array types as any child struct type
@@ -96,36 +98,63 @@ case class StructField(
}
private def metadataJson: JValue = {
- val metadataJsonValue = metadata.jsonValue
- metadataJsonValue match {
- case JObject(fields) if collationMetadata.nonEmpty =>
- val collationFields = collationMetadata.map(kv => kv._1 -> JString(kv._2)).toList
- JObject(fields :+ (DataType.COLLATIONS_METADATA_KEY -> JObject(collationFields)))
-
- case _ => metadataJsonValue
+ if (metadata.contains(DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY)) {
+ throw new SparkIllegalArgumentException(
+ errorClass = "INVALID_CHAR_VARCHAR_COLLATION_METADATA.RESERVED_METADATA_KEY",
+ messageParameters = Map(
+ "metadataKey" -> DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY))
+ }
+ metadata.jsonValue match {
+ case JObject(fields) =>
+ val withString =
+ if (stringCollationMetadata.nonEmpty) {
+ val collationFields =
+ stringCollationMetadata.map(kv => kv._1 -> JString(kv._2)).toList
+ fields :+ (DataType.COLLATIONS_METADATA_KEY -> JObject(collationFields))
+ } else {
+ fields
+ }
+ val withBoth =
+ if (charVarcharCollationMetadata.nonEmpty) {
+ val collationFields =
+ charVarcharCollationMetadata.map(kv => kv._1 -> JString(kv._2)).toList
+ withString :+
+ (DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY -> JObject(collationFields))
+ } else {
+ withString
+ }
+ JObject(withBoth)
+ case other => other
}
}
- /** Map of field path to collation name. */
- private lazy val collationMetadata: Map[String, String] = {
+ /** Map of field path to STRING collation name. */
+ private lazy val stringCollationMetadata: Map[String, String] =
+ collectCollationMetadata(isCollatedPlainString)
+
+ /** Map of field path to CHAR/VARCHAR collation name. */
+ private lazy val charVarcharCollationMetadata: Map[String, String] =
+ collectCollationMetadata(isCollatedCharVarchar)
+
+ private def collectCollationMetadata(include: DataType => Boolean): Map[String, String] = {
val fieldToCollationMap = mutable.Map[String, String]()
def visitRecursively(dt: DataType, path: String): Unit = dt match {
case at: ArrayType =>
- processDataType(at.elementType, path + ".element")
+ processDataType(at.elementType, DataType.appendFieldToPath(path, "element"))
case mt: MapType =>
- processDataType(mt.keyType, path + ".key")
- processDataType(mt.valueType, path + ".value")
+ processDataType(mt.keyType, DataType.appendFieldToPath(path, "key"))
+ processDataType(mt.valueType, DataType.appendFieldToPath(path, "value"))
- case st: StringType if isCollatedString(st) =>
+ case st: StringType if include(st) =>
fieldToCollationMap(path) = schemaCollationValue(st)
case _ =>
}
def processDataType(dt: DataType, path: String): Unit = {
- if (isCollatedString(dt)) {
+ if (include(dt)) {
fieldToCollationMap(path) = schemaCollationValue(dt)
} else {
visitRecursively(dt, path)
@@ -136,7 +165,14 @@ case class StructField(
fieldToCollationMap.toMap
}
- private def isCollatedString(dt: DataType): Boolean = dt match {
+ private def isCollatedCharVarchar(dt: DataType): Boolean = dt match {
+ case c: CharType => c.collation.isDefined
+ case v: VarcharType => v.collation.isDefined
+ case _ => false
+ }
+
+ private def isCollatedPlainString(dt: DataType): Boolean = dt match {
+ case _: CharType | _: VarcharType => false
case st: StringType => !st.isUTF8BinaryCollation
case _ => false
}
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/types/VarcharType.scala b/sql/api/src/main/scala/org/apache/spark/sql/types/VarcharType.scala
index 11d5bc79fd9b..4fc969a693ef 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/types/VarcharType.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/types/VarcharType.scala
@@ -22,6 +22,10 @@ import org.apache.spark.sql.catalyst.util.CollationFactory
/**
* A data type representing variable-length character strings with a specified maximum length.
*
+ * A standalone collated `VarcharType` writes its collation inline in JSON and therefore requires
+ * a current reader. Within a [[StructField]], schema JSON stores the collation in field metadata
+ * and emits an uncollated `varchar(n)` type so older readers can still read the schema.
+ *
* @param length
* The maximum length of the varchar string (must be non-negative)
* @param collation
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/types/StructTypeSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/types/StructTypeSuite.scala
index 3e72cdf0f392..d6274a201bca 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/types/StructTypeSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/types/StructTypeSuite.scala
@@ -18,6 +18,8 @@
package org.apache.spark.sql.types
import com.fasterxml.jackson.databind.ObjectMapper
+import org.json4s.{JArray, JBool, JObject, JString, JValue}
+import org.json4s.jackson.JsonMethods.parse
import org.apache.spark.{SparkException, SparkFunSuite, SparkIllegalArgumentException}
import org.apache.spark.sql.AnalysisException
@@ -45,6 +47,52 @@ class StructTypeSuite extends SparkFunSuite with SQLHelper {
private val UTF8_LCASE_COLLATION = "UTF8_LCASE"
private val mapper = new ObjectMapper()
+ /**
+ * Reduced independent reader copied from the relevant JSON branches at the PR's base commit,
+ * 389de941f002a5c92e22dc3ed0f65af602a174db. It deliberately does not call the current DataType
+ * parser, so unknown metadata continues to exercise the preceding-reader contract.
+ */
+ private def readWithPreCharVarcharCollationReader(json: String): StructType = {
+ def readType(json: JValue): DataType = json match {
+ case JString(name) if name.startsWith("char(") =>
+ CharType(name.stripPrefix("char(").stripSuffix(")").toInt)
+ case JString(name) if name.startsWith("varchar(") =>
+ VarcharType(name.stripPrefix("varchar(").stripSuffix(")").toInt)
+ case JObject(fields) if fields.toMap.get("type").contains(JString("struct")) =>
+ val values = fields.toMap
+ val JArray(nestedFields) = values("fields")
+ StructType(nestedFields.map(readField))
+ case JObject(fields) if fields.toMap.get("type").contains(JString("array")) =>
+ val values = fields.toMap
+ ArrayType(readType(values("elementType")), values("containsNull").asInstanceOf[JBool].value)
+ case JObject(fields) if fields.toMap.get("type").contains(JString("map")) =>
+ val values = fields.toMap
+ MapType(
+ readType(values("keyType")),
+ readType(values("valueType")),
+ values("valueContainsNull").asInstanceOf[JBool].value)
+ case other =>
+ fail(s"Unsupported type in pinned preceding-reader fixture: $other")
+ }
+
+ def readField(json: JValue): StructField = {
+ val values = json.asInstanceOf[JObject].obj.toMap
+ val JString(name) = values("name")
+ val JBool(nullable) = values("nullable")
+ val dataType = values("type")
+ val JObject(metadataFields) = values("metadata")
+ StructField(
+ name,
+ readType(dataType),
+ nullable,
+ Metadata.fromJObject(JObject(metadataFields)))
+ }
+
+ val rootValues = parse(json).asInstanceOf[JObject].obj.toMap
+ val JArray(fields) = rootValues("fields")
+ StructType(fields.map(readField))
+ }
+
test("lookup a single missing field should output existing fields") {
checkError(
exception = intercept[SparkIllegalArgumentException](s("c")),
@@ -673,6 +721,291 @@ class StructTypeSuite extends SparkFunSuite with SQLHelper {
assert(fromDDL(struct.toDDL) === struct)
}
+ test("SPARK-59276: CHAR/VARCHAR collations round trip through JSON") {
+ val dataTypes = Seq(
+ CharType(4, "UTF8_BINARY"),
+ CharType(4, "UTF8_LCASE"),
+ VarcharType(6, "UNICODE_CI"),
+ StructType(
+ StructField("c", CharType(4, "UTF8_LCASE")) ::
+ StructField("v", ArrayType(VarcharType(6, "UNICODE_CI"))) :: Nil),
+ StructType(
+ StructField(
+ "mixed",
+ MapType(CharType(4, "UTF8_BINARY"), VarcharType(6, "UNICODE_CI"))) :: Nil),
+ StructType(StructField("", ArrayType(CharType(4, "UTF8_LCASE"))) :: Nil),
+ StructType(
+ StructField(
+ "",
+ MapType(CharType(4, "UTF8_BINARY"), VarcharType(6, "UNICODE_CI"))) :: Nil))
+
+ dataTypes.foreach { dataType =>
+ assert(DataType.fromJson(dataType.json) === dataType)
+ }
+
+ val compatibilitySchema = StructType(
+ StructField("plain", CharType(3)) ::
+ StructField("binary", CharType(4, "UTF8_BINARY")) ::
+ StructField("nested", ArrayType(VarcharType(6, "UNICODE_CI"))) :: Nil)
+ val expectedJson =
+ s"""
+ |{
+ | "type": "struct",
+ | "fields": [
+ | {
+ | "name": "plain",
+ | "type": "char(3)",
+ | "nullable": true,
+ | "metadata": {}
+ | },
+ | {
+ | "name": "binary",
+ | "type": "char(4)",
+ | "nullable": true,
+ | "metadata": {
+ | "${DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY}": {
+ | "binary": "spark.UTF8_BINARY"
+ | }
+ | }
+ | },
+ | {
+ | "name": "nested",
+ | "type": {
+ | "type": "array",
+ | "elementType": "varchar(6)",
+ | "containsNull": true
+ | },
+ | "nullable": true,
+ | "metadata": {
+ | "${DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY}": {
+ | "nested.element": "icu.UNICODE_CI"
+ | }
+ | }
+ | }
+ | ]
+ |}
+ |""".stripMargin
+ assert(mapper.readTree(compatibilitySchema.json) == mapper.readTree(expectedJson))
+ }
+
+ test("SPARK-59276: preceding readers retain unknown CHAR/VARCHAR collation metadata") {
+ val schema = StructType(
+ StructField("c", CharType(4, "UTF8_LCASE")) ::
+ StructField("nested", ArrayType(VarcharType(6, "UNICODE_CI"))) ::
+ StructField(
+ "mapped",
+ MapType(CharType(3, "UTF8_BINARY"), VarcharType(5, "UTF8_LCASE"))) ::
+ StructField(
+ "nestedStruct",
+ StructType(StructField("c", CharType(2, "UNICODE_CI")) :: Nil)) :: Nil)
+
+ val precedingSchema = readWithPreCharVarcharCollationReader(schema.json)
+ assert(precedingSchema("c").dataType === CharType(4))
+ assert(precedingSchema("nested").dataType === ArrayType(VarcharType(6)))
+ assert(precedingSchema("mapped").dataType === MapType(CharType(3), VarcharType(5)))
+ val nestedStruct = precedingSchema("nestedStruct").dataType.asInstanceOf[StructType]
+ assert(nestedStruct("c").dataType === CharType(2))
+
+ val metadataKey = DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY
+ assert(precedingSchema("c").metadata.getMetadata(metadataKey).getString("c") ===
+ "spark.UTF8_LCASE")
+ assert(
+ precedingSchema("nested").metadata.getMetadata(metadataKey).getString("nested.element") ===
+ "icu.UNICODE_CI")
+ val mapMetadata = precedingSchema("mapped").metadata.getMetadata(metadataKey)
+ assert(mapMetadata.getString("mapped.key") === "spark.UTF8_BINARY")
+ assert(mapMetadata.getString("mapped.value") === "spark.UTF8_LCASE")
+ assert(nestedStruct("c").metadata.getMetadata(metadataKey).getString("c") ===
+ "icu.UNICODE_CI")
+ }
+
+ test("SPARK-59276: STRING and CHAR collations use separate JSON keys") {
+ val schema = StructType(
+ StructField(
+ "mixed",
+ MapType(StringType("UTF8_LCASE"), CharType(4, "UNICODE_CI"))) :: Nil)
+ val json = schema.json
+ assert(json.contains(DataType.COLLATIONS_METADATA_KEY))
+ assert(json.contains(DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY))
+ assert(DataType.fromJson(json) === schema)
+ }
+
+ test("SPARK-59276: reject CHAR/VARCHAR collations stored in __COLLATIONS") {
+ val json =
+ s"""
+ |{
+ | "type": "struct",
+ | "fields": [
+ | {
+ | "name": "c",
+ | "type": "char(4)",
+ | "nullable": true,
+ | "metadata": {
+ | "${DataType.COLLATIONS_METADATA_KEY}": {
+ | "c": "spark.UTF8_LCASE"
+ | }
+ | }
+ | }
+ | ]
+ |}
+ |""".stripMargin
+ checkError(
+ exception = intercept[SparkIllegalArgumentException] {
+ DataType.fromJson(json)
+ },
+ condition = "INVALID_JSON_DATA_TYPE_FOR_COLLATIONS",
+ parameters = Map("jsonType" -> "char(4)"))
+ }
+
+ test("SPARK-59276: reject inline CHAR collation plus metadata") {
+ val json =
+ s"""
+ |{
+ | "type": "struct",
+ | "fields": [
+ | {
+ | "name": "c",
+ | "type": "char(4) collate UTF8_LCASE",
+ | "nullable": true,
+ | "metadata": {
+ | "${DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY}": {
+ | "c": "spark.UTF8_LCASE"
+ | }
+ | }
+ | }
+ | ]
+ |}
+ |""".stripMargin
+ checkError(
+ exception = intercept[SparkIllegalArgumentException] {
+ DataType.fromJson(json)
+ },
+ condition = "INVALID_JSON_DATA_TYPE_FOR_COLLATIONS",
+ parameters = Map("jsonType" -> "char(4) collate UTF8_LCASE"))
+ }
+
+ test("SPARK-59276: reject CHAR/VARCHAR collation metadata on other atomic types") {
+ val json =
+ s"""
+ |{
+ | "type": "struct",
+ | "fields": [
+ | {
+ | "name": "d",
+ | "type": "decimal(10,2)",
+ | "nullable": true,
+ | "metadata": {
+ | "${DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY}": {
+ | "d": "spark.UTF8_LCASE"
+ | }
+ | }
+ | }
+ | ]
+ |}
+ |""".stripMargin
+ checkError(
+ exception = intercept[SparkIllegalArgumentException] {
+ DataType.fromJson(json)
+ },
+ condition = "INVALID_JSON_DATA_TYPE_FOR_COLLATIONS",
+ parameters = Map("jsonType" -> "decimal(10,2)"))
+ }
+
+ test("SPARK-59276: reject CHAR/VARCHAR collation metadata on UDT fields") {
+ val udtTypes = Seq(
+ """{"type":"udt","class":"invalid.Class","pyClass":null,"sqlType":"string"}""",
+ """{"type":"udt","pyClass":"test.Udt","serializedClass":"","sqlType":"string"}""")
+ udtTypes.foreach { udtType =>
+ val json =
+ s"""
+ |{
+ | "type": "struct",
+ | "fields": [
+ | {
+ | "name": "u",
+ | "type": $udtType,
+ | "nullable": true,
+ | "metadata": {
+ | "${DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY}": {
+ | "u": "spark.UTF8_LCASE"
+ | }
+ | }
+ | }
+ | ]
+ |}
+ |""".stripMargin
+ checkError(
+ exception = intercept[SparkIllegalArgumentException] {
+ DataType.fromJson(json)
+ },
+ condition = "INVALID_JSON_DATA_TYPE_FOR_COLLATIONS",
+ parameters = Map("jsonType" -> "udt"))
+ }
+ }
+
+ test("SPARK-59276: reject caller metadata using the CHAR/VARCHAR collation key") {
+ val metadata = new MetadataBuilder()
+ .putString(DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY, "caller")
+ .build()
+ val field = StructField("c", CharType(4, "UTF8_LCASE"), metadata = metadata)
+ checkError(
+ exception = intercept[SparkIllegalArgumentException] {
+ field.jsonValue
+ },
+ condition = "INVALID_CHAR_VARCHAR_COLLATION_METADATA.RESERVED_METADATA_KEY",
+ parameters = Map("metadataKey" -> DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY))
+ }
+
+ test("SPARK-59276: reject malformed CHAR/VARCHAR collation restoration maps") {
+ def schemaJson(metadataValue: String): String =
+ s"""
+ |{
+ | "type": "struct",
+ | "fields": [
+ | {
+ | "name": "c",
+ | "type": "char(4)",
+ | "nullable": true,
+ | "metadata": {
+ | "${DataType.CHAR_VARCHAR_COLLATIONS_METADATA_KEY}": $metadataValue
+ | }
+ | }
+ | ]
+ |}
+ |""".stripMargin
+
+ Seq(
+ (
+ "\"caller\"",
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ Map("value" -> "\"caller\"")),
+ (
+ """{"c": 1}""",
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ Map("value" -> "1")),
+ (
+ """{"c": "spark."}""",
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ Map("value" -> "\"spark.\"")),
+ (
+ """{"c": ".UTF8_LCASE"}""",
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.INVALID_VALUE",
+ Map("value" -> "\".UTF8_LCASE\"")),
+ (
+ """{"c": "spark.UTF8_LCASE", "typo": "spark.UTF8_LCASE"}""",
+ "INVALID_CHAR_VARCHAR_COLLATION_METADATA.UNRECOGNIZED_PATH",
+ Map("fieldPath" -> "typo"))
+ ).foreach {
+ case (metadataValue, condition, parameters) =>
+ checkError(
+ exception = intercept[SparkIllegalArgumentException] {
+ DataType.fromJson(schemaJson(metadataValue))
+ },
+ condition = condition,
+ parameters = parameters)
+ }
+ }
+
test("simple struct with collations to json") {
val simpleStruct = StructType(
StructField("c1", StringType(UNICODE_COLLATION)) :: Nil)
diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/ClientE2ETestSuite.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/ClientE2ETestSuite.scala
index 502f452eed73..7e139a063ccb 100644
--- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/ClientE2ETestSuite.scala
+++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/ClientE2ETestSuite.scala
@@ -659,6 +659,24 @@ class ClientE2ETestSuite
testCapturedStdOut(df.printSchema(5), expectedSchema.treeString(5))
}
+ test("SPARK-59276: CHAR/VARCHAR result schema preserves collations") {
+ withSQLConf("spark.sql.charVarchar.standardSemantics.enabled" -> "true") {
+ val schema = spark
+ .sql("""SELECT
+ | CAST('ab' AS CHAR(4)) AS c,
+ | CAST('cd' AS VARCHAR(6)) AS v,
+ | CAST('ef' AS CHAR(4) COLLATE UTF8_LCASE) AS collated_c,
+ | CAST('gh' AS VARCHAR(6) COLLATE UNICODE_CI) AS collated_v
+ |""".stripMargin)
+ .schema
+
+ assert(schema("c").dataType === CharType(4))
+ assert(schema("v").dataType === VarcharType(6))
+ assert(schema("collated_c").dataType === CharType(4, "UTF8_LCASE"))
+ assert(schema("collated_v").dataType === VarcharType(6, "UNICODE_CI"))
+ }
+ }
+
test("Dataset explain") {
val df = spark.range(10)
val simpleExplainFragments = Seq("== Physical Plan ==")
@@ -1148,6 +1166,76 @@ class ClientE2ETestSuite
checkSameResult(rows.asScala, spark.createDataFrame(rows, schema))
}
+ test("SPARK-59276: SparkSession.createDataFrame preserves CHAR/VARCHAR collations") {
+ val rows = java.util.Arrays.asList(Row("ab", "cd", Row("xy", Seq("z"))))
+ val emptyRows = java.util.Collections.emptyList[Row]()
+ val schema = new StructType()
+ .add("c", CharType(4, "UTF8_LCASE"))
+ .add("v", VarcharType(6, "UNICODE_CI"))
+ .add(
+ "nested",
+ new StructType()
+ .add("c", CharType(3, "UTF8_LCASE"))
+ .add("v", ArrayType(VarcharType(2, "UNICODE_CI"))))
+
+ withSQLConf(
+ "spark.sql.charVarchar.standardSemantics.enabled" -> "true",
+ "spark.sql.legacy.charVarcharAsString" -> "false") {
+ val dataFrame = spark.createDataFrame(rows, schema)
+ assert(dataFrame.schema === schema)
+ checkAnswer(dataFrame, Row("ab ", "cd", Row("xy ", Seq("z"))))
+
+ val emptyDataFrame = spark.createDataFrame(emptyRows, schema)
+ assert(emptyDataFrame.schema === schema)
+ checkAnswer(emptyDataFrame, Seq.empty)
+
+ withSQLConf(SqlApiConf.LOCAL_RELATION_CACHE_THRESHOLD_KEY -> "1") {
+ val cachedDataFrame = spark.createDataFrame(rows, schema)
+ assert(cachedDataFrame.schema === schema)
+ checkAnswer(cachedDataFrame, Row("ab ", "cd", Row("xy ", Seq("z"))))
+ }
+
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ spark
+ .createDataFrame(
+ java.util.Arrays.asList(Row("abcde", "cd", Row("xy", Seq("z")))),
+ schema)
+ .collect()
+ },
+ condition = "EXCEED_LIMIT_LENGTH",
+ parameters = Map("limit" -> "4"))
+ }
+
+ withSQLConf(
+ "spark.sql.charVarchar.standardSemantics.enabled" -> "false",
+ "spark.sql.legacy.charVarcharAsString" -> "true") {
+ val expectedSchema = DataType
+ .replaceCharVarcharWithCollationPreservingString(schema)
+ .asInstanceOf[StructType]
+ val dataFrame = spark.createDataFrame(rows, schema)
+ assert(dataFrame.schema === expectedSchema)
+ checkAnswer(dataFrame, Row("ab", "cd", Row("xy", Seq("z"))))
+
+ val emptyDataFrame = spark.createDataFrame(emptyRows, schema)
+ assert(emptyDataFrame.schema === expectedSchema)
+ checkAnswer(emptyDataFrame, Seq.empty)
+ }
+
+ withSQLConf(
+ "spark.sql.charVarchar.standardSemantics.enabled" -> "false",
+ "spark.sql.legacy.charVarcharAsString" -> "false") {
+ Seq(rows, emptyRows).foreach { input =>
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.createDataFrame(input, schema).schema
+ },
+ condition = "UNSUPPORTED_CHAR_OR_VARCHAR_AS_STRING",
+ parameters = Map.empty)
+ }
+ }
+ }
+
test("SparkSession.createDataFrame - bean") {
def bean(v: String): SimpleBean = {
val bean = new SimpleBean
diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
index 9963ec5620ff..a445b289aea8 100644
--- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
+++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala
@@ -309,14 +309,14 @@ class ArrowEncoderSuite extends ConnectFunSuite {
}
}
- test("SPARK-58794: char/varchar round trip") {
+ test("SPARK-58794, SPARK-59276: char/varchar round trip") {
// The client cannot see the server's charVarchar configuration, so a result schema carrying
// CHAR/VARCHAR must be decodable regardless of the local one. Values are padded and length
// checked by the server, so the client passes them through unchanged.
val encoder = toResultRowEncoder(
new StructType()
- .add("c", CharType(4))
- .add("v", VarcharType(6))
+ .add("c", CharType(4, "UTF8_LCASE"))
+ .add("v", VarcharType(6, "UNICODE_CI"))
.add("s", new StructType().add("c", CharType(4)))
.add("a", ArrayType(VarcharType(6))))
roundTripAndCheckIdentical(encoder) { () =>
@@ -331,6 +331,13 @@ class ArrowEncoderSuite extends ConnectFunSuite {
}
}
+ test("SPARK-59276: local data physical schema does not unwrap UDTs") {
+ val schema = new StructType().add("udt", new UDTNotSupported)
+ val physicalSchema =
+ DataType.replaceCharVarcharWithCollationPreservingString(schema).asInstanceOf[StructType]
+ assert(physicalSchema === schema)
+ }
+
test("single batch") {
val inspector = new CountingBatchInspector
roundTripAndCheckIdentical(singleIntEncoder, inspectBatch = inspector) { () =>
diff --git a/sql/connect/common/src/main/protobuf/spark/connect/types.proto b/sql/connect/common/src/main/protobuf/spark/connect/types.proto
index a6bb292899b5..45c5184bdf93 100644
--- a/sql/connect/common/src/main/protobuf/spark/connect/types.proto
+++ b/sql/connect/common/src/main/protobuf/spark/connect/types.proto
@@ -175,11 +175,17 @@ message DataType {
message Char {
int32 length = 1;
uint32 type_variation_reference = 2;
+ // (Optional) Collation name. When omitted, the collation was not explicitly specified.
+ // Older clients ignore this field and decode the historical uncollated CHAR type.
+ optional string collation = 3;
}
message VarChar {
int32 length = 1;
uint32 type_variation_reference = 2;
+ // (Optional) Collation name. When omitted, the collation was not explicitly specified.
+ // Older clients ignore this field and decode the historical uncollated VARCHAR type.
+ optional string collation = 3;
}
message Decimal {
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala
index ed741e55dde5..755aca770bdb 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala
@@ -56,7 +56,7 @@ import org.apache.spark.sql.connect.client.SparkConnectClient.Configuration
import org.apache.spark.sql.connect.client.arrow.ArrowSerializer
import org.apache.spark.sql.internal.{SessionState, SharedState, SqlApiConf, SubqueryExpression}
import org.apache.spark.sql.sources.BaseRelation
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{DataType, StructType}
import org.apache.spark.sql.util.{CloseableIterator, ExecutionListenerManager}
import org.apache.spark.util.ArrayImplicits._
@@ -116,7 +116,11 @@ class SparkSession private[sql] (
/** @inheritdoc */
def emptyDataset[T: Encoder]: Dataset[T] = createDataset[T](Nil)
- private def createDataset[T](encoder: AgnosticEncoder[T], data: Iterator[T]): Dataset[T] = {
+ private def createDataset[T](
+ encoder: AgnosticEncoder[T],
+ data: Iterator[T],
+ requestedSchema: Option[StructType] = None): Dataset[T] = {
+ val relationSchema = requestedSchema.getOrElse(encoder.schema)
newDataset(encoder) { builder =>
if (data.nonEmpty) {
val confs = conf.getConfigMap(
@@ -150,7 +154,7 @@ class SparkSession private[sql] (
batchSizeCheckInterval = math.min(1024, maxChunkSizeRows))
try {
- val schemaBytes = encoder.schema.json.getBytes
+ val schemaBytes = relationSchema.json.getBytes
// Schema is the first chunk, data chunks follow from the iterator
val currentBatch = scala.collection.mutable.ArrayBuffer[Array[Byte]](schemaBytes)
var totalChunks = 1
@@ -191,7 +195,7 @@ class SparkSession private[sql] (
// Schema + single small data chunk: use LocalRelation with inline data
val arrowData = ByteString.copyFrom(currentBatch.last)
builder.getLocalRelationBuilder
- .setSchema(encoder.schema.json)
+ .setSchema(relationSchema.json)
.setData(arrowData)
} else {
// Multiple data chunks or large data: use ChunkedCachedLocalRelation
@@ -211,7 +215,7 @@ class SparkSession private[sql] (
}
} else {
builder.getLocalRelationBuilder
- .setSchema(encoder.schema.json)
+ .setSchema(relationSchema.json)
}
}
}
@@ -223,7 +227,15 @@ class SparkSession private[sql] (
/** @inheritdoc */
def createDataFrame(rows: java.util.List[Row], schema: StructType): DataFrame = {
- createDataset(RowEncoder.encoderFor(schema), rows.iterator().asScala).toDF()
+ // RowEncoder applies CHAR/VARCHAR semantics from the client process's local SqlApiConf, which
+ // can differ from the server-side configuration visible through SparkSession.conf. Send raw
+ // string values and let the server apply its policy to the separately provided logical schema.
+ val physicalSchema =
+ DataType.replaceCharVarcharWithCollationPreservingString(schema).asInstanceOf[StructType]
+ createDataset(
+ RowEncoder.encoderForResultSchema(physicalSchema),
+ rows.iterator().asScala,
+ requestedSchema = Some(schema)).toDF()
}
/** @inheritdoc */
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala
index 15b966ced2ca..d3384a904aec 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/DataTypeProtoConverter.scala
@@ -51,8 +51,8 @@ object DataTypeProtoConverter {
case proto.DataType.KindCase.DECIMAL => toCatalystDecimalType(t.getDecimal)
case proto.DataType.KindCase.STRING => toCatalystStringType(t.getString)
- case proto.DataType.KindCase.CHAR => CharType(t.getChar.getLength)
- case proto.DataType.KindCase.VAR_CHAR => VarcharType(t.getVarChar.getLength)
+ case proto.DataType.KindCase.CHAR => toCatalystCharType(t.getChar)
+ case proto.DataType.KindCase.VAR_CHAR => toCatalystVarcharType(t.getVarChar)
case proto.DataType.KindCase.DATE => DateType
case proto.DataType.KindCase.TIMESTAMP => TimestampType
@@ -103,6 +103,12 @@ object DataTypeProtoConverter {
private def toCatalystStringType(t: proto.DataType.String): StringType =
if (t.getCollation.nonEmpty) StringType(t.getCollation) else StringType
+ private def toCatalystCharType(t: proto.DataType.Char): CharType =
+ if (t.hasCollation) CharType(t.getLength, t.getCollation) else CharType(t.getLength)
+
+ private def toCatalystVarcharType(t: proto.DataType.VarChar): VarcharType =
+ if (t.hasCollation) VarcharType(t.getLength, t.getCollation) else VarcharType(t.getLength)
+
private def toCatalystYearMonthIntervalType(t: proto.DataType.YearMonthInterval) = {
(t.hasStartField, t.hasEndField) match {
case (true, true) => YearMonthIntervalType(t.getStartField.toByte, t.getEndField.toByte)
@@ -212,15 +218,23 @@ object DataTypeProtoConverter {
.build()
case c: CharType =>
+ val charBuilder = proto.DataType.Char.newBuilder().setLength(c.length)
+ c.collation.foreach { collationId =>
+ charBuilder.setCollation(CollationFactory.fetchCollation(collationId).collationName)
+ }
proto.DataType
.newBuilder()
- .setChar(proto.DataType.Char.newBuilder().setLength(c.length).build())
+ .setChar(charBuilder.build())
.build()
case v: VarcharType =>
+ val varcharBuilder = proto.DataType.VarChar.newBuilder().setLength(v.length)
+ v.collation.foreach { collationId =>
+ varcharBuilder.setCollation(CollationFactory.fetchCollation(collationId).collationName)
+ }
proto.DataType
.newBuilder()
- .setVarChar(proto.DataType.VarChar.newBuilder().setLength(v.length).build())
+ .setVarChar(varcharBuilder.build())
.build()
// StringType must be matched after CharType and VarcharType
diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
index 4803c5865e14..0d11148a2441 100644
--- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
+++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
@@ -1517,7 +1517,10 @@ class SparkConnectPlanner(
if (schema == null) {
throw InvalidInputErrors.schemaRequiredForLocalRelation()
}
- LocalRelation(schema)
+ val physicalSchema = DataType
+ .replaceCharVarcharWithCollationPreservingString(normalizeLocalRelationType(schema))
+ .asInstanceOf[StructType]
+ buildLocalRelationFromRows(Iterator.empty, physicalSchema, Some(schema))
}
}
@@ -1597,6 +1600,59 @@ class SparkConnectPlanner(
case d => StructType(Seq(StructField("value", d)))
}
+ private def normalizeLocalRelationType(dt: DataType): DataType = dt match {
+ case udt: UserDefinedType[_] => normalizeLocalRelationType(udt.sqlType)
+ case StructType(fields) =>
+ val newFields = fields.zipWithIndex.map {
+ case (StructField(_, dataType, nullable, metadata), i) =>
+ StructField(s"col_$i", normalizeLocalRelationType(dataType), nullable, metadata)
+ }
+ StructType(newFields)
+ case ArrayType(elementType, containsNull) =>
+ ArrayType(normalizeLocalRelationType(elementType), containsNull)
+ case MapType(keyType, valueType, valueContainsNull) =>
+ MapType(
+ normalizeLocalRelationType(keyType),
+ normalizeLocalRelationType(valueType),
+ valueContainsNull)
+ case _ => dt
+ }
+
+ /**
+ * Restores requested logical wrappers that remain compatible after local-data reconciliation.
+ *
+ * For example, `createDataFrame(rows, schemaWithChar)` produces CHAR under standard semantics,
+ * STRING under legacy-as-string, and is rejected by the default policy. Existing UDT wrappers
+ * are retained without extending CHAR/VARCHAR reconciliation into their storage types.
+ */
+ private def restoreRequestedLogicalType(actual: DataType, requested: DataType): DataType =
+ (actual, requested) match {
+ case (_, requestedUdt: UserDefinedType[_]) => requestedUdt
+ case (actualString: StringType, requestedString: StringType)
+ if !actualString.isInstanceOf[CharType] &&
+ !actualString.isInstanceOf[VarcharType] &&
+ !requestedString.isInstanceOf[CharType] &&
+ !requestedString.isInstanceOf[VarcharType] &&
+ DataType.equalsIgnoreCompatibleCollation(actualString, requestedString) =>
+ requestedString
+ case (StructType(actualFields), StructType(requestedFields)) =>
+ StructType(actualFields.zip(requestedFields).map { case (actualField, requestedField) =>
+ actualField.copy(
+ name = requestedField.name,
+ dataType = restoreRequestedLogicalType(actualField.dataType, requestedField.dataType))
+ })
+ case (ArrayType(actualElement, containsNull), ArrayType(requestedElement, _)) =>
+ ArrayType(restoreRequestedLogicalType(actualElement, requestedElement), containsNull)
+ case (
+ MapType(actualKey, actualValue, valueContainsNull),
+ MapType(requestedKey, requestedValue, _)) =>
+ MapType(
+ restoreRequestedLogicalType(actualKey, requestedKey),
+ restoreRequestedLogicalType(actualValue, requestedValue),
+ valueContainsNull)
+ case _ => actual
+ }
+
private def buildLocalRelationFromRows(
rows: Iterator[InternalRow],
structType: StructType,
@@ -1613,37 +1669,28 @@ class SparkConnectPlanner(
case None =>
logical.LocalRelation(attributes, data.map(_.copy()).toArray.toImmutableArraySeq)
case Some(schema) =>
- def normalize(dt: DataType): DataType = dt match {
- case udt: UserDefinedType[_] => normalize(udt.sqlType)
- case StructType(fields) =>
- val newFields = fields.zipWithIndex.map {
- case (StructField(_, dataType, nullable, metadata), i) =>
- StructField(s"col_$i", normalize(dataType), nullable, metadata)
- }
- StructType(newFields)
- case ArrayType(elementType, containsNull) =>
- ArrayType(normalize(elementType), containsNull)
- case MapType(keyType, valueType, valueContainsNull) =>
- MapType(normalize(keyType), normalize(valueType), valueContainsNull)
- case _ => dt
- }
-
- val normalized = normalize(schema).asInstanceOf[StructType]
+ val normalized = normalizeLocalRelationType(schema).asInstanceOf[StructType]
import org.apache.spark.util.ArrayImplicits._
val project = Dataset
.ofRows(
session,
- logicalPlan = logical.LocalRelation(normalize(structType).asInstanceOf[StructType]))
+ logicalPlan = logical.LocalRelation(
+ normalizeLocalRelationType(structType).asInstanceOf[StructType]))
.toDF(normalized.names.toImmutableArraySeq: _*)
.to(normalized)
.logicalPlan
.asInstanceOf[Project]
val proj = UnsafeProjection.create(project.projectList, project.child.output)
- logical.LocalRelation(
- DataTypeUtils.toAttributes(schema),
- data.map(proj).map(_.copy()).toSeq)
+ val output = project.output.zip(schema.fields).map { case (attribute, field) =>
+ AttributeReference(
+ field.name,
+ restoreRequestedLogicalType(attribute.dataType, field.dataType),
+ attribute.nullable,
+ field.metadata)()
+ }
+ logical.LocalRelation(output, data.map(proj).map(_.copy()).toSeq)
}
}
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/LiteralExpressionProtoConverterSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/LiteralExpressionProtoConverterSuite.scala
index 242c241d2b7e..b0b8e469d7ee 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/LiteralExpressionProtoConverterSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/LiteralExpressionProtoConverterSuite.scala
@@ -90,6 +90,39 @@ class LiteralExpressionProtoConverterSuite extends AnyFunSuite { // scalastyle:i
}
}
+ test("SPARK-59276: CHAR/VARCHAR DataType proto round-trip preserves collations") {
+ val dataTypes = Seq(
+ CharType(4),
+ CharType(4, "UTF8_BINARY"),
+ CharType(4, "UTF8_LCASE"),
+ VarcharType(6),
+ VarcharType(6, "UTF8_BINARY"),
+ VarcharType(6, "UNICODE_CI"),
+ new StructType()
+ .add("c", CharType(4, "UTF8_LCASE"))
+ .add("v", ArrayType(VarcharType(6, "UNICODE_CI"))))
+
+ dataTypes.foreach { dataType =>
+ val protoType = DataTypeProtoConverter.toConnectProtoType(dataType)
+ assertResult(dataType)(DataTypeProtoConverter.toCatalystType(protoType))
+ }
+
+ val defaultChar = DataTypeProtoConverter.toConnectProtoType(CharType(4)).getChar
+ assert(!defaultChar.hasCollation)
+ val explicitBinaryChar =
+ DataTypeProtoConverter.toConnectProtoType(CharType(4, "UTF8_BINARY")).getChar
+ assert(explicitBinaryChar.hasCollation)
+ assert(explicitBinaryChar.getCollation == "UTF8_BINARY")
+
+ // Clients using an older proto definition ignore the new field. Conversely, a type from an
+ // older client has no field and retains the historical uncollated representation.
+ val oldClientChar = proto.DataType
+ .newBuilder()
+ .setChar(proto.DataType.Char.newBuilder().setLength(4))
+ .build()
+ assertResult(CharType(4))(DataTypeProtoConverter.toCatalystType(oldClientChar))
+ }
+
test("SPARK-57161: TIMESTAMP_NTZ nanosecond literal proto and catalyst value round-trip") {
// Boundary and pre-epoch values, plus a sub-microsecond value that exercises the extra nanos.
val values = Seq(
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
index a8eec9ec7d56..4d26937f9fbe 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.connect.common.InvalidPlanInput
import org.apache.spark.sql.connect.common.LiteralValueProtoConverter.toLiteralProto
import org.apache.spark.sql.execution.arrow.ArrowConverters
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType, TimeType}
+import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
/**
@@ -545,6 +545,19 @@ class SparkConnectPlannerSuite extends SparkFunSuite with SparkConnectPlanTest {
}
}
+ test("SPARK-59276: restore an ordinary requested UDT in a local relation") {
+ val ordinaryUdt = new PythonUserDefinedType(
+ ArrayType(DoubleType, containsNull = false),
+ "pyspark.testing.objects.PythonOnlyUDT",
+ "serialized")
+ val requestedSchema = StructType(StructField("value", ordinaryUdt) :: Nil)
+ val relation = proto.Relation
+ .newBuilder()
+ .setLocalRelation(proto.LocalRelation.newBuilder().setSchema(requestedSchema.json))
+ .build()
+ assert(Dataset.ofRows(spark, transform(relation)).schema("value").dataType === ordinaryUdt)
+ }
+
test("Empty ArrowBatch") {
val schema = StructType(Seq(StructField("int", IntegerType)))
val data = ArrowConverters.createEmptyArrowBatch(schema, null, true, false)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org