Hyukjin Kwon created SPARK-60898:
------------------------------------
Summary: Keep plan IDs stable across re-executions in Spark
Connect Rust to reuse the server plan cache
Key: SPARK-60898
URL: https://issues.apache.org/jira/browse/SPARK-60898
Project: Spark
Issue Type: Improvement
Components: Connect
Affects Versions: connect-rust-4.3.0
Reporter: Hyukjin Kwon
In the SPARK-60675 benchmark
(https://github.com/apache/spark-connect-rust/pull/118), warmed aggregations
over a 136 MB cached Arrow input took 3043.90 ms (median) with the Rust client
and 2042.50 ms with PySpark 4.2.0, against the same local Spark Connect 4.2.0
server: about 49% slower. Omitting the inferred schema from the local relation
did not close the gap.
Hypothesis: the Rust client never hits the server's plan cache.
* The Rust client assigns plan IDs while it builds each request.
{{build_execute_request}} and {{build_analyze_request}} call
{{assign_plan_ids}}, which gives every relation, including the root, a fresh
{{SparkSession::next_plan_id()}}. Re-executing the same DataFrame therefore
sends a different relation each time.
* PySpark assigns plan IDs once, in {{LogicalPlan.__init__}}, so re-executing a
DataFrame sends an identical relation.
* Spark 4.2.0 and 4.3.0 cache resolved plans per session in
{{SessionHolder.usePlanCache}}. The key is the full {{proto.Relation}}, and
only relations that carry a plan ID are cached. ExecutePlan and AnalyzePlan
resolve their root with {{cachePlan = true}}.
{{spark.connect.session.planCache.maxSize}} defaults to 32, and
{{spark.connect.session.planCache.enabled}} to true.
A cache hit skips resolving the relation again. For a
{{ChunkedCachedLocalRelation}}, resolving it includes reading and decoding the
cached Arrow chunks. How much of the gap this explains has not been measured.
Proposed work:
# Rerun the warmed-query benchmark with plan IDs fixed per DataFrame, and
confirm cache hits in the server's debug log ("Using cached plan for relation").
# If confirmed, assign plan IDs once when the plan is built, as PySpark does,
so that re-executing a DataFrame sends an identical relation.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
