From 309a28b45f0852f182b825b9b93a872606c040fb Mon Sep 17 00:00:00 2001 From: ali Date: Mon, 27 Jul 2026 13:32:12 -0700 Subject: [PATCH] refactor(pyamber): replace PyFilesystem2 with tempfile and unpin setuptools `fs` (PyFilesystem2) is unmaintained -- 2.4.16 (May 2022) is its final release and it calls pkg_resources at import time. setuptools 82.0.0 removed pkg_resources, so `import fs` hard-crashes there, which is the only reason amber/requirements.txt has carried a setuptools pin since #4199. ExecutorManager was the only consumer, and fs's TempFS is internally tempfile.mkdtemp + shutil.rmtree, so the swap is behavior-preserving: lazy creation, sys.path append/remove, the never-materialized early return, and the tolerated leak-on-force-kill semantics are unchanged. tempfile.mkdtemp is deliberate over TemporaryDirectory -- the latter attaches a finalizer that would silently clean up abandoned directories. Also calls importlib.invalidate_caches() before importing the freshly written UDF module. The import system caches a directory listing per sys.path entry and refreshes it on mtime change only, so a second UDF written into the same tmp directory within one mtime tick could be invisible to the finder. This reproduces on main as well (four TestUpdateExecutor cases fail locally with fs installed); it is fixed here because the rewritten write-then-import path is where it lives. Dropping fs also drops appdirs (an fs-only transitive) from the binary license manifest; six stays because python-dateutil still needs it. Closes #6917 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01C4UQSRY1RAXr7gJVFLLT7h --- amber/LICENSE-binary-python | 2 - amber/requirements.txt | 3 - .../architecture/managers/executor_manager.py | 66 +++++++++++-------- .../managers/test_executor_manager.py | 25 ++++--- 4 files changed, 54 insertions(+), 42 deletions(-) diff --git a/amber/LICENSE-binary-python b/amber/LICENSE-binary-python index 0b59956f80f..962eac54571 100644 --- a/amber/LICENSE-binary-python +++ b/amber/LICENSE-binary-python @@ -247,7 +247,6 @@ Python packages: - annotated-doc==0.0.4 - annotated-types==0.7.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.29.7 - fonttools==4.63.0 - - fs==2.4.16 - greenlet==3.5.3 - h11==0.16.0 - h2==4.3.0 diff --git a/amber/requirements.txt b/amber/requirements.txt index cafdf99e594..25c30339eaa 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 7f2249476ad..797a62cf7f3 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,13 @@ # specific language governing permissions and limitations # under the License. -import fs import importlib import inspect import itertools +import shutil 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 @@ -48,28 +48,32 @@ def __init__(self): self.operator_module_name: Optional[str] = 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("/")) + # `tempfile.mkdtemp` is deliberate: `tempfile.TemporaryDirectory` + # would attach a finalizer that silently removes the directory on GC or + # interpreter exit, which would change the leak-on-force-kill behavior + # described above. + root = Path(tempfile.mkdtemp(prefix="texera-udf-")) 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,16 +96,22 @@ def load_executor_definition(self, code: str) -> type(Operator): """ 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) + with open(file_path, "w") 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}.") + + # The import system caches a directory listing per sys.path entry and + # only refreshes it when the directory mtime changes. A second UDF + # written into the same tmp directory within one mtime tick would + # otherwise be invisible to the finder and raise ModuleNotFoundError, + # so drop the stale listing before importing (see the note in + # importlib.invalidate_caches on creating modules at runtime). + 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 @@ -113,20 +123,20 @@ def load_executor_definition(self, code: str) -> type(Operator): 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 + shutil.rmtree(root, ignore_errors=True) 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 1a21b106f30..e30f7323e4f 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 @@ -256,28 +256,35 @@ def test_source_operator_mismatch_raises_error(self, executor_manager): assert "SourceOperator API" in str(exc_info.value) 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 = """