cloud-fan commented on code in PR #58581:
URL: https://github.com/apache/spark/pull/58581#discussion_r4040154037
##########
python/pyspark/sql/types.py:
##########
@@ -2429,8 +2493,8 @@ def fromWKB(cls, wkb: bytes, srid: int) -> "Geometry":
"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+))?")
Review Comment:
**Non-blocking (P2):** These patterns are still consumed with `match()`, so
`char(4) collate UTF8_LCASE junk` captures the length and collation and
silently ignores the trailing token. Scala's whole-string extractor rejects the
same JSON type name. Please require a complete match for both CHAR and VARCHAR
and add focused trailing-token cases.
##########
python/pyspark/sql/types.py:
##########
@@ -324,47 +324,82 @@ def isUTF8BinaryCollation(self) -> bool:
class CharType(AtomicType):
- """Char data type
+ """Char data type.
+
+ A standalone collated ``CharType`` writes its collation inline in JSON and
therefore requires
Review Comment:
**Nit (P3):** The immediately preceding Python reader does not require a
current reader here: it prefix-matches `char(n)` or `varchar(n)`, accepts the
value, and silently drops the collation suffix. Please describe that lossy
acceptance explicitly so users do not mistake successful parsing by an older
Python reader for collation preservation.
##########
sql/api/src/main/scala/org/apache/spark/sql/types/DataType.scala:
##########
@@ -425,12 +468,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)
Review Comment:
**Non-blocking (P2):** When this restoration map targets a field whose JSON
type is either UDT object form, both Scala UDT branches return before
`assertValidTypeForCharVarcharCollations` runs. `parseStructField` then removes
the key, so the JVM silently consumes invalid or caller-owned metadata while
Python rejects the same schema. Please validate the outer-field target before
either UDT arm can return; this does not require traversing `UDT.sqlType`.
##########
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:
Confirmed. JVM local-data physicalization now lowers ordinary CHAR/VARCHAR
without unwrapping UDTs, and the focused encoder test protects the restored
boundary. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4037004940","thread_id":"inline:4037004940","verdict_sha256":"97b99468d98c3679edc8479bbc2a570c93408cead229f9aa41bdc01830abc558"}
-->
##########
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:
Agreed. CHAR/VARCHAR inside UDT.sqlType is outside this PR's frozen scope
now that the recursive UDT changes and tests are gone; it should remain
follow-up work.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4037004957","thread_id":"inline:4037004957","verdict_sha256":"97b99468d98c3679edc8479bbc2a570c93408cead229f9aa41bdc01830abc558"}
-->
##########
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:
Confirmed. Both writers now reject a caller-owned __CHAR_VARCHAR_COLLATIONS
key before serialization, removing the silent overwrite, duplication, and
stripping paths under the frozen collision-safety contract. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4037004966","thread_id":"inline:4037004966","verdict_sha256":"97b99468d98c3679edc8479bbc2a570c93408cead229f9aa41bdc01830abc558"}
-->
##########
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:
Confirmed. Python now validates the restoration entry before atomic parsing,
including decimal, so malformed non-CHAR/VARCHAR targets are rejected
consistently with Scala. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4037004975","thread_id":"inline:4037004975","verdict_sha256":"97b99468d98c3679edc8479bbc2a570c93408cead229f9aa41bdc01830abc558"}
-->
##########
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:
Confirmed. The forced cached-path regression preserves the requested logical
CHAR/VARCHAR schema separately from Arrow data and verifies both schema and
values. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4037004984","thread_id":"inline:4037004984","verdict_sha256":"97b99468d98c3679edc8479bbc2a570c93408cead229f9aa41bdc01830abc558"}
-->
--
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]
