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-8278-adc6a4ac4d79f0302534228d119ea2c22f393db7 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 27bc3b7703c7e044c03a8d76edeb708146441fce Author: carloea2 <[email protected]> AuthorDate: Sun Oct 11 23:11:33 2026 +0000 feat(pyamber): skip duplicate output writers (#8278) ### What changes were proposed in this PR? Start output storage writers only when the output port is first registered. Duplicate registration keeps the existing port and writer set. ### Any related issues, documentation, discussions? Closes #8277 ### How was this PR tested? Added coverage proving that duplicate port registration does not start replacement writers, alongside the existing positive writer-setup coverage. `C:\Users\carlo\texera\texera\.venv312\Scripts\python.exe -c "import sys,pytest; sys.path[:0]=[r'C:\Users\carlo\texera\texera-worktrees\investigate-bug74\amber\src\main\python',r'C:\Users\carlo\texera\texera\amber\src\main\python']; raise SystemExit(pytest.main([r'amber\src\test\python\core\architecture\packaging\test_output_manager.py','-p','no:cacheprovider','-q']))"` `C:\Users\carlo\texera\texera\.venv312\Scripts\ruff.exe check amber/src/main/python/core/architecture/packaging/output_manager.py amber/src/test/python/core/architecture/packaging/test_output_manager.py` `C:\Users\carlo\texera\texera\.venv312\Scripts\ruff.exe format --check amber/src/main/python/core/architecture/packaging/output_manager.py amber/src/test/python/core/architecture/packaging/test_output_manager.py` All 40 tests passed. Ruff checks passed. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Codex --- .../main/python/core/architecture/packaging/output_manager.py | 5 ++--- .../python/core/architecture/packaging/test_output_manager.py | 11 +++++++++++ 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/amber/src/main/python/core/architecture/packaging/output_manager.py b/amber/src/main/python/core/architecture/packaging/output_manager.py index bc6829520e..5053fd01f4 100644 --- a/amber/src/main/python/core/architecture/packaging/output_manager.py +++ b/amber/src/main/python/core/architecture/packaging/output_manager.py @@ -124,11 +124,10 @@ class OutputManager: if port_id.internal is None: port_id.internal = False - if storage_uri_base is not None: - self.set_up_port_storage_writer(port_id, storage_uri_base) - # each port can only be added and initialized once. if port_id not in self._ports: + if storage_uri_base is not None: + self.set_up_port_storage_writer(port_id, storage_uri_base) self._ports[port_id] = WorkerPort(schema) def set_up_port_storage_writer(self, port_id: PortIdentity, storage_uri_base: str): diff --git a/amber/src/test/python/core/architecture/packaging/test_output_manager.py b/amber/src/test/python/core/architecture/packaging/test_output_manager.py index d929dda62b..841e122be6 100644 --- a/amber/src/test/python/core/architecture/packaging/test_output_manager.py +++ b/amber/src/test/python/core/architecture/packaging/test_output_manager.py @@ -396,6 +396,17 @@ class TestAddOutputPort: assert output_manager.get_port_ids() == [port_id] assert output_manager.get_port().get_schema() is schema_first + def test_duplicate_port_does_not_start_replacement_storage_writers( + self, output_manager + ): + output_manager.set_up_port_storage_writer = MagicMock() + port_id = PortIdentity(id=0, internal=False) + output_manager.add_output_port(port_id, MagicMock(), "vfs:///first") + output_manager.add_output_port(port_id, MagicMock(), "vfs:///duplicate") + output_manager.set_up_port_storage_writer.assert_called_once_with( + port_id, "vfs:///first" + ) + def test_sets_up_storage_writer_only_when_uri_given(self, output_manager): output_manager.set_up_port_storage_writer = MagicMock() port_a = PortIdentity(id=0, internal=False)
