srielau commented on code in PR #58581:
URL: https://github.com/apache/spark/pull/58581#discussion_r4039080753
##########
python/pyspark/sql/types.py:
##########
@@ -295,8 +298,11 @@ class StringType(AtomicType):
providerICU = "icu"
providers = [providerSpark, providerICU]
- def __init__(self, collation: str = "UTF8_BINARY"):
- self.collation = collation
+ __slots__ = ("_collation_explicit",)
+
+ def __init__(self, collation: str = _DEFAULT_STRING_COLLATION):
Review Comment:
Addressed in 08550e37485 by reverting the Python StringType sentinel,
_collation_explicit state, constructor signature change, and associated tests.
The public StringType constructor and default identity behavior are restored.
##########
python/pyspark/sql/types.py:
##########
@@ -2586,10 +2659,27 @@ 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,
) -> DataType:
+ in_string = collationsMap is not None and fieldPath in collationsMap
Review Comment:
Addressed in 08550e37485. Python now validates restoration metadata before
any atomic parser branch can return. The negative coverage includes metadata
attached to decimal(10,2), which previously bypassed validation.
##########
sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala:
##########
@@ -425,12 +493,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
Review Comment:
Addressed in 08550e37485. Both Scala and Python writers now reject a
caller-owned __CHAR_VARCHAR_COLLATIONS key explicitly, with focused tests. The
value can no longer be silently overwritten, duplicated, or dropped.
##########
python/pyspark/sql/connect/session.py:
##########
@@ -608,9 +652,13 @@ def createDataFrame(
spark_types: List[Optional[DataType]]
if isinstance(schema, StructType):
deduped_schema = cast(StructType,
_deduplicate_field_names(schema))
- spark_types = [field.dataType for field in
deduped_schema.fields]
+ arrow_compatible_schema = cast(
Review Comment:
Not included per the scope freeze. Commit 08550e37485 reverts CHAR/VARCHAR
lowering from pandas and pyarrow.Table paths, along with the global Arrow
conversion changes. Those non-row inputs remain separate follow-up work.
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -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
Review Comment:
Addressed in 08550e37485. The JVM end-to-end createDataFrame test now forces
cached local-relation transport with localRelationCacheThreshold=1 and verifies
that the requested logical CHAR/VARCHAR schema and data survive that path.
##########
sql/api/src/main/scala/org/apache/spark/sql/types/StructField.scala:
##########
@@ -96,18 +98,40 @@ 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
+ 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. */
Review Comment:
Not included per the scope freeze. Supporting CHAR/VARCHAR inside
UDT.sqlType requires a separate UDT and Arrow serialization design. Commit
08550e37485 removes the recursive UDT unwrapping and the UDT-in-CHAR tests;
this remains follow-up work.
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -223,7 +227,14 @@ 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
Review Comment:
Addressed in 08550e37485. Local-data physicalization no longer unwraps
UserDefinedType; it only lowers ordinary CHAR/VARCHAR leaves. The focused Arrow
encoder test verifies that an existing UDT wrapper remains unchanged. This
restores the pre-PR boundary without adding new JVM UDT transport support.
##########
python/pyspark/sql/tests/connect/test_connect_basic.py:
##########
@@ -491,17 +514,246 @@ def test_schema(self):
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)
connect_df = self.connect.sql(query)
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",
+ )
+
+ def test_create_dataframe_with_udt_char_schema(self):
+ schema = StructType([StructField("value", CharValueUDT())])
+ rows = [(CharValue("ab"),)]
+
+ standard_conf = {
+ "spark.sql.charVarchar.standardSemantics.enabled": "true",
+ "spark.sql.legacy.charVarcharAsString": "false",
+ }
+ with self.both_conf(standard_conf):
+ populated = self.connect.createDataFrame(rows, schema)
+ self.assertEqual(populated.schema, schema)
+ self.assertEqual(populated.collect(), [Row(value=CharValue("ab
"))])
+ self.assertEqual(self.connect.createDataFrame([], schema).schema,
schema)
+
+ legacy_conf = {
+ "spark.sql.charVarchar.standardSemantics.enabled": "false",
+ "spark.sql.legacy.charVarcharAsString": "true",
+ }
+ with self.both_conf(legacy_conf):
+ expected = StructType([StructField("value",
StringType("UTF8_LCASE"))])
+ populated = self.connect.createDataFrame(rows, schema)
+ self.assertEqual(populated.schema, expected)
+ self.assertEqual(populated.collect(), [Row(value="ab")])
+ self.assertEqual(self.connect.createDataFrame([], schema).schema,
expected)
+
+ 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",
+ )
+
+ def test_create_dataframe_with_explicit_binary_string_schema(self):
+ explicit_binary = StringType("UTF8_BINARY")
+ schema = StructType(
+ [
+ StructField("implicit", StringType()),
+ StructField("s", explicit_binary),
+ StructField(
+ "nested",
+ StructType(
+ [
+ StructField("s", explicit_binary),
+ StructField("a", ArrayType(explicit_binary)),
+ StructField(
+ "m",
+ MapType(explicit_binary, explicit_binary),
+ ),
+ ]
+ ),
+ ),
+ ]
+ )
+ rows = [
+ (
+ "implicit",
+ "direct",
+ Row(s="nested", a=["array"], m={"key": "value"}),
+ )
+ ]
+
+ populated = self.connect.createDataFrame(rows, schema)
Review Comment:
Not included per the scope freeze. Commit 08550e37485 removes the explicit
UTF8_BINARY StringType identity coverage from this PR. Ordinary STRING behavior
remains as it was before this branch; nested identity work is a follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
