cloud-fan commented on code in PR #58581:
URL: https://github.com/apache/spark/pull/58581#discussion_r4037004940
##########
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:
**Blocking (P1):** This physical schema also unwraps ordinary JVM UDTs, but
the row iterator still contains their user objects. The previous logical-schema
RowEncoder selected UDTEncoder and called serialize; encoderForResultSchema on
the unwrapped sqlType instead expects the storage value directly, so a
non-empty createDataFrame with a normal JVM UDT can fail before reaching the
server. Could this conversion preserve exactly-once UDT serialization before
handing physical values to Arrow, with a focused ordinary-UDT object regression
case?
**Recommended change:** Preserve ordinary UDT serialization while lowering
CHAR/VARCHAR and UDT storage types for JVM Connect local-data transport, with
focused JVM UDT regression coverage.
**Why this works:** Derive an encoder/conversion plan that still observes
the requested UDT wrappers and invokes each UDT serializer exactly once, then
produces values conforming to the lowered Arrow schema instead of asking a
physical-schema encoder to consume the original user objects.
**Scope:** Separate external Row conversion from Arrow physical-schema
lowering in JVM Connect local-data creation.
**Compatibility:** Existing CHAR/VARCHAR policy handling and
PythonUserDefinedType storage-value input remain unchanged.
**Risks:** A conversion applied at both the encoder and Arrow boundary could
serialize a UDT twice. PythonUserDefinedType inputs that already contain
storage values must retain their current behavior.
**Constraints:** The requested logical schema must remain available to the
server. The values handed to Arrow must match the lowered physical schema.
**Success:** An ordinary JVM UDT user object accepted by the supplied schema
is serialized once and reaches Arrow as its SQL storage value. CHAR/VARCHAR
lowering and PythonUserDefinedType local-data behavior remain intact.
##########
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:
**Blocking (P1):** The compatibility traversal still stops when the field
type is a UDT. For the added CharValueUDT, its sqlType is emitted as inline
`char(4) collate ...` with no `__CHAR_VARCHAR_COLLATIONS` entry; a preceding
JVM reader then parses that sqlType through the old Python-UDT path and rejects
it instead of degrading to uncollated `char(4)`. Please collect and restore
constrained-string collations through UDT.sqlType in both Scala and Python, and
cover the preceding-reader shape.
See **Shared repair plan 1** in the review body.
##########
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:
**Non-blocking (P2):** `__CHAR_VARCHAR_COLLATIONS` was not previously
reserved from StructField metadata, but current readers always consume and
remove it. A field whose caller metadata already contains this key therefore
loses that value even for IntegerType; when generated collation metadata is
also present, Python overwrites it and Scala can emit duplicate members. Please
make the compatibility encoding collision-safe and verify that a pre-existing
caller value round-trips.
See **Shared repair plan 1** in the review body.
##########
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:
**Non-blocking (P2):** The requested schema now travels separately in the
cached/multi-batch path, but the focused logical-versus-physical schema cases
never force this branch. A regression that writes `encoder.schema`, drops the
first schema chunk, or rejoins it incorrectly would leave the current empty and
one-row tests green. Please force a differing requested/Arrow schema through
cached or multi-batch transport and assert that the logical schema survives.
##########
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:
**Non-blocking (P2):** The new restoration-map validation is branch-local,
so earlier regex cases can return without checking it. For example,
`decimal(10,2)` with a `__CHAR_VARCHAR_COLLATIONS` entry is accepted and the
metadata is stripped here, while the JVM rejects that encoding as targeting a
non-CHAR/VARCHAR leaf. Could the metadata/type pairing be validated before any
atomic parser return so Scala and Python reject the same malformed schemas?
See **Shared repair plan 1** in the review body.
##########
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:
**Non-blocking (P2):** This non-row branch now has its own
CHAR/VARCHAR-to-Arrow schema conversion, while the new end-to-end policy
coverage exercises only row collections. A pandas or pyarrow.Table path can
still fail client-side or accidentally transport the physical schema as
requested without affecting those tests. Please add constrained-string
createDataFrame cases for both pandas.DataFrame and pyarrow.Table that verify
Arrow encoding and server-side logical-schema policy handling.
##########
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:
**Non-blocking (P2):** Using a plain object sentinel changes the published
constructor signature from `collation='UTF8_BINARY'` to an opaque
address-bearing default in introspection and generated API docs. The omission
marker is useful for explicitness, but could it have a stable string-like
public representation so omitted and explicit UTF8_BINARY remain
distinguishable internally without changing the documented signature?
##########
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:
**Non-blocking (P2):** The broad StructType equality here does not observe
`_collation_explicit`, and only the top-level leaf gets an identity-sensitive
assertion. Returning implicit UTF8_BINARY from any nested struct, array,
map-key, or map-value restoration arm would still pass. Please inspect
explicitness at each changed nested arm so this test protects the state that
later compatibility serialization depends on.
##########
python/pyspark/sql/tests/connect/test_connect_basic.py:
##########
@@ -491,17 +491,28 @@ 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.
Review Comment:
Confirmed: Python Connect now lowers caller-provided CHAR/VARCHAR to
collation-preserving STRING for local Arrow conversion while retaining the
requested schema for server-side reconciliation.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4010878638","thread_id":"inline:4010878638","verdict_sha256":"0a61a499c58554b9add1cb9ff86029da1a2136fd17ae415ba19393ecd088428a"}
-->
##########
sql/api/src/main/scala/org/apache/spark/sql/types/StructField.scala:
##########
@@ -137,6 +137,8 @@ case class StructField(
}
private def isCollatedString(dt: DataType): Boolean = dt match {
+ case c: CharType => c.collation.isDefined
Review Comment:
Confirmed: direct and ordinary nested CHAR/VARCHAR schema JSON now uses
__CHAR_VARCHAR_COLLATIONS with uncollated constrained type text, while STRING
remains on __COLLATIONS.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4010878643","thread_id":"inline:4010878643","verdict_sha256":"0a61a499c58554b9add1cb9ff86029da1a2136fd17ae415ba19393ecd088428a"}
-->
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1641,9 +1647,35 @@ class SparkConnectPlanner(
.asInstanceOf[Project]
val proj = UnsafeProjection.create(project.projectList,
project.child.output)
- logical.LocalRelation(
- DataTypeUtils.toAttributes(schema),
- data.map(proj).map(_.copy()).toSeq)
+ def restoreFieldNames(actual: DataType, requested: DataType): DataType
=
+ (actual, requested) match {
+ case (_, requestedUdt: UserDefinedType[_]) => requestedUdt
+ case (StructType(actualFields), StructType(requestedFields)) =>
+ StructType(
+ actualFields.zip(requestedFields).map { case (actualField,
requestedField) =>
+ actualField.copy(
+ name = requestedField.name,
+ dataType = restoreFieldNames(actualField.dataType,
requestedField.dataType))
+ })
+ case (ArrayType(actualElement, containsNull),
ArrayType(requestedElement, _)) =>
+ ArrayType(restoreFieldNames(actualElement, requestedElement),
containsNull)
+ case (
+ MapType(actualKey, actualValue, valueContainsNull),
+ MapType(requestedKey, requestedValue, _)) =>
+ MapType(
+ restoreFieldNames(actualKey, requestedKey),
+ restoreFieldNames(actualValue, requestedValue),
+ valueContainsNull)
+ case _ => actual
+ }
+ val output = project.output.zip(schema.fields).map { case (attribute,
field) =>
+ AttributeReference(
+ field.name,
+ restoreFieldNames(attribute.dataType, field.dataType),
Review Comment:
Confirmed: local-relation reconstruction now keeps the requested explicit
UTF8_BINARY StringType when reconciliation has not selected a meaningfully
different leaf.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4015081820","thread_id":"inline:4015081820","verdict_sha256":"0a61a499c58554b9add1cb9ff86029da1a2136fd17ae415ba19393ecd088428a"}
-->
--
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]
