Skip to content

feat(ITL-507): ephemeral function-pod output v1#214

Merged
eywalker merged 36 commits into
mainfrom
agent-kurodo/itl-507-ephemeral-function-pod-output
Jul 8, 2026
Merged

feat(ITL-507): ephemeral function-pod output v1#214
eywalker merged 36 commits into
mainfrom
agent-kurodo/itl-507-ephemeral-function-pod-output

Conversation

@kurodo3

@kurodo3 kurodo3 Bot commented Jul 7, 2026

Copy link
Copy Markdown
Contributor

Summary

  • Adds NodeConfig.is_result_ephemeral: bool = False field to route new computation results to an ArrowDatabaseProtocol store instead of the persistent result DB
  • New set_ephemeral_store(store: ArrowDatabaseProtocol | None) abstract method on FunctionNodeBase/OperatorNodeBase with no-op overrides on blueprint nodes; propagated through FunctionNodeProtocol, OperatorNodeProtocol, PipelineProtocol, and AbstractPipelineBase
  • FunctionJobNode gains ephemeral_result_store slot + _ephemeral_cached_pod (a CachedFunctionPod backed by the ephemeral store); attach_databases accepts ephemeral_database: ArrowDatabaseProtocol | None = None (and pipeline_database is now optional) for one-shot init
  • _fetch_joined_records restructured into two independent Polars inner-joins (one per store) merged with an anti-join giving persistent results priority
  • IS_EPHEMERAL_COL boolean column in the pipeline tag table routes record lookups to the correct store (preserves DATA_RECORD_ID as pa.large_binary())
  • All Phase 2 writes (sync + async) use skip_cache_lookup=True to bypass the pre-write DB lookup
  • Persistent DB miss → WARNING log; ephemeral store miss → silently recomputed (expected cross-session behavior)
  • 21 test scenarios in tests/test_core/function_pod/test_ephemeral_result.py

Fixes ITL-507

Test plan

  • uv run pytest tests/ -q — all tests pass (~4180 passed, 56 skipped, 6 xfailed)
  • test_ephemeral_result_written_to_memory_not_persistent_db — result goes to ephemeral store only
  • test_within_session_ephemeral_hit — second call does not recompute
  • test_cross_session_miss_recomputes — fresh store triggers recompute
  • test_persistent_hit_served_when_ephemeral_true — persistent cache still works when ephemeral store attached
  • test_persistent_result_outcompetes_ephemeral — persistent wins anti-join (verified with direct DB manipulation)
  • test_persistent_miss_warns_and_recomputes — WARNING log emitted when result DB trimmed
  • test_store_not_assigned_raises — clear RuntimeError for is_result_ephemeral=True without store
  • test_ephemeral_only_node_no_persistent_db — works without a result_database

🤖 Generated with Claude Code

kurodo3 Bot and others added 30 commits July 7, 2026 04:57
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ITL-507)

Replace the single-tag Phase 1 lookup pseudocode with the full bulk
resolution mechanism: partition tag table entries by record_id prefix,
join each group against its respective store, union the hits, and derive
the needs-computing set. Add three new test cases covering mixed-store
bulk resolution and silent-drop miss paths.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ITL-507)

Introduce explicit set_ephemeral_store() method on NodeBase (default no-op)
and PersistentFunctionNode (stores the value), plus Pipeline.set_ephemeral_store()
that propagates the store to all nodes uniformly. Replace direct attribute
assignment in the pipeline section with the new method-based API. Update Scope,
Design §2-4, File Layout, and Public API accordingly.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…(ITL-507)

Replace vague "NodeBase / node protocol" references with the actual files:
FunctionNodeProtocol and OperatorNodeProtocol in node_protocols.py, and
PipelineProtocol in pipeline_protocols.py. Update Scope, Design §2/§4,
File Layout, and Public API sections accordingly.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…(ITL-507)

Distinguish the two miss paths in Phase 1 Step 3: a record_id present in the
tag table but absent from the persistent result_database is unexpected (data
loss / external modification) and must emit a WARNING-level log. An ephemeral
store miss is the normal cross-session path and remains silent. Update the
corresponding test name to assert on the warning.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Phase 2 always calls add_pipeline_record with skip_cache_lookup=True so that
a new pipeline DB row is appended even when a stale entry_id already exists.
The inner join in Phase 1 provides natural deduplication — stale rows find no
join partner and are dropped, valid rows survive. Documents the recovery
scenario (both R1 and R2 present) and the ephemeral accumulation behaviour.
Adds two new tests covering the append path and the no-infinite-cycle guarantee.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ITL-507)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…TL-507)

Update all signatures to InMemoryArrowDatabase | None. Passing None detaches
the ephemeral store from the node (or all nodes via pipeline), reverting to
persistent-only behaviour for subsequent writes. Updated node protocol,
PersistentFunctionNode implementation, PipelineProtocol, Pipeline concrete
implementation, and all Public API table rows.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…TL-507)

Step 4 of Phase 1 bulk resolution is now an explicit priority merge:
ephemeral_hits merged first, persistent_hits overwrites on collision.
Persistent results are authoritative — they win whenever both stores have
a valid entry for the same input. Update the recovery scenario note and
add test_persistent_result_outcompetes_ephemeral to the testing plan.
Also add test_set_ephemeral_store_none_detaches (missed in prior commit).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ependent joins (ITL-507)

Step 3 now describes two separate inner joins (one against result_database,
one against ephemeral_result_store), each producing its own entry_id-keyed
dict. Step 4 merges them with persistent_hits second so it overwrites any
collision — this is the concrete mechanism that enforces persistent priority.
Notes that the same entry_id appearing in both dicts only occurs when the tag
table has both a persistent and an ephemeral entry for the same input.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ge (ITL-507)

Replace the Python dict-merge approach with a vectorised Polars anti-join:
ephemeral rows whose entry_id already exists in persistent_df are excluded
by the anti-join, then persistent_df and ephemeral_only_df are concatenated.
O(n+m), zero-copy concat, no Python dict conversion needed. Update the
recovery scenario note to describe the anti-join exclusion mechanism.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…-up (ITL-507)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…or naming consistency

Aligns with the INPUT_DATA_HASH_COL and NODE_CONTENT_HASH_COL pattern where
the module-level constant name matches the SystemConstant property name.
Also strengthens test assertion to pin exact column name.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ratorNodeProtocol, PipelineProtocol

Implements Task 3 of ITL-507 (ephemeral function-pod output v1). Adds the
set_ephemeral_store() method to all node and pipeline protocols, allowing
orchestrators to manage ephemeral result stores. In v1, the implementation
is a no-op for operator nodes (full ephemeral support deferred to ITL-509).

Changes:
- Add InMemoryArrowDatabase to TYPE_CHECKING imports in:
  * src/orcapod/protocols/node_protocols.py
  * src/orcapod/protocols/pipeline_protocols.py
  * src/orcapod/pipeline/graph.py
  * src/orcapod/pipeline/job.py
  * src/orcapod/core/nodes/function_node.py
  * src/orcapod/core/nodes/operator_node.py

- Add set_ephemeral_store() method signatures to protocols:
  * FunctionNodeProtocol
  * OperatorNodeProtocol
  * PipelineProtocol

- Implement set_ephemeral_store() in concrete classes:
  * FunctionNode (no-op for blueprint)
  * FunctionJobNode (no-op in v1, deferred to ITL-508)
  * OperatorNode (no-op for blueprint)
  * OperatorJobNode (no-op in v1, deferred to ITL-509)
  * Pipeline (propagates store to all nodes)
  * PipelineJob (propagates store to all nodes)

- Create tests/test_protocols/test_ephemeral_store_protocol.py with
  basic protocol conformance tests

- Update test_node_protocols.py mock classes to include set_ephemeral_store

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…lasses

Move no-op set_ephemeral_store from concrete FunctionNode/OperatorNode/
OperatorJobNode to FunctionNodeBase and OperatorNodeBase respectively.
Move the pipeline propagation from Pipeline/PipelineJob to
AbstractPipelineBase with a node_type != 'source' guard that prevents
AttributeError on SourceNode when iterating _nodes.values().

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…l_store override to FunctionJobNode

Add ephemeral_result_store and _ephemeral_cached_pod instance variables to FunctionJobNode.__init__,
and implement set_ephemeral_store(store) method to assign or remove the ephemeral result store.
When store is not None, creates a CachedFunctionPod backed by the store so ephemeral writes use
the same format as persistent writes. When store is None, clears both the store and ephemeral pod.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ument read-only node behavior

Remove unnecessary string-quoting from CachedFunctionPod annotation (hard
import at top of file). Add Note section to set_ephemeral_store docstring
explaining the invariant when _function_pod is None.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…rue in add_pipeline_record

When skip_cache_lookup=True we bypass the pre-check, so we must also tell
the database to silently skip duplicates (skip_duplicates=True) rather than
raise ValueError on repeated calls with the same input.
…nd anti-join priority tests

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Mirrors the sync _process_data_internal ephemeral/persistent branch logic
in the async counterpart so that nodes with ephemeral_result=True work
correctly when executed via the async path.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…resolution_mixed_stores

The test was passing config=cfg which silently set _orcapod_config instead
of node_config, meaning ephemeral_result=True was never applied. Fixed to
use node_config=cfg and added an assertion verifying the ephemeral store
was actually written to.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…se _fetch_joined_records priority merge

Replaces the previous test which used a fake UUID and add_pipeline_record
(which silently skipped insertion due to skip_duplicates) with a proper
DB manipulation approach: flush() to commit the persistent row, then insert
the ephemeral row with skip_duplicates=False (succeeds because _pending_record_ids
is empty after flush). Also writes a real ephemeral result to the ephemeral store
so the inner-join produces a row, making ephemeral_df.height > 0 and triggering
the anti-join branch.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…y node tests

Adds TestPersistentMissWarning and TestEphemeralOnlyNode to
test_ephemeral_result.py as part of ITL-507 task 10.
@codecov

codecov Bot commented Jul 7, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 99.02913% with 1 line in your changes missing coverage. Please review.

Files with missing lines Patch % Lines
src/orcapod/core/nodes/function_node.py 98.87% 1 Missing ⚠️

📢 Thoughts on this report? Let us know!

Comment thread src/orcapod/core/nodes/function_node.py Outdated
f"input_stream={self._input_stream!r})"
)

def set_ephemeral_store(self, store: "InMemoryArrowDatabase | None") -> None:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is actually no reason to make it InMemoryArrowDatabase -- we should make it generally accept ArrowDatabaseProtocol

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. Changed type hint from InMemoryArrowDatabase | None to ArrowDatabaseProtocol | None in FunctionNodeBase.set_ephemeral_store, FunctionJobNode.set_ephemeral_store, OperatorNodeBase.set_ephemeral_store, both protocol definitions (FunctionNodeProtocol, OperatorNodeProtocol), PipelineProtocol, and AbstractPipelineBase.

Comment thread src/orcapod/core/nodes/function_node.py Outdated
f"input_stream={self._input_stream!r})"
)

def set_ephemeral_store(self, store: "InMemoryArrowDatabase | None") -> None:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if this is not going to be implemented in the base, I want to make sure we assign this to be an abstraction method

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. FunctionNodeBase.set_ephemeral_store is now decorated with @abstractmethod. Added an explicit no-op override on FunctionNode (the thin blueprint). Did the same for OperatorNodeBase / OperatorNode / OperatorJobNode for consistency.

Comment thread src/orcapod/core/nodes/function_node.py Outdated
self._cached_function_pod: CachedFunctionPod | None = None

# Ephemeral result store (None until set_ephemeral_store() is called by the pipeline)
self.ephemeral_result_store: "InMemoryArrowDatabase | None" = None

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As above, this should be a general ArrowDatabaseProtocol

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. self.ephemeral_result_store in FunctionJobNode.__init__ is now typed as ArrowDatabaseProtocol | None.

Comment thread src/orcapod/core/nodes/function_node.py Outdated
self._ephemeral_cached_pod: CachedFunctionPod | None = None

if pipeline_database is not None:
self.attach_databases(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not just go ahead and make attach_database optionally take in ephemeral_database (and also accept it in the init). This way, you could initialize everything correctly in one shot?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. Added ephemeral_database: ArrowDatabaseProtocol | None = None to both FunctionJobNode.__init__ and attach_databases. When pipeline_database is provided, ephemeral_database is forwarded through attach_databases which calls set_ephemeral_store internally. When only ephemeral_database is provided (no pipeline DB), set_ephemeral_store is called directly.

def attach_databases(
self,
pipeline_database: ArrowDatabaseProtocol,
result_database: ArrowDatabaseProtocol | None = None,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

expand this to optionally accept ephemeral_database: ArrowDatabaseProtocol | None = None. If present, invoke attach ephemeral database on node

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. attach_databases now accepts ephemeral_database: ArrowDatabaseProtocol | None = None and calls self.set_ephemeral_store(ephemeral_database) at the end when it is not None.

Comment thread src/orcapod/core/nodes/function_node.py Outdated
self._function_pod.node_config if self._function_pod is not None else None
)
ephemeral_result = (
node_config.ephemeral_result if node_config is not None else False

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

let's make the property for configuring ephemeral or not to be more obvious, something like node_config.is_result_ephemeral so that it's obvious that it's a boolean flag.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Renamed throughout. NodeConfig.ephemeral_result is now NodeConfig.is_result_ephemeral in types.py, all internal usages in function_node.py (both sync and async paths, error messages, docstrings), and all test files.

eywalker and others added 2 commits July 7, 2026 12:02
…phemeral_store, rename to is_result_ephemeral, add ephemeral_database to attach_databases

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@kurodo3

kurodo3 Bot commented Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Review round 1 — changes addressed

All six comments resolved in commit f802d08:

Comment Change
set_ephemeral_store type: InMemoryArrowDatabase → protocol Changed to `ArrowDatabaseProtocol
set_ephemeral_store should be abstract @abstractmethod added to FunctionNodeBase and OperatorNodeBase; explicit no-op overrides added to FunctionNode, OperatorNode, OperatorJobNode
ephemeral_result_store attribute type Changed to `ArrowDatabaseProtocol
attach_databases / __init__ one-shot init Added `ephemeral_database: ArrowDatabaseProtocol
ephemeral_resultis_result_ephemeral Renamed in NodeConfig (types.py), all internal read sites in function_node.py (sync + async paths, error messages, docstrings), test_node_config.py, and test_ephemeral_result.py

Full test suite: 4168 passed, 56 skipped, 6 xfailed.

Comment thread src/orcapod/core/nodes/function_node.py Outdated

if pipeline_database is not None:
self.attach_databases(
pipeline_database=pipeline_database,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we should simply make attach_databases accept pipeline_database = None and skip setting it on nodes if None

Comment thread src/orcapod/core/nodes/function_node.py Outdated
@@ -720,6 +745,7 @@ def attach_databases(
self,
pipeline_database: ArrowDatabaseProtocol,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just make this pipeline_database: ArrowDatabaseProtocol | None = None

"""
if self._cached_function_pod is not None:
node_config = (
self._function_pod.node_config if self._function_pod is not None else None

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It makes no sense that node_config is attached onto _function_pod. Make a Linear issue to directly follow up and address this.

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This PR introduces an “ephemeral result” mode for function-pod execution nodes so newly computed results can be written to an in-memory store while preserving the existing persistent-cache read path, using a boolean pipeline tag-table column to route lookups to the appropriate store.

Changes:

  • Adds NodeConfig.is_result_ephemeral and plumbs ephemeral-store injection via set_ephemeral_store(...) across node/pipeline protocols and pipeline base.
  • Extends FunctionJobNode with an ephemeral store + CachedFunctionPod and restructures _fetch_joined_records to resolve hits from persistent+ephemeral stores with persistent priority.
  • Adds a large set of tests covering write routing, cache hits/misses, cross-session behavior, and priority resolution.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 3 comments.

Show a summary per file
File Description
tests/test_system_constants.py Adds a small assertion test for the new IS_EPHEMERAL_COL constant value.
tests/test_protocols/test_ephemeral_store_protocol.py Verifies new set_ephemeral_store API presence on key protocols.
tests/test_pipeline/test_node_protocols.py Updates protocol conformance tests to include set_ephemeral_store.
tests/test_core/test_node_config.py Adds tests for the new NodeConfig.is_result_ephemeral field default/behavior/frozenness.
tests/test_core/function_pod/test_ephemeral_result.py Comprehensive end-to-end tests for ephemeral result routing, hits/misses, and store priority.
superpowers/specs/2026-07-07-itl-507-ephemeral-function-pod-output-design.md Design doc describing the ephemeral result store feature and intended semantics.
superpowers/plans/2026-07-07-itl-507-ephemeral-function-pod-output.md Implementation plan detailing tasks, APIs, and test coverage for ITL-507.
src/orcapod/types.py Adds NodeConfig.is_result_ephemeral to control ephemeral result writes.
src/orcapod/system_constants.py Adds constants.IS_EPHEMERAL_COL used to mark tag-table rows as ephemeral/persistent.
src/orcapod/protocols/pipeline_protocols.py Adds PipelineProtocol.set_ephemeral_store(...).
src/orcapod/protocols/node_protocols.py Adds set_ephemeral_store(...) to function/operator node protocols.
src/orcapod/pipeline/base.py Implements AbstractPipelineBase.set_ephemeral_store(...) propagation logic.
src/orcapod/core/nodes/operator_node.py Adds no-op set_ephemeral_store(...) implementations for operator nodes.
src/orcapod/core/nodes/function_node.py Implements ephemeral-store plumbing, write routing, and two-store join resolution in FunctionJobNode.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

entry_id,
combined_record,
skip_duplicates=False,
skip_duplicates=skip_cache_lookup,

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're correct that skip_duplicates=skip_cache_lookup (i.e. skip_duplicates=True) silently skips the write when the entry_id already exists, which leaves a stale pipeline record in place after a cross-session ephemeral-store reset. The stale entry is never overwritten with the new result ID, causing the same entry to re-enter Phase 2 on every subsequent iter_data call.

Flipping to skip_duplicates=not skip_cache_lookup (i.e. skip_duplicates=False when skip_cache_lookup=True) is not viable: it raises a ValueError from InMemoryArrowDatabase.add_records when _process_data_internal is called twice for the same input within the same batch (a scenario covered by test_execute_data_second_call_same_input_deduplicates).

The proper fix is an upsert/overwrite operation on ArrowDatabaseProtocol so that stale pipeline records can be atomically replaced. That requires updating the protocol and all four implementations (InMemoryArrowDatabase, ConnectorArrowDatabase, DeltaLakeArrowDatabase, NoopDatabase), which is out of scope for this PR. Filed ITL-513 to track it.

Comment thread src/orcapod/types.py
Comment on lines +342 to +351
is_result_ephemeral: If ``True``, new computation results are written to
the pipeline-scoped ephemeral store (``ArrowDatabaseProtocol``)
instead of the persistent result database. Persistent cache hits
are still served when available. Raises ``RuntimeError`` at
execution time if ``is_result_ephemeral=True`` but no ephemeral
store has been injected via ``set_ephemeral_store()``.
"""

max_concurrency: int | None = None
is_result_ephemeral: bool = False

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The field was renamed from ephemeral_result to is_result_ephemeral in response to eywalker's review request (round 1, comment #214 (comment): "let's make the property for configuring ephemeral or not to be more obvious, something like node_config.is_result_ephemeral"). The PR description and all call sites have been updated to use the new name. No compatibility alias is provided per project convention (greenfield, pre-v0.1.0).

Comment thread src/orcapod/pipeline/base.py Outdated
@kurodo3

kurodo3 Bot commented Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Review round 3 (Copilot) — changes addressed

Comment Action
skip_duplicates=skip_cache_lookup leaves stale entries No code change + filed ITL-513. The observation is correct — skip_duplicates=True silently skips the write when entry_id already exists, preventing stale pipeline records from being overwritten on ephemeral-store reset. Flipping to skip_duplicates=False crashes when the same input is processed twice in one batch. The proper fix is an upsert operation on ArrowDatabaseProtocol (requires updating all four DB implementations). Filed ITL-513 to track it.
PR description/spec say NodeConfig.ephemeral_result PR description updated. The rename to is_result_ephemeral was intentional (per eywalker's round-1 review request). Updated PR summary and test plan bullets to use the current name throughout. No code change.
set_ephemeral_store docstring says InMemoryArrowDatabase Fixed in commit 77ea8bd. pipeline/base.py docstring now reads ArrowDatabaseProtocol to match the actual parameter type.

Full test suite remains clean: 4180 passed, 56 skipped, 6 xfailed.

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 4 comments.

Comment on lines +64 to +73
"""Assign or remove the ephemeral result store for all nodes.

Propagates ``store`` to every node in the pipeline by calling
``node.set_ephemeral_store(store)`` on each. Nodes that do not
support ephemeral storage (e.g. operator nodes in v1) treat this
as a no-op.

Pass ``None`` to detach the ephemeral store from all nodes,
reverting them to persistent-only writes for subsequent runs.
"""

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. The docstring for PipelineProtocol.set_ephemeral_store now explicitly states that only function-pod and operator-pod nodes receive the store, and that source nodes are skipped because they produce data rather than consume it. See commit 4edd0c56.

Comment on lines +43 to +50
def _make_source_stream(rows: list[dict], tag_columns: list[str] | None = None) -> ArrowTableStream:
if tag_columns is None:
tag_columns = ["id"]
table = pa.table(
{k: pa.array([r[k] for r in rows], type=pa.int64()) for k in rows[0]}
)
source = ArrowTableSource(table, tag_columns=tag_columns, source_id="test_src", infer_nullable=True)
return source

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. _make_source_stream return type annotation corrected from ArrowTableStream to ArrowTableSource, matching the actual value returned. See commit 4edd0c56.

Comment thread src/orcapod/core/nodes/function_node.py Outdated
Comment on lines +1650 to +1658
if results is None:
# Tag table has persistent entries but result DB is empty — data loss
for row_dict in persistent_taginfo_df.to_dicts():
logger.warning(
"Pipeline DB entry '%s' has no match in persistent result DB "
"— data may have been deleted externally. "
"This input will be recomputed.",
row_dict.get(_PIPELINE_ENTRY_ID_COL),
)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. The per-entry loop (for row_dict in persistent_taginfo_df.to_dicts(): logger.warning(...)) has been replaced with a single aggregated warning: logger.warning("%d pipeline DB entries have no match in persistent result DB ...", persistent_taginfo_df.height). This emits at most one log record regardless of how many entries are missing. See commit 4edd0c56.

Comment thread src/orcapod/core/nodes/function_node.py Outdated
Comment on lines +1666 to +1680
# Warn about persistent tag rows that found no match in the result DB
if full_persistent_df.height < persistent_taginfo_df.height:
matched_ids = set(
full_persistent_df.select(_PIPELINE_ENTRY_ID_COL)
.to_series()
.to_list()
)
for row_dict in persistent_taginfo_df.to_dicts():
if row_dict[_PIPELINE_ENTRY_ID_COL] not in matched_ids:
logger.warning(
"Pipeline DB entry '%s' has no match in persistent result DB "
"— data may have been deleted externally. "
"This input will be recomputed.",
row_dict[_PIPELINE_ENTRY_ID_COL],
)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. The second per-entry loop (join-miss case) has also been replaced with a single aggregated warning: we now compute missing_count = persistent_taginfo_df.height - full_persistent_df.height and emit one logger.warning("%d pipeline DB entries have no match ...", missing_count) when missing_count > 0. The matched_ids set construction and inner loop are gone entirely. See commit 4edd0c56.

- Replace two per-entry warning loops in `_fetch_joined_records` with
  single aggregated `logger.warning("%d ... have no match ...", count)`
  calls to avoid log flooding and O(n) overhead.
- Fix `_make_source_stream` return-type annotation: `ArrowTableStream`
  → `ArrowTableSource` (matches actual return value).
- Clarify `PipelineProtocol.set_ephemeral_store` docstring: source nodes
  are skipped (they produce data, not consume it).
- Update test assertion to match new plural wording ("have no match"
  instead of the old per-entry "has no match").

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@kurodo3

kurodo3 Bot commented Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Round 4 (Copilot) — all comments addressed in commit 4edd0c56

Comment Change
PipelineProtocol.set_ephemeral_store docstring says "every node" Updated to "every function-pod and operator-pod node"; added explicit note that source nodes are skipped.
_make_source_stream annotated as → ArrowTableStream but returns ArrowTableSource Return-type annotation corrected to ArrowTableSource.
First per-entry warning loop (results is None branch) floods logs Replaced loop + to_dicts() with one logger.warning("%d ... have no match ...", persistent_taginfo_df.height).
Second per-entry warning loop (join-miss branch) floods logs + builds a set Removed matched_ids set and inner loop; replaced with missing_count = persistent_taginfo_df.height - full_persistent_df.height then one aggregated warning.

The test assertion in TestPersistentMissWarning was also updated from "has no match" to "have no match" to align with the new plural wording.

All 3727 tests pass (uv run pytest tests/ --ignore=tests/test_databases).

Four previously uncovered code paths in the ephemeral function-pod
feature, identified from the codecov patch report:

- TestAsyncEphemeralExecution: async_execute with is_result_ephemeral=True
  (happy path + RuntimeError guard when no store assigned). Covers
  _async_process_data_internal ephemeral branch (lines 1311–1342).

- TestAddPipelineRecordDeduplication: calling add_pipeline_record twice
  with skip_cache_lookup=False verifies the duplicate-detection early
  return fires without inserting a second row (lines 1430, 1436–1439).

- TestFetchJoinedRecordsGuards: get_all_records() on a no-DB node
  returns None (line 1611); and records written without IS_EPHEMERAL_COL
  (legacy/backward-compat) are treated as persistent (lines 1637–1638).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
@eywalker eywalker merged commit 7eeb89c into main Jul 8, 2026
11 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants