This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 27bc3b7703 feat(pyamber): skip duplicate output writers (#8278)
27bc3b7703 is described below
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)
