This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6946-e7cd5b1ee030908455b65c79defc9c98e60a9372 in repository https://gitbox.apache.org/repos/asf/texera.git
commit d8bb0f180fa0ea12f25babd5af5dd24e316c3477 Author: ali risheh <[email protected]> AuthorDate: Mon Aug 3 12:28:33 2026 -0700 refactor(pyamber): replace PyFilesystem2 with tempfile and unpin setuptools (#6946) ### What changes were proposed in this PR? Implements the migration planned in #6917: `ExecutorManager` drops `fs` (PyFilesystem2) for stdlib `tempfile`, and `setuptools` is unpinned. The issue covers the rationale, the before/after shape, the required footprint, and the alternatives that were ruled out — this description only covers what it doesn't. **Two behaviours `fs` was providing implicitly, now explicit.** Both came out of review: - **Encoding.** `OSFS.open` defaults to `encoding="utf-8"` with `newline=""`; the builtin `open()` uses the locale encoding and translates newlines. Since `importlib` always decodes source as UTF-8 (PEP 3120), a UDF containing non-ASCII text would have failed at *write* time under a non-UTF-8 locale. The write is now pinned to `encoding="utf-8", newline="\n"`, with a regression test. - **Reclaim on GC.** `FS.__del__` → `TempFS.close()` → `clean()` meant an `ExecutorManager` abandoned without `close()` still had its directory removed when collected. A bare `mkdtemp` would have turned that into a permanent leak, so the `TemporaryDirectory` handle is held on the instance and `close()` calls `cleanup()`. Force kill runs no finalizers either way — that case is unchanged, and is what the long-standing TODO in `tmp_dir` is about. **Not included:** `importlib.invalidate_caches()`. This branch carried its own copy; #7173 landed the same fix on `main` first, and the hunks merge cleanly into a doubled call. Dropped on rebase in favour of what's on `main`. ### Any related issues, documentation, discussions? Closes #6917. Supersedes #6911 (the setuptools v83 bump — closed, it could not merge while `fs` was present) and #6412 (`setuptools<82`, the interim bound). Same goal as #6928 (closed); the non-ASCII round-trip test is lifted from it. Original pin: #4199, re-affirmed by the #6110 pin audit. ### How was this PR tested? Rebased onto `main` (`e7cd5b1`) and re-verified on Python 3.12 in a clean venv built from `amber/requirements.txt` + `amber/dev-requirements.txt`, proto bindings generated via `bin/python-proto-gen.sh`. `import fs` in that venv raises `ModuleNotFoundError`, confirming nothing reaches it. - `pytest -m "not integration"` → **997 passed, 1 deselected, 1 xfailed**. Exercises `ExecutorManager` directly and end to end through UDF loading (`pytexera/udf/`, `test_initialize_executor_handler.py`, `test_update_executor_handler.py`, `runnables/test_main_loop.py`). `test_executor_manager.py` run 5× consecutively → 25 passed each time. - The encoding regression test was checked in both directions under `LC_ALL=C PYTHONCOERCECLOCALE=0 PYTHONUTF8=0` (interpreter reports `preferred: ANSI_X3.4-1968`): passes with the fix, and fails with `UnicodeEncodeError: 'ascii' codec can't encode characters in position 70-73` without it. CI runners are UTF-8, so this would not have been caught there. - `ruff check` and `ruff format --check` over `src/main/python src/test/python` → clean. - License drift, the same check CI runs: `pip-licenses` over the runtime closure + `bin/licensing/check_binary_deps.py --ignore-transitive-version python` → `OK: 109 Python packages match LICENSE-binary`. Running that same package list against `main`'s `LICENSE-binary-python` reports exactly `STALE: appdirs==1.4.4, fs==2.4.16` — the removals here are precisely what's required, and nothing more. - The unpin was checked independently of torch: a `requirements.txt`-only venv on 3.12 ships **no** `setuptools` or `pkg_resources` at all, and importing every runtime dependency with both names blocked from `sys.meta_path` raises nothing. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) Co-authored-by: Claude Opus 5 (1M context) <[email protected]> --- amber/LICENSE-binary-python | 2 - amber/requirements.txt | 3 - .../core/architecture/managers/executor_manager.py | 71 +++++++++++++--------- .../architecture/managers/test_executor_manager.py | 58 +++++++++++++++--- 4 files changed, 92 insertions(+), 42 deletions(-) diff --git a/amber/LICENSE-binary-python b/amber/LICENSE-binary-python index 9faae2f76b..978689ffc5 100644 --- a/amber/LICENSE-binary-python +++ b/amber/LICENSE-binary-python @@ -247,7 +247,6 @@ Python packages: - annotated-doc==0.0.5 - annotated-types==0.8.0 - anyio==4.14.2 - - appdirs==1.4.4 - asn1crypto==1.5.1 - attrs==26.1.0 - betterproto==2.0.0b7 @@ -255,7 +254,6 @@ Python packages: - charset-normalizer==3.4.9 - filelock==3.32.2 - fonttools==4.63.0 - - fs==2.4.16 - greenlet==3.5.4 - h11==0.16.0 - h2==4.4.0 diff --git a/amber/requirements.txt b/amber/requirements.txt index cafdf99e59..25c30339ea 100644 --- a/amber/requirements.txt +++ b/amber/requirements.txt @@ -15,8 +15,6 @@ # specific language governing permissions and limitations # under the License. -# Not imported directly: `fs` needs pkg_resources at import time (#4199). -setuptools==80.10.2 numpy==2.1.0 pandas==2.2.3 loguru==0.7.3 @@ -25,7 +23,6 @@ betterproto==2.0.0b7 pampy==0.3.0 overrides==7.7.0 typing_extensions==4.14.1 -fs==2.4.16 bidict==0.22.0 cached_property==2.0.1 psutil==7.2.2 diff --git a/amber/src/main/python/core/architecture/managers/executor_manager.py b/amber/src/main/python/core/architecture/managers/executor_manager.py index 375033ad60..8daa0b9e64 100644 --- a/amber/src/main/python/core/architecture/managers/executor_manager.py +++ b/amber/src/main/python/core/architecture/managers/executor_manager.py @@ -15,13 +15,12 @@ # specific language governing permissions and limitations # under the License. -import fs import importlib import inspect import itertools import sys +import tempfile from cached_property import cached_property -from fs.base import FS from loguru import logger from pathlib import Path from typing import Tuple, Optional @@ -46,30 +45,42 @@ class ExecutorManager: def __init__(self): self.executor: Optional[Operator] = None self.operator_module_name: Optional[str] = None + # Owns the tmp directory's lifetime; see `tmp_dir`. + self._tmp_dir_handle: Optional[tempfile.TemporaryDirectory] = None @cached_property - def fs(self) -> FS: + def tmp_dir(self) -> Path: """ - Creates a tmp fs for storing source code, which will be removed when the - workflow is completed. + Creates a tmp directory for storing source code, which will be removed + when the workflow is completed. :return: """ # TODO: # For various reasons when the workflow is not completed successfully, - # the tmp fs could not be closed properly. This means it may leave files - # in the /var/tmp folder after a partially started or failed execution. - # A full-life-cycle management of tmp fs is required to consider all - # possible errors happened during execution. However, the full-life-cycle - # management could be hard due to errors from JAVA side which causes force - # kill on the Python process. + # the tmp directory could not be removed properly. This means it may leave + # files in the /var/tmp folder after a partially started or failed + # execution. + # A full-life-cycle management of the tmp directory is required to + # consider all possible errors happened during execution. However, the + # full-life-cycle management could be hard due to errors from JAVA side + # which causes force kill on the Python process. # As each python file is usually tiny in size, and the OS can # periodically clean up /var/tmp anyway, the full-life-cycle management is # not a priority to be fixed. - temp_fs = fs.open_fs("temp://") - root = Path(temp_fs.getsyspath("/")) + # `TemporaryDirectory` is held on the instance rather than using a + # bare `mkdtemp` so that a manager abandoned without `close()` still has + # its directory reclaimed when it is garbage-collected. `fs` gave us that + # for free (`FS.__del__` -> `TempFS.close()` -> `clean()`); dropping it + # would turn every abandoned manager into a permanent leak. Force kill + # runs no finalizers either way, which is the case the TODO above is about. + # `ignore_cleanup_errors=True` matches TempFS's `ignore_clean_errors=True`. + self._tmp_dir_handle = tempfile.TemporaryDirectory( + prefix="texera-udf-", ignore_cleanup_errors=True + ) + root = Path(self._tmp_dir_handle.name) logger.debug(f"Opening a tmp directory at {root}.") sys.path.append(str(root)) - return temp_fs + return root def gen_module_file_name(self) -> Tuple[str, str]: """ @@ -92,19 +103,22 @@ class ExecutorManager: """ module_name, file_name = self.gen_module_file_name() - with self.fs.open(file_name, "w") as file: + file_path = self.tmp_dir.joinpath(file_name) + # Pin the encoding: importlib always decodes source as UTF-8 (PEP + # 3120), while the builtin open() writes in the locale encoding, so a + # UDF containing non-ASCII text would fail to write under a non-UTF-8 + # locale (cp1252, LC_ALL=C). `fs` was passing encoding="utf-8" and + # newline="" to io.open on our behalf; both are now explicit. + with open(file_path, "w", encoding="utf-8", newline="\n") as file: file.write(code) - logger.debug( - "A tmp py file is written to " - f"{Path(self.fs.getsyspath('/')).joinpath(file_name)}." - ) + logger.debug(f"A tmp py file is written to {file_path}.") # Clear importlib's directory listing cache so freshly written # temporary modules are discoverable on systems with coarse mtime. importlib.invalidate_caches() # gen_module_file_name guarantees module_name is unique across # the process, so import_module will always cleanly load source - # from the tmp fs we just wrote — no re-import / reload dance. + # from the tmp directory we just wrote — no re-import / reload dance. executor_module = importlib.import_module(module_name) self.operator_module_name = module_name @@ -116,20 +130,21 @@ class ExecutorManager: def close(self) -> None: """ - Close the tmp fs and release all resources created within it. + Remove the tmp directory and release all resources created within it. This also evicts the loaded operator module from ``sys.modules`` - and removes the tmp fs path from ``sys.path`` so a single call - fully reverses every global side-effect performed by ``fs`` and + and removes the tmp directory from ``sys.path`` so a single call + fully reverses every global side-effect performed by ``tmp_dir`` and ``load_executor_definition``. :return: """ - if "fs" not in self.__dict__: - # fs was never materialized; nothing to clean up. + if "tmp_dir" not in self.__dict__: + # the tmp directory was never materialized; nothing to clean up. return - root = self.fs.getsyspath("/") - self.fs.close() + root = self.tmp_dir + self._tmp_dir_handle.cleanup() + self._tmp_dir_handle = None try: - sys.path.remove(str(Path(root))) + sys.path.remove(str(root)) except ValueError: pass if self.operator_module_name is not None: diff --git a/amber/src/test/python/core/architecture/managers/test_executor_manager.py b/amber/src/test/python/core/architecture/managers/test_executor_manager.py index 1a21b106f3..07ac054c06 100644 --- a/amber/src/test/python/core/architecture/managers/test_executor_manager.py +++ b/amber/src/test/python/core/architecture/managers/test_executor_manager.py @@ -15,6 +15,7 @@ # specific language governing permissions and limitations # under the License. +import gc import sys import pytest from unittest.mock import MagicMock @@ -39,6 +40,17 @@ class TestSourceOperator(UDFSourceOperator): yield Tuple({"test": "data"}) """ +NON_ASCII_OPERATOR_CODE = """ +from pytexera import * + +class NonAsciiOperator(UDFOperatorV2): + # コメント: user code may contain any Unicode text. + GREETING = "café" + + def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]: + yield tuple_ +""" + class TestExecutorManager: """Test suite for ExecutorManager, focusing on R UDF plugin support.""" @@ -255,29 +267,57 @@ class TestExecutorManager: ) assert "SourceOperator API" in str(exc_info.value) + def test_non_ascii_udf_source_round_trip(self, executor_manager): + # importlib decodes the written source file as UTF-8 (PEP 3120), so + # the write side must be pinned to UTF-8 as well; the locale default + # (e.g. cp1252 on Windows, or LC_ALL=C) would break the write. + executor_manager.initialize_executor( + code=NON_ASCII_OPERATOR_CODE, is_source=False, language="python" + ) + assert executor_manager.executor.GREETING == "café" + + def test_tmp_dir_is_reclaimed_when_manager_is_abandoned(self): + # `fs` reclaimed the tmp directory on GC (FS.__del__ -> TempFS.close() + # -> clean()). A manager dropped without close() must still not leak, + # so the TemporaryDirectory finalizer has to reproduce that. + manager = ExecutorManager() + root = manager.tmp_dir + assert root.exists() + sys.path.remove(str(root)) + del manager + gc.collect() + assert not root.exists() + def test_close_when_sys_path_entry_already_removed(self): - # Exercise the except ValueError branch: if the tmp fs path has + # Exercise the except ValueError branch: if the tmp directory has # already been pulled out of sys.path by something else, close() # should swallow the error and finish cleanly. - from pathlib import Path - manager = ExecutorManager() - root = Path(manager.fs.getsyspath("/")) + root = manager.tmp_dir sys.path.remove(str(root)) manager.close() assert str(root) not in sys.path + assert not root.exists() - def test_close_when_fs_materialized_but_no_executor_loaded(self): - # Exercise the branch where self.fs was touched (so the early + def test_close_when_tmp_dir_materialized_but_no_executor_loaded(self): + # Exercise the branch where self.tmp_dir was touched (so the early # return is skipped) but operator_module_name is still None, # meaning the sys.modules.pop branch must NOT execute. - from pathlib import Path - manager = ExecutorManager() - root = Path(manager.fs.getsyspath("/")) + root = manager.tmp_dir assert manager.operator_module_name is None manager.close() assert str(root) not in sys.path + assert not root.exists() + + def test_close_is_noop_when_tmp_dir_never_materialized(self): + # The early return: no tmp directory was ever created, so close() + # must not touch sys.path and must not raise. + manager = ExecutorManager() + before = list(sys.path) + manager.close() + assert sys.path == before + assert "tmp_dir" not in manager.__dict__ REPLACEMENT_OPERATOR_CODE = """
