Hyukjin Kwon created SPARK-60739:
------------------------------------

             Summary: Fix truncated results and unreleased executions in the 
Rust client's reattachable execution, and four drop-in parity gaps
                 Key: SPARK-60739
                 URL: https://issues.apache.org/jira/browse/SPARK-60739
             Project: Spark
          Issue Type: Bug
          Components: Connect
    Affects Versions: connect-rust-4.3.0
            Reporter: Hyukjin Kwon
             Fix For: connect-rust-4.3.0


The Rust client's reattachable execution and four drop-in APIs do not match the 
reference PySpark client (Spark 4.3.0). The official parity tests run the 
reference pyspark over the Rust transport, so they do not exercise these.

*Reattachable execution* (spark-connect-core, used by every DataFrame action of 
the native API and the drop-in):
- The server completes the response stream of a reattachable execution before 
the result is complete, on purpose, after 
spark.connect.execute.reattachable.senderMaxStreamDuration (2m) or 
senderMaxStreamSize (1g); the client must reattach to read the rest. The Rust 
client takes the end of the stream for the end of the result, so it silently 
returns a truncated result: with a 1 MiB senderMaxStreamSize, collect() of 
range(300000) returns 135000 rows, and a 3-second query on a server with a 1s 
senderMaxStreamDuration returns no rows and count() = 0.
- It never sends ReleaseExecute. The server keeps every finished execution, 
with its request and up to 10 MiB of its responses, until 
spark.connect.execute.manager.detachedTimeout (5m) or until the session closes. 
With large local relations this exhausted a 1 GB server heap: spark.stop() 
failed with "Java heap space".

*Drop-in API parity*:
- StructType.add() does not update the DataType that mapInArrow/mapInPandas 
(and the grouped applyIn* functions) resolve their schema from, so 
df.mapInArrow(f, StructType().add("a", "long").add("b", "long")) has no 
columns; str() of such a StructType is "struct<>".
- DataFrame.repartitionById(n, "col") takes the column name for a string 
literal, and fails analysis.
- RuntimeConf.get(key, None) raises SQL_CONF_NOT_FOUND for an unset key instead 
of returning None.
- SparkSession.createDataFrame(pyarrow.Table) raises TypeError.




--
This message was sent by Atlassian Jira
(v10.3.26#10030026)

---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to