[SPARK-58021][CONNECT] Add local server pool filesystem storage

[ericm-db]

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.

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)

[HyukjinKwon]

0 blocking, 0 non-blocking, 0 nits.
High-quality, defensively-designed pool storage; concurrency and crash-safety handled carefully. No issues found.

Verification

Verified the concurrency discipline: every mutating accessor asserts the exclusive lock is held (_assert_locked); state transitions use atomic os.rename; read_json returns None on OSError/ValueError so partial writes are treated as absent and reaped; _pid_alive is zombie-aware on Linux so dead-client reclamation is correct. Build/test: python/run-tests --testnames 'pyspark.sql.tests.connect.test_connect_local_server_pool'.

[dtenedor]

Findings

1. Nesting the context manager deadlocks silently. I confirmed this by running it — sequential re-entry works as the docstring promises, but a nested with on the same object hangs forever (killed at a 10s timeout). The second os.open creates a new open file description, so flock(LOCK_EX) blocks against the fd the same process already holds.

Nothing in this PR nests it, but four more layers build on this foundation, and acquisition-with-polling is precisely where someone writes a helper that enters the lock while a caller already holds it. The failure mode is the bad kind: no exception, just a wedged getOrCreate(). A one-line guard converts it into an immediate, obvious error and makes the docstring's "entered again after exiting" contract enforceable:

def __enter__(self) -> "PoolDirectory":
    import fcntl

    assert self._lock_fd is None, "PoolDirectory is not reentrant"
    os.makedirs(self.path, mode=0o700, exist_ok=True)

2. parse_entry's member- branch is looser than its documented contract. The member- check runs before the .json gate, so it swallows anything with that prefix. Probe results:

Entry	Result
member-abc	('member', 'abc')
member-abc.json.swp	('member', 'abc.json.swp')
server-abc.json.swp	(None, None)
server-.json	('server', '')
claimed-1234-.json	('claimed', '')

The docstring says non-matching entries — "the lock file, editor droppings, ..." — return (None, None), and that holds for every JSON kind but not for member-. A stray member-* file becomes a phantom uid in uids(), and it doesn't self-heal: remove_member_dir calls shutil.rmtree on a file with ignore_errors=True, so the dropping survives and the phantom reappears on every subsequent scan. The empty-uid cases are unreachable with generated uids but point the same direction. Validating the uid shape once inside parse_entry fixes all of these together.

3. parse_entry has no test at all. It's the only nontrivial logic in the module and it's a pure static function, so it's the cheapest thing here to test. The three tests you have cover directory selection, permissions plus malformed JSON, and cross-process lock contention; parse_entry, claiming_pid, uids, states, paths_of_kind, rename, remove, remove_member_dir, and the _assert_locked guard are all uncovered. For a PR whose stated purpose is to make the filesystem contract "reviewable on its own," a table-driven test over the filename grammar is the single highest-value addition — and it's what would have caught finding 2.

4. The filename grammar is parsed twice. parse_entry and claiming_pid each hand-parse claimed-<pid>-<uid> independently, so a future naming change can drift between them. Having parse_entry surface the pid, or claiming_pid delegate to it, keeps one parser.

5. states() silently collapses duplicates. found[kind] = path keeps whichever entry sorts last, and claimed is the one kind where two entries for a uid is conceivable (two client pids). The rename-based claiming in #57907 should make that structurally impossible, but since this layer is the contract, it's worth an assert or an explicit note — picking the wrong file would make the reaper read the wrong pid via claiming_pid and reclaim a live client's member.

6. The test skips are over-restrictive. is_remote_only() and should_test_connect are copied from the sibling suite, but this one starts no server, and local_server_pool.py imports only stdlib at module scope (the runtime_dir import is deferred). These pure-filesystem tests would pass fine in a remote-only build. The skip message "Requires Spark Connect test dependencies" also doesn't describe what is_remote_only gates.

[dtenedor]

Reviewed the test coverage again and ran the tests, the testing looks good now.

[dtenedor]

LGTM, merging to master + 4.x

[dtenedor]

Merge Summary:

• merged into master 001e89c
• merged into branch-4.x 44bae99

Posted by merge_spark_pr.py