refactor(ENG-379): make iter_packets() read-only, consolidate DB join sites

test-objective/unit/test_nodes.py

@@ -139,7 +139,8 @@ def test_caches_computed_results(self):
             pipeline_database=pipeline_db,
             result_database=result_db,
         )
-        # First iteration computes all
+        # run() computes all; iter_packets() is read-only and serves from cache
+        node.run()
         packets = list(node.iter_packets())
         assert len(packets) == 3
 

tests/test_channels/test_node_async_execute.py

@@ -678,12 +678,13 @@ def test_function_node_sequential_uses_execute_packet(self):
         # Monkey-patch to verify routing through internal path
         original = node._process_packet_internal
 
-        def patched(tag, packet):
+        def patched(tag, packet, *, logger=None):
             call_log.append("_process_packet_internal")
-            return original(tag, packet)
+            return original(tag, packet, logger=logger)
 
         node._process_packet_internal = patched
 
+        node.run()  # computation now triggered via run(), not iter_packets()
         results = list(node.iter_packets())
         assert len(results) == 3
         assert len(call_log) == 3

tests/test_core/function_pod/test_function_node_attach_db.py

@@ -45,6 +45,7 @@ def test_construction_without_database(self):
 
     def test_iter_packets_without_database(self):
         node = FunctionNode(function_pod=_make_pod(), input_stream=_make_stream(n=3))
+        node.run()
         results = list(node.iter_packets())
         assert len(results) == 3
         assert results[0][1]["result"] == 0
@@ -76,7 +77,7 @@ def test_attach_databases_creates_cached_function_pod(self):
 
     def test_attach_databases_clears_caches(self):
         node = FunctionNode(function_pod=_make_pod(), input_stream=_make_stream())
-        list(node.iter_packets())  # populate cache
+        node.run()  # populate cache
         assert len(node._cached_output_packets) > 0
         db = InMemoryArrowDatabase()
         node.attach_databases(pipeline_database=db, result_database=db)
@@ -107,6 +108,7 @@ def test_iter_packets_after_attach_works(self):
         node = FunctionNode(function_pod=_make_pod(), input_stream=_make_stream(n=2))
         db = InMemoryArrowDatabase()
         node.attach_databases(pipeline_database=db, result_database=db)
+        node.run()
         results = list(node.iter_packets())
         assert len(results) == 2
 
@@ -140,5 +142,6 @@ def test_iter_packets_with_database(self):
             pipeline_database=db,
             result_database=db,
         )
+        node.run()
         results = list(node.iter_packets())
         assert len(results) == 3

tests/test_core/function_pod/test_function_node_caching.py

@@ -249,6 +249,7 @@ def test_phase2_processes_novel_entry_ids_only(self):
             [{"id": 0, "x": 10}, {"id": 1, "x": 20}, {"id": 2, "x": 30}]
         )
         node2, _ = _make_node(stream2, db=db)
+        node2.run()
         results = list(node2.iter_packets())
 
         # Should yield 3 total: 2 from Phase 1 + 1 from Phase 2
@@ -279,6 +280,7 @@ def test_same_packet_new_tag_triggers_phase2(self):
         assert node1.node_identity_path == node2.node_identity_path
         assert node1.node_identity_path[-1].startswith("schema:")
 
+        node2.run()
         results = list(node2.iter_packets())
 
         # Phase 1 finds no records for node2's content_hash → Phase 2 processes the row

tests/test_core/function_pod/test_function_pod_node.py

@@ -302,11 +302,13 @@ class TestFunctionNodeStreamInterface:
     @pytest.fixture
     def node(self, double_pf) -> FunctionNode:
         db = InMemoryArrowDatabase()
-        return FunctionNode(
+        node = FunctionNode(
             function_pod=FunctionPod(packet_function=double_pf),
             input_stream=make_int_stream(n=3),
             pipeline_database=db,
         )
+        node.run()
+        return node
 
     def test_iter_packets_correct_values(self, node):
         assert [packet["result"] for _, packet in node.iter_packets()] == [0, 2, 4]

tests/test_core/function_pod/test_function_pod_node_stream.py

@@ -61,11 +61,13 @@ class TestFunctionNodeStreamBasic:
     @pytest.fixture
     def node(self, double_pf) -> FunctionNode:
         db = InMemoryArrowDatabase()
-        return FunctionNode(
+        node = FunctionNode(
             function_pod=FunctionPod(packet_function=double_pf),
             input_stream=make_int_stream(n=3),
             pipeline_database=db,
         )
+        node.run()
+        return node
 
     def test_iter_packets_yields_correct_count(self, node):
         assert len(list(node.iter_packets())) == 3
@@ -117,6 +119,7 @@ def test_output_schema_has_result_in_packet_schema(self, node):
 class TestFunctionNodeColumnConfig:
     def test_as_table_content_hash_column(self, double_pf):
         node = _make_node(double_pf, n=3)
+        node.run()
         table = node.as_table(columns={"content_hash": True})
         assert "_content_hash" in table.column_names
         assert len(table.column("_content_hash")) == 3
@@ -138,6 +141,7 @@ def test_as_table_sort_by_tags(self, double_pf):
             input_stream=input_stream,
             pipeline_database=db,
         )
+        node.run()
         result = node.as_table(columns={"sort_by_tags": True})
         ids: list[int] = result.column("id").to_pylist()  # type: ignore[assignment]
         assert ids == sorted(ids)
@@ -166,6 +170,7 @@ def test_as_table_returns_cached_results_when_packet_function_inactive(
         n = 3
         db = InMemoryArrowDatabase()
         node1 = _make_node(double_pf, n=n, db=db)
+        node1.run()
         table1 = node1.as_table()
         assert len(table1) == n
 
@@ -242,8 +247,12 @@ def test_db_served_results_have_correct_values(self, double_pf):
         n = 4
         db = InMemoryArrowDatabase()
 
-        table1 = _make_node(double_pf, n=n, db=db).as_table()
-        table2 = _make_node(double_pf, n=n, db=db).as_table()
+        node1 = _make_node(double_pf, n=n, db=db)
+        node1.run()
+        table1 = node1.as_table()
+        node2 = _make_node(double_pf, n=n, db=db)
+        node2.run()
+        table2 = node2.as_table()
 
         assert sorted(table1.column("result").to_pylist()) == sorted(
             table2.column("result").to_pylist()
@@ -300,14 +309,18 @@ def test_partial_fill_total_row_count_correct(self, double_pf):
         n = 4
         db = InMemoryArrowDatabase()
         _fill_node(_make_node(double_pf, n=2, db=db))
-        packets = list(_make_node(double_pf, n=n, db=db).iter_packets())
+        node = _make_node(double_pf, n=n, db=db)
+        node.run()
+        packets = list(node.iter_packets())
         assert len(packets) == n
 
     def test_partial_fill_all_values_correct(self, double_pf):
         n = 4
         db = InMemoryArrowDatabase()
         _fill_node(_make_node(double_pf, n=2, db=db))
-        table = _make_node(double_pf, n=n, db=db).as_table()
+        node = _make_node(double_pf, n=n, db=db)
+        node.run()
+        table = node.as_table()
         assert sorted(table.column("result").to_pylist()) == [0, 2, 4, 6]
 
     def test_partial_fill_as_table_after_run_on_same_node(self, double_pf):
@@ -401,7 +414,7 @@ def test_is_stale_false_after_clear_cache(self, double_pf):
 
     def test_clear_cache_resets_output_packets(self, double_pf):
         node = _make_node(double_pf, n=3)
-        list(node.iter_packets())
+        node.run()
         assert len(node._cached_output_packets) == 3
 
         node.clear_cache()
@@ -410,6 +423,7 @@ def test_clear_cache_resets_output_packets(self, double_pf):
 
     def test_clear_cache_produces_same_results_on_re_iteration(self, double_pf):
         node = _make_node(double_pf, n=3)
+        node.run()
         table_before = node.as_table()
 
         node.clear_cache()
@@ -451,6 +465,7 @@ def test_as_table_auto_detects_stale_and_repopulates(self, double_pf):
             input_stream=input_stream,
             pipeline_database=db,
         )
+        node.run()
         table_before = node.as_table()
         assert len(table_before) == 3
 

tests/test_core/function_pod/test_pipeline_hash_integration.py

@@ -499,6 +499,7 @@ def test_shared_db_results_are_correct_values(self, double_pf):
             input_stream=make_int_stream(n=5),
             pipeline_database=db,
         )
+        node2.run()
         results = sorted(cast(int, p["result"]) for _, p in node2.iter_packets())
         assert results == [0, 2, 4, 6, 8]
 

tests/test_core/nodes/test_function_node_iteration.py

@@ -0,0 +1,164 @@
+"""Tests for the refactored FunctionNode iteration semantics.
+
+After ENG-379:
+- iter_packets() is strictly read-only — never triggers computation
+- Computation only via run() / execute() / async_execute()
+"""
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from orcapod.core.function_pod import FunctionPod
+from orcapod.core.nodes import FunctionNode
+from orcapod.core.packet_function import PythonPacketFunction
+from orcapod.core.sources import ArrowTableSource
+from orcapod.databases import InMemoryArrowDatabase
+from orcapod.core.executors import LocalExecutor
+
+
+def _make_source(n: int = 3) -> ArrowTableSource:
+    table = pa.table(
+        {
+            "id": pa.array(list(range(n)), type=pa.int64()),
+            "x": pa.array(list(range(n)), type=pa.int64()),
+        },
+        schema=pa.schema(
+            [
+                pa.field("id", pa.int64(), nullable=False),
+                pa.field("x", pa.int64(), nullable=False),
+            ]
+        ),
+    )
+    return ArrowTableSource(table, tag_columns=["id"])
+
+
+def _make_node(n: int = 3, db: InMemoryArrowDatabase | None = None) -> FunctionNode:
+    def double(x: int) -> int:
+        return x * 2
+
+    pf = PythonPacketFunction(double, output_keys="result")
+    pod = FunctionPod(pf)
+    pipeline_db = db if db is not None else InMemoryArrowDatabase()
+    return FunctionNode(pod, _make_source(n=n), pipeline_database=pipeline_db)
+
+
+class TestIterPacketsReadOnly:
+    def test_fresh_node_no_db_yields_nothing(self):
+        """iter_packets() on a fresh node with no run() and empty DB yields nothing."""
+        node = _make_node()
+        assert list(node.iter_packets()) == []
+
+    def test_iter_does_not_call_process_packet_internal(self):
+        """iter_packets() never calls _process_packet_internal under any non-compute path."""
+        node = _make_node()
+        with patch.object(node, "_process_packet_internal") as mock_proc:
+            list(node.iter_packets())
+            mock_proc.assert_not_called()
+
+    def test_iter_after_db_populated_hot_loads_without_compute(self):
+        """iter_packets() on a node with DB records hot-loads without _process_packet_internal."""
+        db = InMemoryArrowDatabase()
+        node1 = _make_node(n=3, db=db)
+        node1.run()  # populate DB
+
+        node2 = _make_node(n=3, db=db)
+        with patch.object(node2, "_process_packet_internal") as mock_proc:
+            results = list(node2.iter_packets())
+            mock_proc.assert_not_called()
+        assert len(results) == 3
+
+    def test_after_run_iter_yields_from_cache_no_db_query(self):
+        """After run(), iter_packets() yields from _cached_output_packets without DB query."""
+        node = _make_node()
+        node.run()
+        initial_count = len(node._cached_output_packets)
+        assert initial_count == 3
+
+        with patch.object(node, "_load_cached_entries") as mock_load:
+            results = list(node.iter_packets())
+            mock_load.assert_not_called()
+        assert len(results) == 3
+
+    def test_iter_twice_same_order_db_queried_once(self):
+        """Two successive iter_packets() calls return same order; DB queried at most once."""
+        db = InMemoryArrowDatabase()
+        node1 = _make_node(n=3, db=db)
+        node1.run()
+
+        node2 = _make_node(n=3, db=db)
+        with patch.object(node2, "_load_cached_entries", wraps=node2._load_cached_entries) as mock_load:
+            first = [(t["id"], p["result"]) for t, p in node2.iter_packets()]
+            second = [(t["id"], p["result"]) for t, p in node2.iter_packets()]
+            assert mock_load.call_count <= 1  # at most one DB query
+        assert first == second
+
+    def test_cached_output_packets_keyed_by_entry_id_strings(self):
+        """After run(), _cached_output_packets keys are entry_id strings, not ints."""
+        node = _make_node()
+        node.run()
+        assert len(node._cached_output_packets) == 3
+        for key in node._cached_output_packets:
+            assert isinstance(key, str), f"Expected str key, got {type(key)}: {key!r}"
+
+    def test_as_table_fresh_node_returns_empty_no_compute(self):
+        """as_table() on a fresh node with no run() and empty DB returns empty table."""
+        node = _make_node()
+        with patch.object(node, "_process_packet_internal") as mock_proc:
+            table = node.as_table()
+            mock_proc.assert_not_called()
+        assert isinstance(table, pa.Table)
+        assert len(table) == 0
+
+    def test_run_cache_only_is_noop(self):
+        """run() on a CACHE_ONLY node returns without error and without computation."""
+        from orcapod.pipeline.serialization import LoadStatus
+
+        node = _make_node()
+        node._load_status = LoadStatus.CACHE_ONLY
+        node._input_stream = None  # simulate no upstream
+
+        with patch.object(node, "execute") as mock_exec:
+            node.run()
+            mock_exec.assert_not_called()
+
+    def test_run_unavailable_raises(self):
+        """run() on an UNAVAILABLE node raises RuntimeError."""
+        from orcapod.pipeline.serialization import LoadStatus
+
+        node = _make_node()
+        node._load_status = LoadStatus.UNAVAILABLE
+        with pytest.raises(RuntimeError, match="unavailable"):
+            node.run()
+
+    def test_execute_error_policy_continue_skips_failures(self):
+        """execute() fires on_packet_crash per failing packet and returns successes when error_policy='continue'.
+
+        Uses LocalExecutor (non-concurrent) to test the sequential execute() path.
+        """
+        errors = []
+
+        def sometimes_fail(x: int) -> int:
+            if x == 1:
+                raise ValueError("intentional failure")
+            return x * 2
+
+        pf = PythonPacketFunction(sometimes_fail, output_keys="result")
+        pf.executor = LocalExecutor()  # sets executor (LocalExecutor.supports_concurrent_execution is False)
+        pod = FunctionPod(pf)
+        db = InMemoryArrowDatabase()
+        node = FunctionNode(pod, _make_source(n=3), pipeline_database=db)
+
+        from orcapod.pipeline.observer import NoOpObserver
+
+        class CapturingObserver(NoOpObserver):
+            def on_packet_crash(self, node_label, tag, packet, exc):
+                errors.append(exc)
+
+        results = node.execute(node._input_stream, observer=CapturingObserver(), error_policy="continue")
+        assert len(errors) == 1
+        assert isinstance(errors[0], ValueError)
+        # Two non-failing packets should succeed
+        assert len(results) == 2