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 = """