This is an automated email from the ASF dual-hosted git repository.
dtenedor 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 001e89cda589 [SPARK-58021][CONNECT] Add local server pool filesystem
storage
001e89cda589 is described below
commit 001e89cda589bc74a28765a8df746360e522878b
Author: Eric Marnadi <[email protected]>
AuthorDate: Wed Aug 12 10:13:00 2026 -0700
[SPARK-58021][CONNECT] Add local server pool filesystem storage
### What changes were proposed in this pull request?
This is layer 2 of the seven-PR local Connect pool stack:
#57684 -> #57685 -> #57907 -> #57686 -> #57687 -> #57102 -> #57688
The review unit introduced here is commit `23f806b64ad`.
This layer adds the filesystem-backed storage foundation for pool members:
- stable state-file paths keyed by member ID and per-member directories;
- an overridable private pool directory under the per-user runtime
directory;
- a per-pool cross-process POSIX file lock;
- private directory, lock-file, and JSON state-file permissions; and
- locked helpers for listing, reading, writing, renaming, and removing
member state.
Member validation, compatibility fingerprints, and atomic claiming are
isolated in #57907.
Process lifecycle, acquisition, SparkSession integration, and JIT warmup
remain in later PRs.
### Why are the changes needed?
The pool needs a small, independently reviewable state model before adding
compatibility checks,
claiming, and process supervision. Keeping this layer limited to path
layout, locking, and state
file access makes its filesystem and concurrency contract reviewable on its
own.
### Does this PR introduce _any_ user-facing change?
No. The storage model is internal and is not wired into SparkSession in
this layer.
### How was this patch tested?
Added three focused tests covering directory selection, private permissions
and malformed JSON,
and cross-process lock contention.
```bash
python/run-tests --testnames
pyspark.sql.tests.connect.test_connect_local_server_pool
```
These cases passed on Python 3.11 as part of the combined suite before the
stack was split. The
rebuilt commit passed `git diff --check`, Python AST parsing, and
changed-line ASCII and 100-column
checks.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5) and OpenAI Codex (GPT-5)
Closes #57685 from ericm-db/local-connect-pool-storage.
Authored-by: Eric Marnadi <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>
---
dev/sparktestsupport/modules.py | 1 +
python/pyspark/sql/connect/local_server_pool.py | 231 +++++++++++++++++++++
.../connect/test_connect_local_server_pool.py | 229 ++++++++++++++++++++
3 files changed, 461 insertions(+)
diff --git a/dev/sparktestsupport/modules.py b/dev/sparktestsupport/modules.py
index f272075525ef..f495634ce0d5 100644
--- a/dev/sparktestsupport/modules.py
+++ b/dev/sparktestsupport/modules.py
@@ -1200,6 +1200,7 @@ pyspark_connect = Module(
"pyspark.sql.tests.connect.test_connect_retry",
"pyspark.sql.tests.connect.test_connect_session",
"pyspark.sql.tests.connect.test_connect_local_server",
+ "pyspark.sql.tests.connect.test_connect_local_server_pool",
"pyspark.sql.tests.connect.test_connect_stat",
"pyspark.sql.tests.connect.test_parity_geographytype",
"pyspark.sql.tests.connect.test_parity_geometrytype",
diff --git a/python/pyspark/sql/connect/local_server_pool.py
b/python/pyspark/sql/connect/local_server_pool.py
new file mode 100644
index 000000000000..72b13d5b52f4
--- /dev/null
+++ b/python/pyspark/sql/connect/local_server_pool.py
@@ -0,0 +1,231 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Filesystem-backed state storage for local Spark Connect server pools.
+
+This internal foundation owns the pool directory layout, locking, and JSON
state-file access.
+"""
+
+import contextlib
+import json
+import os
+import shutil
+from typing import Any, Dict, List, Optional, Tuple
+
+
+class PoolDirectory:
+ """Path layout, file access, and the cross-process lock of one pool
directory.
+
+ Used as a context manager that holds the directory's exclusive lock:
+
+ directory = PoolDirectory()
+ with directory:
+ path = directory.pending_path(uid)
+ directory.write_json(path, data)
+ stored = directory.read_json(path)
+ directory.rename(path, directory.server_path(uid))
+
+ Entering the context creates the directory and acquires its lock. Callers
then use path
+ builders and the locked accessors to enumerate, read, write, rename, or
remove state. Exiting
+ the context releases the lock.
+
+ Pool operations are infrequent, so one exclusive lock for every state
transition is simpler
+ than a finer-grained scheme. A context can be entered again after exiting,
allowing callers
+ to release the lock between polling attempts so other processes can update
the directory.
+ """
+
+ def __init__(self, path: Optional[str] = None):
+ if path is None:
+ path = os.environ.get("SPARK_LOCAL_CONNECT_POOL_DIR")
+ if path is None:
+ from pyspark.sql.connect.local_server import runtime_dir
+
+ path = os.path.join(runtime_dir(), "pool")
+ self.path = os.path.abspath(path)
+ self._lock_fd: Optional[int] = None
+
+ def __enter__(self) -> "PoolDirectory":
+ import fcntl
+
+ # Not reentrant: a nested enter would os.open a second fd and
flock(LOCK_EX) would block
+ # forever against the fd this process already holds. Fail loudly
instead of deadlocking.
+ assert self._lock_fd is None, "PoolDirectory is not reentrant"
+ os.makedirs(self.path, mode=0o700, exist_ok=True)
+ # Re-assert privacy for an existing override directory: state files
contain auth tokens,
+ # and directory write access would allow replacing them or bypassing
the shared lock.
+ os.chmod(self.path, 0o700)
+ lock_fd = os.open(os.path.join(self.path, ".lock"), os.O_RDWR |
os.O_CREAT, 0o600)
+ try:
+ os.fchmod(lock_fd, 0o600)
+ fcntl.flock(lock_fd, fcntl.LOCK_EX)
+ except BaseException:
+ os.close(lock_fd)
+ raise
+ self._lock_fd = lock_fd
+ return self
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ assert self._lock_fd is not None
+ os.close(self._lock_fd) # closing releases the lock
+ self._lock_fd = None
+
+ def _assert_locked(self) -> None:
+ assert self._lock_fd is not None, "PoolDirectory must be used as a
context manager"
+
+ # Path builders; these do not touch the filesystem and need no lock.
+
+ def pending_path(self, uid: str) -> str:
+ return os.path.join(self.path, f"pending-{uid}.json")
+
+ def conf_path(self, uid: str) -> str:
+ return os.path.join(self.path, f"conf-{uid}.json")
+
+ def server_path(self, uid: str) -> str:
+ return os.path.join(self.path, f"server-{uid}.json")
+
+ def claimed_path(self, client_pid: int, uid: str) -> str:
+ return os.path.join(self.path, f"claimed-{client_pid}-{uid}.json")
+
+ def retired_path(self, uid: str) -> str:
+ return os.path.join(self.path, f"retired-{uid}.json")
+
+ def member_dir(self, uid: str) -> str:
+ return os.path.join(self.path, f"member-{uid}")
+
+ # uids are generated as ``uuid.uuid4().hex[:12]`` (see the acquisition
layer), so a valid
+ # uid is a nonempty run of lowercase hex. Validating the shape keeps
editor droppings such
+ # as ``member-abc.json.swp`` and empty stems like ``server-.json`` from
becoming phantom uids.
+ _UID_CHARS = frozenset("0123456789abcdef")
+
+ @classmethod
+ def _is_uid(cls, uid: str) -> bool:
+ return bool(uid) and all(c in cls._UID_CHARS for c in uid)
+
+ @classmethod
+ def _split_claimed(cls, stem: str) -> Optional[Tuple[str, str]]:
+ """Split a well-formed ``claimed-<pid>-<uid>`` stem (without the
``.json`` suffix) into
+ ``(client_pid, uid)`` as strings, or ``None`` otherwise. The pid is
returned unparsed:
+ ``parse_entry`` classifies over every directory entry and must never
raise, and
+ ``str.isdigit()`` accepts characters ``int()`` rejects (e.g.
superscripts), so the
+ ``isascii()`` guard keeps the eventual ``int()`` in ``claiming_pid``
total."""
+ if not stem.startswith("claimed-"):
+ return None
+ client_pid, sep, uid = stem[len("claimed-") :].partition("-")
+ if not sep or not (client_pid.isascii() and client_pid.isdigit()) or
not cls._is_uid(uid):
+ return None
+ return client_pid, uid
+
+ @classmethod
+ def parse_entry(cls, name: str) -> Tuple[Optional[str], Optional[str]]:
+ """The ``(kind, uid)`` of a pool directory entry, ``(None, None)`` for
anything
+ else (the lock file, editor droppings, entries with a malformed uid,
...)."""
+ if name.startswith("member-"):
+ uid = name[len("member-") :]
+ return ("member", uid) if cls._is_uid(uid) else (None, None)
+ if not name.endswith(".json"):
+ return None, None
+ stem = name[: -len(".json")]
+ for kind in ("pending", "conf", "server", "retired"):
+ if stem.startswith(kind + "-"):
+ uid = stem[len(kind) + 1 :]
+ return (kind, uid) if cls._is_uid(uid) else (None, None)
+ claimed = cls._split_claimed(stem)
+ return ("claimed", claimed[1]) if claimed is not None else (None, None)
+
+ @classmethod
+ def claiming_pid(cls, claimed_path: str) -> int:
+ """The client pid recorded in a ``claimed-<pid>-<uid>.json`` file
name."""
+ name = os.path.basename(claimed_path)
+ stem = name[: -len(".json")] if name.endswith(".json") else name
+ claimed = cls._split_claimed(stem)
+ assert claimed is not None, f"not a claimed entry: {claimed_path!r}"
+ return int(claimed[0])
+
+ # Locked accessors.
+
+ def uids(self) -> List[str]:
+ self._assert_locked()
+ seen = []
+ for name in self._entries():
+ _, uid = self.parse_entry(name)
+ if uid is not None and uid not in seen:
+ seen.append(uid)
+ return seen
+
+ def states(self, uid: str) -> Dict[str, str]:
+ """The state entries currently existing for ``uid``, as ``{kind:
path}`` with kinds
+ ``pending``, ``conf``, ``server``, ``claimed``, ``retired``, and
``member`` (the
+ member's directory)."""
+ self._assert_locked()
+ found: Dict[str, str] = {}
+ for name in self._entries():
+ kind, entry_uid = self.parse_entry(name)
+ if kind is not None and entry_uid == uid:
+ # At most one entry per kind. Claiming renames a single file
into place (see the
+ # claiming layer), so two claimed entries for one uid means
the pid a reaper would
+ # read via claiming_pid is ambiguous; surface that rather than
pick one silently.
+ assert kind not in found, f"duplicate {kind} entries for uid
{uid}"
+ found[kind] = os.path.join(self.path, name)
+ return found
+
+ def paths_of_kind(self, kind: str) -> List[Tuple[str, str]]:
+ """All ``(uid, path)`` of one state kind."""
+ self._assert_locked()
+ return [
+ (uid, os.path.join(self.path, name))
+ for name in self._entries()
+ for entry_kind, uid in (self.parse_entry(name),)
+ if entry_kind == kind and uid is not None
+ ]
+
+ def _entries(self) -> List[str]:
+ try:
+ return sorted(os.listdir(self.path))
+ except OSError:
+ return []
+
+ def read_json(self, path: str) -> Optional[Dict[str, Any]]:
+ """``None`` for files that are missing or unreadable -- callers treat
both like the
+ state not existing, and the reaping rules remove unreadable
leftovers."""
+ self._assert_locked()
+ try:
+ with open(path, "r") as f:
+ data = json.load(f)
+ except (OSError, ValueError):
+ return None
+ return data if isinstance(data, dict) else None
+
+ def write_json(self, path: str, data: Dict[str, Any]) -> None:
+ self._assert_locked()
+ # 0600 like the reuse discovery file: server entries hold the auth
token.
+ fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+ with os.fdopen(fd, "w") as f:
+ os.fchmod(fd, 0o600)
+ f.write(json.dumps(data))
+
+ def rename(self, src: str, dst: str) -> None:
+ self._assert_locked()
+ os.rename(src, dst)
+
+ def remove(self, path: str) -> None:
+ self._assert_locked()
+ with contextlib.suppress(FileNotFoundError):
+ os.remove(path)
+
+ def remove_member_dir(self, uid: str) -> None:
+ self._assert_locked()
+ shutil.rmtree(self.member_dir(uid), ignore_errors=True)
diff --git a/python/pyspark/sql/tests/connect/test_connect_local_server_pool.py
b/python/pyspark/sql/tests/connect/test_connect_local_server_pool.py
new file mode 100644
index 000000000000..7e8d2326fbf8
--- /dev/null
+++ b/python/pyspark/sql/tests/connect/test_connect_local_server_pool.py
@@ -0,0 +1,229 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from pyspark.testing.connectutils import should_test_connect,
connect_requirement_message
+
+if should_test_connect:
+ from pyspark.sql.connect.local_server_pool import PoolDirectory
+
+
+_SAVED_ENV_KEYS = ("SPARK_LOCAL_CONNECT_POOL_DIR",)
+
+
+# These tests start no server and exercise only stdlib filesystem code, so
they do not need JVM
+# access (no is_remote_only gate). should_test_connect is still required
because importing
+# PoolDirectory pulls in the pyspark.sql.connect package, which checks Connect
dependencies.
[email protected](
+ not should_test_connect,
+ connect_requirement_message or "Requires Spark Connect dependencies to
import the pool module",
+)
[email protected](os.name == "posix", "the pool relies on POSIX file locks")
+class LocalConnectServerPoolUnitTests(unittest.TestCase):
+ """Tests for the pool filesystem storage; no real servers are started."""
+
+ def setUp(self) -> None:
+ self._tmpdir = tempfile.mkdtemp()
+ self._saved_env = {k: os.environ.get(k) for k in _SAVED_ENV_KEYS}
+ for k in _SAVED_ENV_KEYS:
+ os.environ.pop(k, None)
+ os.environ["SPARK_LOCAL_CONNECT_POOL_DIR"] =
os.path.join(self._tmpdir, "pool")
+ self._directory = PoolDirectory()
+
+ def tearDown(self) -> None:
+ for k, v in self._saved_env.items():
+ if v is None:
+ os.environ.pop(k, None)
+ else:
+ os.environ[k] = v
+ shutil.rmtree(self._tmpdir, ignore_errors=True)
+
+ def test_pool_directory_location(self) -> None:
+ self.assertEqual(self._directory.path, os.path.join(self._tmpdir,
"pool"))
+ os.environ.pop("SPARK_LOCAL_CONNECT_POOL_DIR")
+ default = PoolDirectory()
+ self.assertEqual(os.path.basename(default.path), "pool")
+ self.assertTrue(default.path.startswith(tempfile.gettempdir()))
+
+ def test_pool_directory_lock_and_state_file_permissions(self) -> None:
+ os.makedirs(self._directory.path, mode=0o755)
+ os.chmod(self._directory.path, 0o755)
+ state_path = self._directory.server_path("secure")
+ with self._directory as directory:
+ self.assertEqual(os.stat(directory.path).st_mode & 0o777, 0o700)
+ lock_path = os.path.join(directory.path, ".lock")
+ self.assertEqual(os.stat(lock_path).st_mode & 0o777, 0o600)
+ directory.write_json(state_path, {"token": "secret"})
+ self.assertEqual(os.stat(state_path).st_mode & 0o777, 0o600)
+ self.assertEqual(directory.read_json(state_path), {"token":
"secret"})
+
+ # O_CREAT does not apply its mode to an existing file, so writes
must re-assert it.
+ os.chmod(state_path, 0o644)
+ directory.write_json(state_path, {"token": "new-secret"})
+ self.assertEqual(os.stat(state_path).st_mode & 0o777, 0o600)
+
+ with open(state_path, "w") as state_file:
+ state_file.write("not json")
+ with self._directory as directory:
+ self.assertIsNone(directory.read_json(state_path))
+
+ def test_pool_directory_lock_blocks_another_process(self) -> None:
+ child = (
+ "import errno\n"
+ "import fcntl\n"
+ "import os\n"
+ "import sys\n"
+ "fd = os.open(sys.argv[1], os.O_RDWR)\n"
+ "try:\n"
+ " fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)\n"
+ "except OSError as error:\n"
+ " if error.errno not in (errno.EACCES, errno.EAGAIN):\n"
+ " raise\n"
+ "else:\n"
+ " raise RuntimeError('acquired a held lock')\n"
+ "finally:\n"
+ " os.close(fd)\n"
+ )
+ with self._directory:
+ result = subprocess.run(
+ [sys.executable, "-c", child,
os.path.join(self._directory.path, ".lock")],
+ capture_output=True,
+ text=True,
+ timeout=10,
+ )
+ self.assertEqual(result.returncode, 0, result.stderr)
+
+ def test_parse_entry_grammar(self) -> None:
+ # uids match the acquisition layer's uuid4().hex[:12]: nonempty
lowercase hex.
+ uid = "0123456789ab"
+ cases = [
+ # Well-formed state entries of every kind.
+ (f"pending-{uid}.json", ("pending", uid)),
+ (f"conf-{uid}.json", ("conf", uid)),
+ (f"server-{uid}.json", ("server", uid)),
+ (f"retired-{uid}.json", ("retired", uid)),
+ (f"claimed-4321-{uid}.json", ("claimed", uid)),
+ (f"member-{uid}", ("member", uid)),
+ # Short but valid hex uids.
+ ("member-abc", ("member", "abc")),
+ ("server-abc.json", ("server", "abc")),
+ # The lock file and unrelated entries.
+ (".lock", (None, None)),
+ ("random.txt", (None, None)),
+ # Editor droppings: the finding-2 cases that used to slip through
as phantom uids.
+ (f"member-{uid}.json.swp", (None, None)),
+ (f"server-{uid}.json.swp", (None, None)),
+ # Empty uids are rejected for every kind.
+ ("server-.json", (None, None)),
+ ("member-", (None, None)),
+ ("claimed-1234-.json", (None, None)),
+ # Non-hex uids (uppercase, out-of-range letters) are rejected.
+ ("server-ABCDEF.json", (None, None)),
+ ("server-ghij.json", (None, None)),
+ # Malformed claimed stems (non-numeric or missing pid, or a
malformed uid).
+ (f"claimed-notapid-{uid}.json", (None, None)),
+ (f"claimed-{uid}.json", (None, None)),
+ ("claimed-4321-ABCDEF.json", (None, None)),
+ # A pid that is str.isdigit() but not int()-parsable (superscript
two, U+00B2) must
+ # classify as "not claimed", never raise, since parse_entry runs
over every entry.
+ (f"claimed-{chr(0xB2)}-{uid}.json", (None, None)),
+ ]
+ for name, expected in cases:
+ with self.subTest(name=name):
+ self.assertEqual(PoolDirectory.parse_entry(name), expected)
+
+ def test_claiming_pid(self) -> None:
+ uid = "0123456789ab"
+ path = self._directory.claimed_path(4321, uid)
+ self.assertEqual(PoolDirectory.claiming_pid(path), 4321)
+ # Parses from the basename alone, independent of the directory prefix.
+ self.assertEqual(PoolDirectory.claiming_pid(f"claimed-7-{uid}.json"),
7)
+
+ def test_locked_accessors_enumerate_state(self) -> None:
+ uid_a, uid_b = "aaaaaaaaaaaa", "bbbbbbbbbbbb"
+ with self._directory as directory:
+ directory.write_json(directory.server_path(uid_a), {"a": 1})
+ directory.write_json(directory.pending_path(uid_a), {"a": 2})
+ directory.write_json(directory.server_path(uid_b), {"b": 1})
+ os.makedirs(directory.member_dir(uid_a), mode=0o700)
+
+ self.assertEqual(sorted(directory.uids()), [uid_a, uid_b])
+
+ states_a = directory.states(uid_a)
+ self.assertEqual(set(states_a), {"server", "pending", "member"})
+ self.assertEqual(states_a["server"], directory.server_path(uid_a))
+ self.assertEqual(states_a["member"], directory.member_dir(uid_a))
+
+ servers = dict(directory.paths_of_kind("server"))
+ self.assertEqual(set(servers), {uid_a, uid_b})
+ self.assertEqual(servers[uid_a], directory.server_path(uid_a))
+ self.assertEqual(directory.paths_of_kind("retired"), [])
+
+ def test_rename_remove_and_member_dir(self) -> None:
+ uid = "cccccccccccc"
+ with self._directory as directory:
+ src = directory.pending_path(uid)
+ dst = directory.server_path(uid)
+ directory.write_json(src, {"x": 1})
+ directory.rename(src, dst)
+ self.assertFalse(os.path.exists(src))
+ self.assertEqual(directory.read_json(dst), {"x": 1})
+
+ directory.remove(dst)
+ self.assertFalse(os.path.exists(dst))
+ # Removing a missing path is a no-op.
+ directory.remove(dst)
+
+ member = directory.member_dir(uid)
+ os.makedirs(member, mode=0o700)
+ with open(os.path.join(member, "inner"), "w") as f:
+ f.write("data")
+ directory.remove_member_dir(uid)
+ self.assertFalse(os.path.exists(member))
+ # Removing a missing member directory is a no-op.
+ directory.remove_member_dir(uid)
+
+ def test_states_rejects_duplicate_claimed(self) -> None:
+ uid = "dddddddddddd"
+ with self._directory as directory:
+ directory.write_json(directory.claimed_path(111, uid), {})
+ directory.write_json(directory.claimed_path(222, uid), {})
+ with self.assertRaisesRegex(AssertionError, "duplicate claimed"):
+ directory.states(uid)
+
+ def test_accessors_require_the_lock(self) -> None:
+ # The locked accessors must refuse to run outside the context manager.
+ with self.assertRaisesRegex(AssertionError, "context manager"):
+ self._directory.uids()
+
+ def test_not_reentrant(self) -> None:
+ with self._directory:
+ with self.assertRaisesRegex(AssertionError, "not reentrant"):
+ with self._directory:
+ pass
+
+
+if __name__ == "__main__":
+ from pyspark.testing import main
+
+ main()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
