This is an automated email from the ASF dual-hosted git repository.
zhengruifeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 7ce639e04383 [SPARK-58846][CONNECT][PYTHON] Add a client-side deadline
to the RemoveRemoteCachedRelation release RPC
7ce639e04383 is described below
commit 7ce639e04383074f9e2458054a14fcf6331981b1
Author: kishor-rkrishnan <[email protected]>
AuthorDate: Mon Aug 24 16:35:30 2026 +0800
[SPARK-58846][CONNECT][PYTHON] Add a client-side deadline to the
RemoveRemoteCachedRelation release RPC
### What changes were proposed in this pull request?
`CachedRemoteRelation.__del__` (`python/pyspark/sql/connect/plan.py`)
releases a server-cached
DataFrame by issuing a blocking, non-reattachable `ExecutePlan` RPC (a
`RemoveRemoteCachedRelation`
command) through a raw `unary_unary` channel, with **no client-side
deadline**. This PR bounds that
call:
- Add a `release_relation` field to `RpcDeadlines`
(`python/pyspark/sql/connect/client/core.py`),
default `60` (seconds), and include it in `RpcDeadlines.disabled()`.
- Pass it as the gRPC `timeout` on the release call in `plan.py`.
- Document in the `RpcDeadlines` docstring why this non-reattachable
`ExecutePlan` call *is* given a
deadline (unlike a query `ExecutePlan`): a timeout here abandons only a
best-effort cache eviction
that the server performs independently, so nothing recoverable is killed.
No wire-protocol change; the change is entirely client-local.
`DefaultPolicy.can_retry` already
returns `False` for `DEADLINE_EXCEEDED` on non-reattachable RPCs, so on
timeout the release call is
not retried — it raises, is caught by the existing `except Exception`,
logged, and the finalizer
proceeds.
### Why are the changes needed?
Without a deadline this finalizer can block forever if the release response
is never delivered. On
the foreachBatch Spark Connect path it runs on the per-batch critical path:
the Python worker
releases the batch DataFrame (firing this finalizer) *before* it writes its
completion signal, so a
blocked release means the signal is never sent and the driver JVM
(`StreamingForeachBatchHelper`)
stays blocked on its per-batch `dataIn.readInt()` — which also has no
timeout. The streaming query
then stalls indefinitely with no error surfaced (observed as multi-hour
hangs), recovering only on
a manual restart. A bounded deadline converts this indefinite hang into a
bounded wait after which
the worker proceeds.
### Does this PR introduce _any_ user-facing change?
Yes — a behavior change, defaulted to preserve safety:
- The `RemoveRemoteCachedRelation` release RPC now has a default
client-side deadline of 60s (via
the new `RpcDeadlines.release_relation`). Previously it had no deadline
and could block forever if
the response was never delivered.
- On timeout, the eviction is abandoned (the existing warning is logged)
and execution continues.
The eviction is best effort and is also performed server-side, so this
does not leak cached state
in normal operation (and on the foreachBatch path the JVM `finally`
evicts the same relation id).
- Configurable/opt-out via `RpcDeadlines(release_relation=<seconds>)` or
`RpcDeadlines.disabled()`.
### How was this patch tested?
- New unit test
`pyspark.sql.tests.connect.client.test_client.SparkConnectClientTestCase.test_remove_cached_relation_uses_release_relation_deadline`:
triggers `CachedRemoteRelation.__del__` against a captured channel and
asserts (a) the configured
`release_relation` deadline is forwarded as the gRPC `timeout`, and (b)
`RpcDeadlines.disabled()`
forwards `None`.
- Existing `test_each_rpc_receives_configured_deadline` still passes,
confirming the `disabled()`
and per-RPC forwarding paths remain consistent with the new field.
- Standalone end-to-end reproduction (no cluster/build): an in-process
`SparkConnectService` whose
`ExecutePlan` never responds, with a real `SparkConnectClient` pointed at
it. With no deadline the
finalizer blocks indefinitely; with a `release_relation` deadline it
returns within the deadline
(returning in exactly one deadline interval — no retry, since
`DEADLINE_EXCEEDED` is not retried
on this path).
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with : Claude Code
Closes #58085 from kishor-rkrishnan/spark-connect-release-relation-deadline.
Lead-authored-by: kishor-rkrishnan
<[email protected]>
Co-authored-by: Kishor Radhakrishnan <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>
---
python/pyspark/sql/connect/client/core.py | 10 ++++++
python/pyspark/sql/connect/plan.py | 10 +++++-
.../sql/tests/connect/client/test_client.py | 42 ++++++++++++++++++++++
3 files changed, 61 insertions(+), 1 deletion(-)
diff --git a/python/pyspark/sql/connect/client/core.py
b/python/pyspark/sql/connect/client/core.py
index 1712a9fb41b8..9c036aa6810d 100644
--- a/python/pyspark/sql/connect/client/core.py
+++ b/python/pyspark/sql/connect/client/core.py
@@ -146,6 +146,14 @@ class RpcDeadlines:
fires, the server-side operation continues running; the client opens a new
ReattachExecute
stream to resume receiving results. Non-reattachable ExecutePlan has no
deadline because a
timeout there would kill the execution with no recovery path.
+
+ Note on ``release_relation``: the RemoveRemoteCachedRelation cleanup
command is sent over a
+ blocking, non-reattachable ExecutePlan call issued from
+ :meth:`CachedRemoteRelation.__del__`. Unlike a query ExecutePlan, a
timeout here does not kill
+ any recoverable execution -- it only abandons a best-effort cache eviction
that the server also
+ performs independently -- so this call is given a bounded deadline.
Without it the finalizer
+ can block forever if the release response is never delivered, which (on
the foreachBatch
+ Connect path) stalls the streaming query indefinitely.
"""
reattachable_execute_plan: Optional[float] = 10 * 60 # 10 min
@@ -155,6 +163,7 @@ class RpcDeadlines:
config: Optional[float] = 10 * 60 # 10 min
interrupt: Optional[float] = 10 * 60 # 10 min
release_session: Optional[float] = 10 * 60 # 10 min
+ release_relation: Optional[float] = 60 # 1 min; short: per-batch finalizer
artifact_status: Optional[float] = 10 * 60 # 10 min
clone_session: Optional[float] = 10 * 60 # 10 min
get_status: Optional[float] = 10 * 60 # 10 min
@@ -185,6 +194,7 @@ class RpcDeadlines:
config=None,
interrupt=None,
release_session=None,
+ release_relation=None,
artifact_status=None,
clone_session=None,
get_status=None,
diff --git a/python/pyspark/sql/connect/plan.py
b/python/pyspark/sql/connect/plan.py
index 709071b8acc2..1efd02551da7 100644
--- a/python/pyspark/sql/connect/plan.py
+++ b/python/pyspark/sql/connect/plan.py
@@ -785,7 +785,15 @@ class CachedRemoteRelation(LogicalPlan):
response_deserializer=response_deserializer,
)
metadata =
session.client._execute_plan_metadata(req.operation_id)
- channel(req, metadata=metadata) # type:
ignore[arg-type]
+ # Bound this blocking call with a client-side
deadline. It is issued from a
+ # finalizer with no other timeout at any layer;
without a deadline it can
+ # block forever if the response is never delivered,
which stalls the
+ # foreachBatch Connect handshake (the Python worker
never sends its
+ # completion signal and the driver JVM blocks on the
per-batch read). A
+ # timeout here is non-fatal: the eviction is best
effort and the server
+ # performs it independently, so on timeout we log and
move on.
+ timeout =
session.client._rpc_deadlines.release_relation
+ channel(req, metadata=metadata, timeout=timeout) #
type: ignore[arg-type]
except Exception as e:
logger.warning(f"RemoveRemoteCachedRelation failed with
exception: {e}.")
diff --git a/python/pyspark/sql/tests/connect/client/test_client.py
b/python/pyspark/sql/tests/connect/client/test_client.py
index 5e24bd686517..9a6ac163cc28 100644
--- a/python/pyspark/sql/tests/connect/client/test_client.py
+++ b/python/pyspark/sql/tests/connect/client/test_client.py
@@ -967,6 +967,48 @@ class SparkConnectClientTestCase(unittest.TestCase):
client._get_operation_statuses()
self.assertEqual(mock.captured_timeouts["GetStatus"], 55.0)
+ def test_remove_cached_relation_uses_release_relation_deadline(self):
+ """CachedRemoteRelation.__del__ must bound its release RPC with the
release_relation
+ deadline.
+
+ Regression test: the RemoveRemoteCachedRelation cleanup is a blocking,
non-reattachable
+ ExecutePlan call issued from a finalizer with no other timeout at any
layer. Without a
+ client-side deadline it can hang forever if the response is never
delivered, which on the
+ foreachBatch Connect path stalls the streaming query indefinitely (the
Python worker never
+ sends its completion signal and the driver JVM blocks on the per-batch
read).
+ """
+ from types import SimpleNamespace
+ from unittest.mock import MagicMock
+ from pyspark.sql.connect.plan import CachedRemoteRelation
+
+ def make_relation(deadlines):
+ client = SparkConnectClient(
+ "sc://foo/",
+ use_reattachable_execute=False,
+ rpc_deadlines=deadlines,
+ retry_policy=dict(max_retries=0),
+ )
+ captured = {}
+
+ def fake_call(req, metadata=None, timeout="unset"):
+ captured["timeout"] = timeout
+ return proto.ExecutePlanResponse()
+
+ client._channel = MagicMock()
+ client._channel.unary_unary.return_value = fake_call
+ rel = CachedRemoteRelation("df-id-123",
spark_session=SimpleNamespace(client=client))
+ return rel, captured
+
+ # A configured deadline is forwarded as the gRPC call timeout.
+ rel, captured = make_relation(RpcDeadlines(release_relation=44.0))
+ rel.__del__()
+ self.assertEqual(captured["timeout"], 44.0)
+
+ # Disabled deadlines forward None (opt out; rely on
transport/server-side timeouts).
+ rel, captured = make_relation(RpcDeadlines.disabled())
+ rel.__del__()
+ self.assertIsNone(captured["timeout"])
+
@unittest.skipIf(not should_test_connect, connect_requirement_message)
class SparkConnectClientReattachTestCase(unittest.TestCase):
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
