cloud-fan commented on code in PR #58581:
URL: https://github.com/apache/spark/pull/58581#discussion_r4010878638
##########
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:
**Blocking (P1):** A non-empty Python Connect `createDataFrame` still passes
this logical `CharType`/`VarcharType` schema directly to Arrow, whose type
mapping handles `StringType` but not these sibling `AtomicType`s. I reproduced
`UNSUPPORTED_DATA_TYPE_FOR_ARROW_CONVERSION` with `CharType(4, "UTF8_LCASE")`,
before the request reaches the server; the same empty input bypasses Arrow, so
the server-policy behavior diverges by row count. The new Python test only
inspects a SQL-produced result schema and cannot catch this path. Please mirror
the JVM physical-STRING/logical-schema separation for Python local data and
cover non-empty caller-provided constrained strings across the server policy
modes.
**Recommended change:** Mirror the JVM client's physical-versus-logical
schema separation in Python Connect: recursively replace caller-provided
CHAR/VARCHAR leaves with collation-preserving STRING leaves only for local
Arrow conversion, retain the original StructType on the LocalRelation, and add
Python Connect createDataFrame coverage for non-empty caller-provided
constrained strings and the server policy outcomes.
**Why this works:** Build Arrow from raw string values under a physical
STRING schema so to_arrow_schema never receives CharType or VarcharType.
Serialize the original requested schema separately so existing server
reconciliation applies standard padding and overflow, legacy-as-string
normalization, or default rejection consistently for empty and populated input.
**Scope:** python/pyspark/sql/connect, python/pyspark/sql/tests/connect
**Compatibility:** Existing Python result-schema decoding, uncollated schema
handling, and the server's three-mode CHAR/VARCHAR policy remain unchanged.
**Risks:** A shallow transform could miss nested array, map, or struct
leaves. Dropping the collation on the physical STRING leaf could change string
interpretation before server reconciliation. Using the physical schema as the
LocalRelation's requested schema would silently normalize standard-semantics
input to STRING.
**Constraints:** Do not change CharType or VarcharType inheritance. Do not
pad, truncate, or length-check values on the client before server policy
selection. Preserve nested nullability, field metadata, collation, and the
original requested logical schema. Keep empty and populated caller data on the
same server-side policy matrix.
**Success:** A non-empty Python Connect createDataFrame with collated
CHAR/VARCHAR reaches the server instead of raising
UNSUPPORTED_DATA_TYPE_FOR_ARROW_CONVERSION. Standard semantics preserves the
logical constrained schema and applies existing padding and overflow behavior.
Legacy-as-string returns collation-preserving STRING schema and unpadded raw
values. The default unsupported mode rejects empty and populated inputs
consistently. Nested constrained leaves and omitted, explicit UTF8_BINARY, and
non-binary collations retain their intended state.
##########
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:
**Blocking (P1):** Adding `CharType`/`VarcharType` to this metadata path
emits, for example, type `char(4)` plus a `__COLLATIONS` entry. The immediately
preceding JVM `DataType.fromJson` sees that reserved entry before parsing the
type and rejects anything other than literal `string` with
`INVALID_JSON_DATA_TYPE_FOR_COLLATIONS`. As a result, schema JSON written by
this version cannot be read during a JVM downgrade or mixed-version exchange,
despite the nearby backward-compatibility contract. Please use a
constrained-type restoration representation that the older JVM reader treats as
ignorable metadata, while current Scala and Python readers still restore the
exact length and optional collation.
**Recommended change:** Encode CHAR/VARCHAR restoration data in a dedicated
reserved metadata form that older readers treat as unknown metadata while
keeping plain char(n)/varchar(n) type text. Teach current Scala and Python
readers and writers to consume and strip that form, retain the existing
__COLLATIONS behavior for StringType, and add cross-version compatibility
fixtures plus current round-trip and invalid-encoding coverage.
**Why this works:** An older JVM reader ignores an unknown metadata key and
parses the uncollated constrained type, while current readers recognize the
dedicated key and restore optional collation and length. Separating namespaces
avoids triggering the released __COLLATIONS validator without weakening its
type checks.
**Scope:** sql/api/src/main/scala/org/apache/spark/sql/types,
sql/catalyst/src/test/scala/org/apache/spark/sql/types, python/pyspark/sql
**Compatibility:** Current StringType compatibility serialization and
same-version constrained-type collation semantics remain intact while older
readers degrade only the newly unsupported collation attribute.
**Risks:** Older readers will retain the unknown reserved key as ordinary
field metadata unless the compatibility contract deliberately accounts for
that. Scala and Python could diverge on nested path syntax, provider
qualification, or omission versus explicit UTF8_BINARY. Accepting both old and
new restoration forms without duplicate validation could reintroduce ambiguous
dual encodings.
**Constraints:** Do not change the existing StringType __COLLATIONS JSON
shape or behavior. Preserve CHAR/VARCHAR length, nested paths, provider
validation, and the distinction between absent and explicit UTF8_BINARY
collation in current readers. Reject ambiguous inline-plus-metadata and
duplicate restoration encodings. Ensure the immediately preceding JVM parser
can consume the emitted document as uncollated char(n)/varchar(n).
**Success:** The immediately preceding JVM Spark reader parses newly emitted
collated CHAR/VARCHAR schema JSON without error and obtains the historical
uncollated constrained type. Current Scala and Python readers round-trip direct
and nested CHAR/VARCHAR lengths and optional collations exactly. Existing
collated StringType documents remain byte-shape compatible and preserve their
current behavior. Malformed, conflicting, or dual restoration encodings still
fail explicitly.
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -223,7 +223,9 @@ class SparkSession private[sql] (
/** @inheritdoc */
def createDataFrame(rows: java.util.List[Row], schema: StructType):
DataFrame = {
- createDataset(RowEncoder.encoderFor(schema),
rows.iterator().asScala).toDF()
+ // The client cannot observe the server's CHAR/VARCHAR configuration.
Encode an explicitly
+ // provided schema independently of the client's local configuration, as
for result schemas.
+ createDataset(RowEncoder.encoderForResultSchema(schema),
rows.iterator().asScala).toDF()
Review Comment:
Confirmed. Empty and populated inputs now share server-side reconciliation
across standard, legacy-as-string, and default modes, including unpadded STRING
output in legacy mode. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3960045571","thread_id":"inline:3960045571","verdict_sha256":"ae8dcffe4c707be0bb2e2c3138a8182a2db8f6643f48323831dc0886612431ac"}
-->
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -223,7 +223,9 @@ class SparkSession private[sql] (
/** @inheritdoc */
def createDataFrame(rows: java.util.List[Row], schema: StructType):
DataFrame = {
- createDataset(RowEncoder.encoderFor(schema),
rows.iterator().asScala).toDF()
+ // The client cannot observe the server's CHAR/VARCHAR configuration.
Encode an explicitly
Review Comment:
Confirmed. The comment now limits the client-local configuration distinction
to CHAR/VARCHAR semantics and no longer implies that other RowEncoder settings
are overridden. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3962398470","thread_id":"inline:3962398470","verdict_sha256":"ae8dcffe4c707be0bb2e2c3138a8182a2db8f6643f48323831dc0886612431ac"}
-->
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -223,7 +223,10 @@ class SparkSession private[sql] (
/** @inheritdoc */
def createDataFrame(rows: java.util.List[Row], schema: StructType):
DataFrame = {
- createDataset(RowEncoder.encoderFor(schema),
rows.iterator().asScala).toDF()
+ // RowEncoder consults the client process's local SqlApiConf, which can
differ from the
+ // server-side configuration visible through SparkSession.conf. Encode an
explicitly provided
+ // schema independently of that local configuration, as for result schemas.
+ createDataset(RowEncoder.encoderForResultSchema(schema),
rows.iterator().asScala).toDF()
Review Comment:
Confirmed. Explicit Row input now stays as raw collation-preserving STRING
data until server reconciliation, and the final schema and values match
standard, legacy-as-string, and default modes for empty and populated input.
Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3992292084","thread_id":"inline:3992292084","verdict_sha256":"ae8dcffe4c707be0bb2e2c3138a8182a2db8f6643f48323831dc0886612431ac"}
-->
##########
python/pyspark/sql/types.py:
##########
@@ -330,19 +330,31 @@ class CharType(AtomicType):
----------
length : int
the length limitation.
+ collation : str, optional
Review Comment:
Confirmed. Both constructor docs now clearly distinguish the default None
value from an explicit UTF8_BINARY declaration. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3992292095","thread_id":"inline:3992292095","verdict_sha256":"ae8dcffe4c707be0bb2e2c3138a8182a2db8f6643f48323831dc0886612431ac"}
-->
##########
python/pyspark/sql/types.py:
##########
@@ -2671,7 +2708,12 @@ def _parse_datatype_json_value( # type: ignore[return]
def _assert_valid_type_for_collation(
fieldPath: str, fieldType: Any, collationMap: Dict[str, str]
) -> None:
- if fieldPath in collationMap and fieldType != "string":
+ is_string_type = (
+ fieldType == "string"
+ or (isinstance(fieldType, str) and _LENGTH_CHAR.fullmatch(fieldType)
is not None)
Review Comment:
Confirmed. Restoration metadata is now accepted only with uncollated
char(n)/varchar(n) text, and both matching and conflicting dual encodings are
rejected by the new tests. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3992292102","thread_id":"inline:3992292102","verdict_sha256":"ae8dcffe4c707be0bb2e2c3138a8182a2db8f6643f48323831dc0886612431ac"}
-->
--
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]
