From d13da159eac552f520e715e79343c581fd80cafd Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Fri, 24 Jul 2026 20:08:58 -0700 Subject: [PATCH 1/2] chore(pyamber): silence pkg_resources deprecation warning from fs import Every spawned Python UDF worker imports fs (PyFilesystem2), whose import of pkg_resources emits a UserWarning on stderr. Filter that one message before the core import chain, and mirror the filter for pytest output. Follow-up item from #6796. --- amber/pyproject.toml | 7 ++++++- amber/src/main/python/texera_run_python_worker.py | 11 +++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/amber/pyproject.toml b/amber/pyproject.toml index dd9f2f939e4..bfa83112bc0 100644 --- a/amber/pyproject.toml +++ b/amber/pyproject.toml @@ -42,4 +42,9 @@ testpaths = ["src/test/python"] addopts = "--import-mode=importlib" markers = [ "integration: end-to-end test routed to the amber-integration CI job", -] \ No newline at end of file +] +# `fs` pulls in pkg_resources at import time (#4199); the same filter lives in +# texera_run_python_worker.py for spawned workers. +filterwarnings = [ + "ignore:pkg_resources is deprecated as an API.*:UserWarning", +] diff --git a/amber/src/main/python/texera_run_python_worker.py b/amber/src/main/python/texera_run_python_worker.py index 1e131d45739..d6b447db483 100644 --- a/amber/src/main/python/texera_run_python_worker.py +++ b/amber/src/main/python/texera_run_python_worker.py @@ -18,8 +18,19 @@ import base64 import json import sys +import warnings from loguru import logger +# `fs` (imported transitively by the core import below) pulls in pkg_resources +# at import time (#4199), which emits this deprecation warning on stderr in +# every spawned worker. It is not actionable on our side, so filter it before +# the import chain runs. Mirrored for pytest in amber/pyproject.toml. +warnings.filterwarnings( + "ignore", + message="pkg_resources is deprecated as an API.*", + category=UserWarning, +) + try: from core.python_worker import PythonWorker from core.storage.storage_config import StorageConfig From 849a48c55e50fc32672a207a4f849a301427162c Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Mon, 27 Jul 2026 00:41:30 -0700 Subject: [PATCH 2/2] refactor(pyamber): drop PyFilesystem2 for stdlib tempfile fs (PyFilesystem2) is EOL: 2.4.16 (2022) is the last release ever and its import-time pkg_resources use is the sole reason setuptools was pinned below 82 (#4199, #6412). ExecutorManager was the only consumer; its temp:// filesystem becomes tempfile.mkdtemp + shutil.rmtree, with the UDF source encoding pinned to UTF-8 exactly as fs did (new non-ASCII round-trip regression test). Removes the fs and setuptools pins, syncs LICENSE-binary-python (drops fs-only transitive appdirs), and reverts the now-unneeded pkg_resources warning filters from #6880. Closes #6917. --- amber/LICENSE-binary-python | 2 - amber/pyproject.toml | 5 -- amber/requirements.txt | 3 - .../architecture/managers/executor_manager.py | 64 ++++++++++--------- .../main/python/texera_run_python_worker.py | 11 ---- .../managers/test_executor_manager.py | 36 ++++++++--- .../python/core/runnables/test_main_loop.py | 10 +-- 7 files changed, 64 insertions(+), 67 deletions(-) diff --git a/amber/LICENSE-binary-python b/amber/LICENSE-binary-python index b2d772c7f60..9ad569bc7e5 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/pyproject.toml b/amber/pyproject.toml index bfa83112bc0..28169079516 100644 --- a/amber/pyproject.toml +++ b/amber/pyproject.toml @@ -43,8 +43,3 @@ addopts = "--import-mode=importlib" markers = [ "integration: end-to-end test routed to the amber-integration CI job", ] -# `fs` pulls in pkg_resources at import time (#4199); the same filter lives in -# texera_run_python_worker.py for spawned workers. -filterwarnings = [ - "ignore:pkg_resources is deprecated as an API.*:UserWarning", -] diff --git a/amber/requirements.txt b/amber/requirements.txt index 8baa0aee951..d7dd6c1053e 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..ffeae1ff1fe 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,27 @@ 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 OS temp 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("/")) + # periodically clean up its temp folder anyway, the full-life-cycle + # management is not a priority to be fixed. + 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 +91,19 @@ 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: + # importlib decodes the source file as UTF-8 (PEP 3120), so the write + # side must be pinned to UTF-8 too — the locale default (e.g. cp1252 + # on Windows) would break non-ASCII UDF code. newline="\n" keeps the + # written bytes identical across platforms. + with (self.tmp_dir / file_name).open( + "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 {self.tmp_dir / file_name}.") # 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 file we just wrote — no re-import / reload dance. executor_module = importlib.import_module(module_name) self.operator_module_name = module_name @@ -113,25 +115,25 @@ 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 path 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__: + # tmp_dir 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: sys.modules.pop(self.operator_module_name, None) - logger.debug(f"Tmp directory {root} is closed and cleared.") + logger.debug(f"Tmp directory {root} is removed and cleared.") @staticmethod def is_concrete_operator(cls: type) -> bool: diff --git a/amber/src/main/python/texera_run_python_worker.py b/amber/src/main/python/texera_run_python_worker.py index d6b447db483..1e131d45739 100644 --- a/amber/src/main/python/texera_run_python_worker.py +++ b/amber/src/main/python/texera_run_python_worker.py @@ -18,19 +18,8 @@ import base64 import json import sys -import warnings from loguru import logger -# `fs` (imported transitively by the core import below) pulls in pkg_resources -# at import time (#4199), which emits this deprecation warning on stderr in -# every spawned worker. It is not actionable on our side, so filter it before -# the import chain runs. Mirrored for pytest in amber/pyproject.toml. -warnings.filterwarnings( - "ignore", - message="pkg_resources is deprecated as an API.*", - category=UserWarning, -) - try: from core.python_worker import PythonWorker from core.storage.storage_config import StorageConfig 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..af815f3af69 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 @@ -39,6 +39,17 @@ def produce(self) -> Iterator[Union[TupleLike, TableLike, None]]: 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,26 +266,31 @@ def test_source_operator_mismatch_raises_error(self, executor_manager): ) 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) would break either the write or the import. + executor_manager.initialize_executor( + code=NON_ASCII_OPERATOR_CODE, is_source=False, language="python" + ) + assert executor_manager.executor.GREETING == "café" + def test_close_when_sys_path_entry_already_removed(self): - # Exercise the except ValueError branch: if the tmp fs path has - # already been pulled out of sys.path by something else, close() + # Exercise the except ValueError branch: if the tmp directory path + # 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 - 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 diff --git a/amber/src/test/python/core/runnables/test_main_loop.py b/amber/src/test/python/core/runnables/test_main_loop.py index 2bbefe8c432..af072043bd5 100644 --- a/amber/src/test/python/core/runnables/test_main_loop.py +++ b/amber/src/test/python/core/runnables/test_main_loop.py @@ -2613,8 +2613,8 @@ def test_two_main_loops_load_distinct_operator_classes(self): Regression test for executor-module contamination (#4705): executor modules were once named ``udf-v``, so every loop's first executor was ``udf-v1`` in the process-wide - ``sys.modules``. A loop whose worker never completes never closes its - temp fs, so its ``udf-v1.py`` lingered on ``sys.path`` and the next + ``sys.modules``. A loop whose worker never completes never removes its + temp directory, so its ``udf-v1.py`` lingered on ``sys.path`` and the next loop re-resolved ``udf-v1`` to that older file, silently running the wrong operator. This test uses NO monkeypatch of ``gen_module_file_name`` -- module names must be process-globally @@ -2633,9 +2633,9 @@ def test_two_main_loops_load_distinct_operator_classes(self): second = Context("worker-second", InternalQueue()) try: # The first loop loads EchoOperator and is intentionally left "unfinished" - # until after the second loop is initialized: its temp fs is not closed yet, - # so its udf module and sys.path entry linger exactly as a crashed / - # never-completed worker's would. + # until after the second loop is initialized: its temp directory is not + # removed yet, so its udf module and sys.path entry linger exactly as a + # crashed / never-completed worker's would. first.executor_manager.initialize_executor( echo_code, is_source=False, language="python" )