From dbeb7d6e3dfcb290e7b51ea8302b64c83fbfc828 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 03:35:18 +0000 Subject: [PATCH 01/15] docs(specs): add design spec for empty table nullability fix (ITL-563) Co-Authored-By: Claude Sonnet 4.6 --- ...26-07-22-empty-table-nullability-design.md | 143 ++++++++++++++++++ 1 file changed, 143 insertions(+) create mode 100644 docs/metamorphic/specs/2026-07-22-empty-table-nullability-design.md diff --git a/docs/metamorphic/specs/2026-07-22-empty-table-nullability-design.md b/docs/metamorphic/specs/2026-07-22-empty-table-nullability-design.md new file mode 100644 index 00000000..7ce8ae1c --- /dev/null +++ b/docs/metamorphic/specs/2026-07-22-empty-table-nullability-design.md @@ -0,0 +1,143 @@ +# Design: Empty Table Nullability Fix (ITL-563) + +**Date:** 2026-07-22 +**Issue:** [ITL-563](https://linear.app/metamorphic/issue/ITL-563/empty-failed-outputs-lose-tag-nullability-and-abort-error) +**Status:** Approved + +## Overview + +When a function fails under `error_policy="continue"`, its empty output table is +constructed by looping over schema fields with `python_type_to_arrow_type` and +passing the resulting dict to `pa.table()`. Because `pa.table(dict_of_arrays)` +marks every field `nullable=True` regardless of the input types, required fields +(`str`, `int`, …) lose their non-nullable annotation. A downstream `Join` then +sees incompatible schemas between its inputs and raises an `InputValidationError` +that aborts the entire pipeline — bypassing the `continue` policy. + +The same pattern appears in five locations across the codebase. + +## Root Cause + +`pa.table(dict_of_arrays)` always produces `nullable=True` fields. The existing +`type_converter.python_schema_to_arrow_schema()` method already handles +nullability correctly (`T` → `nullable=False`, `T | None` → `nullable=True`) +and completes the bidirectional round-trip with `arrow_schema_to_python_schema`, +but it is not used when building empty tables. + +## Goals & Success Criteria + +- Materializing an empty buffer preserves the declared required/optional schema + (verified by reading `ArrowTableStream.output_schema()` after construction). +- A failed function under `error_policy="continue"` feeding a `Join` does not + abort orchestration — the pipeline completes and the join produces zero rows. +- The same nullability preservation holds for empty side-effect stream tables + and empty `DerivedSource` cache tables. +- All five previously buggy construction sites are replaced by a single shared + utility that cannot regress independently. + +## Scope & Boundaries + +**In scope:** +- Add `make_empty_table(python_schema, type_converter)` to `arrow_utils.py`. +- Replace all five buggy sites with calls to the new utility. +- Add unit and integration tests covering nullability preservation and the + `error_policy="continue"` + `Join` pipeline scenario. +- Strengthen the existing `test_as_table_empty_schema_matches_non_empty_schema` + with a nullability assertion. + +**Out of scope:** +- Changing `Join` input-compatibility logic or loosening its validation. +- Making tags optional by default. +- The two `pa.table({})` fallbacks in `operator_node.py` (L1065) and + `tag_data.py` (L346) — these are intentionally schema-free edge cases. +- `async_orchestrator.py` — it uses channels, not materialized buffers, and is + not affected. + +## Architecture + +### New utility: `arrow_utils.make_empty_table` + +```python +# src/orcapod/utils/arrow_utils.py + +def make_empty_table(python_schema: dict, type_converter) -> pa.Table: + """Return a zero-row PyArrow table whose field nullability matches python_schema. + + Uses python_schema_to_arrow_schema so that plain types (str, int, …) produce + nullable=False fields and Optional types (str | None) produce nullable=True + fields. This preserves the round-trip guarantee through + ArrowTableStream.output_schema(). + + Args: + python_schema: Mapping of field name to Python type annotation. + type_converter: A UniversalTypeConverter instance. + + Returns: + A zero-row pa.Table with the correct Arrow schema. + """ + arrow_schema = type_converter.python_schema_to_arrow_schema(python_schema) + return pa.Table.from_batches([], schema=arrow_schema) +``` + +### Call-site changes (5 sites) + +| File | Location | Change | +|---|---|---| +| `src/orcapod/pipeline/sync_orchestrator.py` | L181–198 `_materialize_as_stream()` | Replace 8-line loop + `pa.table()` with `make_empty_table({**tag_schema, **data_schema}, type_converter)` | +| `src/orcapod/core/nodes/operator_node.py` | L835–849 `_make_empty_table()` | Replace body with `make_empty_table({**tag_schema, **data_schema}, self.data_context.type_converter)` | +| `src/orcapod/side_effects.py` | L160–176 `SideEffectFunctionStream.as_table()` | Same one-liner replacement | +| `src/orcapod/side_effects.py` | L720–736 `SideEffectJobFunctionStream.as_table()` | Same one-liner replacement | +| `src/orcapod/core/sources/derived_source.py` | L75–95 DerivedSource cache | Build `python_schema = {k: tag_schema[k] for k in tag_keys}; python_schema.update(data_schema)`, then call `make_empty_table` | + +`operator_node.py`'s `_make_empty_table()` is kept as a thin wrapper (it is +already called from three internal sites) so its callers need no changes. + +### Error handling + +No new error surface. `python_schema_to_arrow_schema` raises `TypeError` on +unsupported types — the same exception that `python_type_to_arrow_type` raises +today. `pa.Table.from_batches([], schema=...)` cannot fail on a valid schema. + +## Testing Plan + +### Unit tests — `tests/test_utils/test_arrow_utils.py` (new file) + +1. `test_make_empty_table_preserves_required_fields` — schema with `str`, `int` + produces `nullable=False` Arrow fields; table has zero rows. +2. `test_make_empty_table_preserves_optional_fields` — schema with `str | None`, + `int | None` produces `nullable=True` fields. +3. `test_make_empty_table_mixed_nullability` — mixed schema (`str`, `int | None`) + produces the correct `nullable` per field. +4. `test_make_empty_table_round_trips_through_arrow_table_stream` — pass the + result to `ArrowTableStream.output_schema()` and assert the Python schema is + identical to the input schema. + +### Integration tests — `tests/test_pipeline/test_error_policy_continue.py` (new file) + +5. `test_failed_function_with_join_does_not_abort_pipeline` — topology from the + issue: `source → failing_function → Join ← source`, under + `error_policy="continue"`. Assert: failing function is logged, Join produces + zero rows, orchestration completes without raising. +6. `test_empty_buffer_schema_preserves_nullability` — run + `_materialize_as_stream()` on a node with an empty buffer; assert the + returned `ArrowTableStream.output_schema()` matches the declared schema + exactly. + +### Regression tests for side_effects and derived_source + +7. `test_side_effect_empty_table_schema_preserves_nullability` — empty + `SideEffectFunctionStream.as_table()` returns a table with correct + nullable/non-nullable fields. +8. `test_derived_source_empty_cache_preserves_nullability` — empty + `DerivedSource` cache table has correct field nullability. + +### Existing test strengthened + +- `test_as_table_empty_schema_matches_non_empty_schema` — add a nullability + assertion: + ```python + assert all( + empty_table.schema.field(n).nullable == full_table.schema.field(n).nullable + for n in empty_table.column_names + ) + ``` From 74de25a631158e9901ebcae4da29b8b61b117622 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 03:44:00 +0000 Subject: [PATCH 02/15] docs(specs): move empty-table nullability spec to superpowers/specs/ (ITL-563) Co-Authored-By: Claude Sonnet 4.6 --- .../specs/2026-07-22-empty-table-nullability-design.md | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename {docs/metamorphic => superpowers}/specs/2026-07-22-empty-table-nullability-design.md (100%) diff --git a/docs/metamorphic/specs/2026-07-22-empty-table-nullability-design.md b/superpowers/specs/2026-07-22-empty-table-nullability-design.md similarity index 100% rename from docs/metamorphic/specs/2026-07-22-empty-table-nullability-design.md rename to superpowers/specs/2026-07-22-empty-table-nullability-design.md From b65069c92469bf891ac9341bd7577ec7038b83aa Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 03:58:13 +0000 Subject: [PATCH 03/15] docs(plans): add implementation plan for empty table nullability fix (ITL-563) Co-Authored-By: Claude Sonnet 4.6 --- DESIGN_ISSUES.md | 24 + .../2026-07-22-empty-table-nullability.md | 730 ++++++++++++++++++ 2 files changed, 754 insertions(+) create mode 100644 superpowers/plans/2026-07-22-empty-table-nullability.md diff --git a/DESIGN_ISSUES.md b/DESIGN_ISSUES.md index 856cec27..33544633 100644 --- a/DESIGN_ISSUES.md +++ b/DESIGN_ISSUES.md @@ -13,6 +13,30 @@ Each item has a status: `open`, `in progress`, or `resolved`. --- +## Cross-cutting + +### CC1 — Empty tables lose field nullability, aborting `error_policy="continue"` pipelines +**Status:** in progress +**Severity:** high +**Issue:** ITL-563 + +Five sites construct empty PyArrow tables by building a dict of `pa.array([], type=...)` and +passing it to `pa.table()`. Because `pa.table(dict_of_arrays)` marks every field +`nullable=True`, required fields (`str`, `int`, …) lose their non-nullable annotation. +When a failing function's empty output is fed into a `Join`, the mismatched schemas raise an +`InputValidationError` that bypasses `error_policy="continue"` and aborts orchestration. + +Sites: `sync_orchestrator.py:_materialize_as_stream`, `operator_node.py:_make_empty_table`, +`side_effects.py:SideEffectFunctionStream.as_table`, +`side_effects.py:SideEffectJobFunctionStream.as_table`, +`derived_source.py:DerivedSource._get_stream`. + +**Fix:** Extract `arrow_utils.make_empty_table(python_schema, type_converter)` using +`python_schema_to_arrow_schema` + `pa.Table.from_batches([], schema=...)` and replace all +five sites. + +--- + ## `src/orcapod/core/nodes/function_node.py` ### FN1 — `FunctionNodeBase.as_table()` returned empty schema when no data existed diff --git a/superpowers/plans/2026-07-22-empty-table-nullability.md b/superpowers/plans/2026-07-22-empty-table-nullability.md new file mode 100644 index 00000000..79ee505b --- /dev/null +++ b/superpowers/plans/2026-07-22-empty-table-nullability.md @@ -0,0 +1,730 @@ +# Empty Table Nullability Fix Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use sensei:subagent-driven-development (recommended) or sensei:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Fix five empty-table construction sites that silently drop field nullability, causing `error_policy="continue"` pipelines to abort when a failing function feeds a `Join`. + +**Architecture:** Add a single `make_empty_table(python_schema, type_converter)` utility to `arrow_utils.py` that uses the existing `python_schema_to_arrow_schema` method (which correctly preserves `nullable=False` for required fields). Replace all five buggy call sites with this utility. Write tests first at every step. + +**Tech Stack:** Python, PyArrow, pytest, `uv run` + +--- + +## File Map + +| Action | Path | Purpose | +|---|---|---| +| Modify | `src/orcapod/utils/arrow_utils.py` | Add `make_empty_table()` | +| Modify | `src/orcapod/core/nodes/operator_node.py` | Fix `_make_empty_table()` body | +| Modify | `src/orcapod/pipeline/sync_orchestrator.py` | Fix `_materialize_as_stream()` empty branch | +| Modify | `src/orcapod/side_effects.py` | Fix `SideEffectFunctionStream.as_table()` and `SideEffectJobFunctionStream.as_table()` | +| Modify | `src/orcapod/core/sources/derived_source.py` | Fix `DerivedSource._get_stream()` empty branch | +| Modify | `tests/test_utils/test_arrow_utils.py` | Add `make_empty_table` unit tests | +| Create | `tests/test_pipeline/test_error_policy_continue.py` | Integration test for the bug scenario | +| Modify | `tests/test_core/nodes/test_function_node_iteration.py` | Strengthen nullability assertion | +| Modify | `tests/test_core/sources/test_derived_source.py` | Add nullability regression test | +| Modify | `tests/test_core/side_effect_function/test_side_effect_function_pod.py` | Add nullability regression test | +| Modify | `DESIGN_ISSUES.md` | Mark CC1 resolved | + +--- + +## Task 1: Create the feature branch + +**Files:** none (git only) + +- [ ] **Step 1: Check out the feature branch** + +```bash +cd /path/to/orcapod-python # your actual working directory +git checkout -b eywalker/itl-563-empty-failed-outputs-lose-tag-nullability-and-abort +git branch --show-current +``` + +Expected output: `eywalker/itl-563-empty-failed-outputs-lose-tag-nullability-and-abort` + +--- + +## Task 2: Unit tests for `make_empty_table` (write tests first, implementation not yet added) + +**Files:** +- Modify: `tests/test_utils/test_arrow_utils.py` + +- [ ] **Step 1: Write four failing tests at the bottom of the existing test file** + +Append to `tests/test_utils/test_arrow_utils.py`: + +```python +# --------------------------------------------------------------------------- +# make_empty_table — ITL-563 +# --------------------------------------------------------------------------- + + +class TestMakeEmptyTable: + """make_empty_table preserves field nullability from python_schema.""" + + def _converter(self): + from orcapod.contexts import DataContext + return DataContext().type_converter + + def test_required_fields_are_non_nullable(self): + """Plain types produce nullable=False Arrow fields.""" + from orcapod.utils.arrow_utils import make_empty_table + + schema = {"name": str, "count": int} + table = make_empty_table(schema, self._converter()) + + assert table.num_rows == 0 + assert table.schema.field("name").nullable is False + assert table.schema.field("count").nullable is False + + def test_optional_fields_are_nullable(self): + """Optional types produce nullable=True Arrow fields.""" + from orcapod.utils.arrow_utils import make_empty_table + + schema = {"name": str | None, "count": int | None} + table = make_empty_table(schema, self._converter()) + + assert table.num_rows == 0 + assert table.schema.field("name").nullable is True + assert table.schema.field("count").nullable is True + + def test_mixed_nullability_per_field(self): + """Mixed schema: required field non-nullable, optional field nullable.""" + from orcapod.utils.arrow_utils import make_empty_table + + schema = {"subject": str, "score": int | None} + table = make_empty_table(schema, self._converter()) + + assert table.num_rows == 0 + assert table.schema.field("subject").nullable is False + assert table.schema.field("score").nullable is True + + def test_round_trips_through_arrow_table_stream(self): + """Python schema → make_empty_table → ArrowTableStream.output_schema() is identity.""" + from orcapod.utils.arrow_utils import make_empty_table + from orcapod.core.streams import ArrowTableStream + + converter = self._converter() + python_schema = {"subject": str, "score": int | None} + table = make_empty_table(python_schema, converter) + + # ArrowTableStream requires at least one data column; tag=subject, data=score + stream = ArrowTableStream(table, tag_columns=["subject"]) + _, data_schema = stream.output_schema() + + assert data_schema["score"] == (int | None) +``` + +- [ ] **Step 2: Run tests to verify they fail (function not yet defined)** + +```bash +uv run pytest tests/test_utils/test_arrow_utils.py::TestMakeEmptyTable -v +``` + +Expected: 4 failures with `ImportError` or `cannot import name 'make_empty_table'`. + +--- + +## Task 3: Implement `make_empty_table` in `arrow_utils.py` + +**Files:** +- Modify: `src/orcapod/utils/arrow_utils.py` + +- [ ] **Step 1: Add the function after the last import block, before `schema_select`** + +In `src/orcapod/utils/arrow_utils.py`, insert after line 14 (the `pa = LazyModule("pyarrow")` line) and before `def schema_select`: + +```python + +def make_empty_table(python_schema: "Mapping[str, Any]", type_converter: Any) -> "pa.Table": + """Return a zero-row PyArrow table whose field nullability matches ``python_schema``. + + Uses ``python_schema_to_arrow_schema`` so that plain types (``str``, ``int``, …) + produce ``nullable=False`` fields and Optional types (``str | None``) produce + ``nullable=True`` fields. This preserves the bidirectional round-trip through + ``ArrowTableStream.output_schema()``. + + Args: + python_schema: Mapping of field name to Python type annotation. + type_converter: A ``UniversalTypeConverter`` instance. + + Returns: + A zero-row ``pa.Table`` with the correct Arrow schema. + """ + arrow_schema = type_converter.python_schema_to_arrow_schema(python_schema) + return pa.Table.from_batches([], schema=arrow_schema) +``` + +Also update the `Mapping` import at the top — it is already imported from `collections.abc`. + +- [ ] **Step 2: Run unit tests to verify they pass** + +```bash +uv run pytest tests/test_utils/test_arrow_utils.py::TestMakeEmptyTable -v +``` + +Expected: 4 tests PASS. + +- [ ] **Step 3: Commit** + +```bash +git add src/orcapod/utils/arrow_utils.py tests/test_utils/test_arrow_utils.py +git commit -m "feat(arrow_utils): add make_empty_table preserving field nullability (ITL-563)" +``` + +--- + +## Task 4: Fix `operator_node.py` — `_make_empty_table()` + +**Files:** +- Modify: `src/orcapod/core/nodes/operator_node.py` +- Modify: `tests/test_core/nodes/test_function_node_iteration.py` + +`operator_node.py` already has a `_make_empty_table()` method (lines 835–849) called from +three internal sites. We replace its body with a call to `arrow_utils.make_empty_table()`. + +- [ ] **Step 1: Strengthen the existing test to assert nullability (it currently only checks column names)** + +In `tests/test_core/nodes/test_function_node_iteration.py`, update `test_as_table_empty_schema_matches_non_empty_schema`: + +```python +def test_as_table_empty_schema_matches_non_empty_schema(): + """as_table() empty table has the same columns and nullability as the populated table.""" + db = InMemoryArrowDatabase() + node_after = _make_node(db=db) + node_after.run() + full_table = node_after.as_table() + + node_before = _make_node() + empty_table = node_before.as_table() + + assert empty_table.num_rows == 0 + assert full_table.num_rows > 0 + assert set(empty_table.column_names) == set(full_table.column_names) + # Nullability must match — this failed before the fix + assert all( + empty_table.schema.field(n).nullable == full_table.schema.field(n).nullable + for n in empty_table.column_names + ) +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +uv run pytest tests/test_core/nodes/test_function_node_iteration.py::test_as_table_empty_schema_matches_non_empty_schema -v +``` + +Expected: FAIL — `nullable` flags differ for required fields. + +- [ ] **Step 3: Fix the body of `_make_empty_table()` in `operator_node.py`** + +Replace the current body of `_make_empty_table()` (lines 835–849) with: + +```python + def _make_empty_table(self) -> "pa.Table": + """Build a zero-row PyArrow table matching this node's full output schema. + + Uses ``arrow_utils.make_empty_table`` so field nullability is preserved + (``nullable=False`` for required fields, ``nullable=True`` for optional). + Requires ``self._operator is not None`` (pre-existing limitation shared + with ``_replay_from_cache``). + """ + from orcapod.utils.arrow_utils import make_empty_table + + tag_schema, data_schema = self.output_schema() + return make_empty_table( + {**tag_schema, **data_schema}, + self.data_context.type_converter, + ) +``` + +- [ ] **Step 4: Run the test to verify it passes** + +```bash +uv run pytest tests/test_core/nodes/test_function_node_iteration.py::test_as_table_empty_schema_matches_non_empty_schema -v +``` + +Expected: PASS. + +- [ ] **Step 5: Run the full `test_core/nodes` suite to check for regressions** + +```bash +uv run pytest tests/test_core/nodes/ -v +``` + +Expected: all PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/orcapod/core/nodes/operator_node.py tests/test_core/nodes/test_function_node_iteration.py +git commit -m "fix(operator_node): use make_empty_table to preserve field nullability (ITL-563)" +``` + +--- + +## Task 5: Fix `sync_orchestrator.py` — `_materialize_as_stream()` + integration test + +**Files:** +- Create: `tests/test_pipeline/test_error_policy_continue.py` +- Modify: `src/orcapod/pipeline/sync_orchestrator.py` + +- [ ] **Step 1: Write the failing integration tests** + +Create `tests/test_pipeline/test_error_policy_continue.py`: + +```python +"""Integration tests for error_policy='continue' + Join schema compatibility. + +Regression for ITL-563: failed functions produced empty tables with wrong +nullable flags, causing Join to raise InputValidationError and abort the pipeline. +""" +from __future__ import annotations + +import pyarrow as pa +import pytest + +from orcapod.core.data_function import PythonDataFunction +from orcapod.core.function_pod import FunctionPod +from orcapod.core.operators.join import Join +from orcapod.core.sources import ArrowTableSource +from orcapod.databases import InMemoryArrowDatabase +from orcapod.pipeline.job import PipelineJob + + +def _make_source_with_required_field(subjects: list[str], values: list[int]) -> ArrowTableSource: + """Source with tag=subject (str, non-nullable) and data=value (int, non-nullable).""" + schema = pa.schema([ + pa.field("subject", pa.large_string(), nullable=False), + pa.field("value", pa.int64(), nullable=False), + ]) + table = pa.table( + {"subject": subjects, "value": values}, + schema=schema, + ) + return ArrowTableSource(table, tag_columns=["subject"]) + + +def test_failed_function_with_join_does_not_abort_pipeline(): + """Topology: source → failing_function → Join ← source. + + Under error_policy='continue', the failing function should be logged, + the Join should produce zero rows (empty × non-empty = empty), and + orchestration should complete without raising. + """ + src = _make_source_with_required_field(["a", "b"], [1, 2]) + + def always_fails(value: int) -> int: + raise RuntimeError("intentional failure") + + pf = PythonDataFunction(always_fails, output_keys="transformed") + failing_pod = FunctionPod(pf) + + job = PipelineJob(name="test_join_continue", store=InMemoryArrowDatabase()) + with job: + failing_out = failing_pod(src, label="failing") + Join()(failing_out, src, label="joined") + + # Should complete without raising despite the failing function + job.run(error_policy="continue") + + joined_records = job.nodes["joined"].get_all_records() + # Join of empty × non-empty = empty (zero rows), not an error + assert joined_records is None or joined_records.num_rows == 0 + + +def test_empty_buffer_schema_preserves_nullability(): + """_materialize_as_stream on an empty buffer preserves required field nullability.""" + from orcapod.pipeline.sync_orchestrator import SyncPipelineOrchestrator + from orcapod.core.nodes.function_node import FunctionJobNode + + src = _make_source_with_required_field(["x"], [10]) + + def identity(value: int) -> int: + return value + + pf = PythonDataFunction(identity, output_keys="result") + pod = FunctionPod(pf) + + db = InMemoryArrowDatabase() + job = PipelineJob(name="test_empty_schema", store=db) + with job: + pod(src, label="fn") + + # Don't run — buffer will be empty + node = job.nodes["fn"] + stream = SyncPipelineOrchestrator._materialize_as_stream([], node) + tag_schema, data_schema = stream.output_schema() + + # "result" is declared as int (required), must NOT become int | None + assert data_schema["result"] == int +``` + +- [ ] **Step 2: Run the tests to verify they fail** + +```bash +uv run pytest tests/test_pipeline/test_error_policy_continue.py -v +``` + +Expected: `test_failed_function_with_join_does_not_abort_pipeline` raises `InputValidationError`; `test_empty_buffer_schema_preserves_nullability` asserts `int | None` but expects `int`. + +- [ ] **Step 3: Fix `_materialize_as_stream()` in `sync_orchestrator.py`** + +In `src/orcapod/pipeline/sync_orchestrator.py`, replace the empty-branch of +`_materialize_as_stream()` (lines 181–198). The current code is: + +```python + if not buf: + # Build an empty stream with the correct schema from the upstream node + tag_schema, data_schema = upstream_node.output_schema( + columns={"system_tags": True, "source": True} + ) + type_converter = upstream_node.data_context.type_converter + empty_fields = {} + for name, py_type in {**tag_schema, **data_schema}.items(): + arrow_type = type_converter.python_type_to_arrow_type(py_type) + empty_fields[name] = pa.array([], type=arrow_type) + empty_table = pa.table(empty_fields) + tag_keys = upstream_node.keys()[0] + return ArrowTableStream( + empty_table, + tag_columns=tag_keys, + producer=upstream_node.producer, + upstreams=upstream_node.upstreams, + ) +``` + +Replace with: + +```python + if not buf: + # Build an empty stream preserving declared field nullability. + tag_schema, data_schema = upstream_node.output_schema( + columns={"system_tags": True, "source": True} + ) + type_converter = upstream_node.data_context.type_converter + empty_table = arrow_utils.make_empty_table( + {**tag_schema, **data_schema}, type_converter + ) + tag_keys = upstream_node.keys()[0] + return ArrowTableStream( + empty_table, + tag_columns=tag_keys, + producer=upstream_node.producer, + upstreams=upstream_node.upstreams, + ) +``` + +Note: `arrow_utils` is already imported at line 176 (`from orcapod.utils import arrow_utils`). + +- [ ] **Step 4: Run the integration tests to verify they pass** + +```bash +uv run pytest tests/test_pipeline/test_error_policy_continue.py -v +``` + +Expected: both PASS. + +- [ ] **Step 5: Run the full pipeline test suite to check for regressions** + +```bash +uv run pytest tests/test_pipeline/ -v +``` + +Expected: all PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/orcapod/pipeline/sync_orchestrator.py tests/test_pipeline/test_error_policy_continue.py +git commit -m "fix(sync_orchestrator): use make_empty_table to preserve field nullability (ITL-563)" +``` + +--- + +## Task 6: Fix `side_effects.py` — two `as_table()` methods + +**Files:** +- Modify: `src/orcapod/side_effects.py` +- Modify: `tests/test_core/side_effect_function/test_side_effect_function_pod.py` + +- [ ] **Step 1: Write a failing test for `SideEffectFunctionStream.as_table()`** + +Append to `tests/test_core/side_effect_function/test_side_effect_function_pod.py`: + +```python +class TestSideEffectEmptyTableNullability: + """Regression for ITL-563: empty as_table() must preserve field nullability.""" + + def test_empty_stream_schema_preserves_required_fields(self): + """SideEffectFunctionStream.as_table() with no rows preserves nullable=False.""" + from orcapod.core.function_pod import FunctionPod + from orcapod.side_effects import InvocationContext + + def my_fn(value: int, ctx: InvocationContext) -> str: + return str(value) + + pod = FunctionPod.from_fn(my_fn, output_keys=["result"], ctx_arg_name="ctx") + stream = _make_stream(n=0) # empty input → empty output + + # Build an empty stream (iter_data yields nothing) + from orcapod.side_effects import SideEffectFunctionStream + se_stream = SideEffectFunctionStream(pod=pod, input_stream=stream) + + table = se_stream.as_table() + assert table.num_rows == 0 + # "result" is declared as str (required) — must not become str | None + assert table.schema.field("result").nullable is False +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +uv run pytest "tests/test_core/side_effect_function/test_side_effect_function_pod.py::TestSideEffectEmptyTableNullability" -v +``` + +Expected: FAIL — `nullable` is `True` but should be `False`. + +- [ ] **Step 3: Fix `SideEffectFunctionStream.as_table()` in `side_effects.py`** + +Locate the empty-branch inside `SideEffectFunctionStream.as_table()` (around lines 165–176). +Current code: + +```python + if not tag_tables: + # Return an empty table with the correct schema + tag_schema, data_schema = self.output_schema( + columns=column_config + ) + tc = self._pod.data_context.type_converter + fields = {} + for name, py_type in {**tag_schema, **data_schema}.items(): + fields[name] = pa.array( + [], type=tc.python_type_to_arrow_type(py_type) + ) + return pa.table(fields) +``` + +Replace with: + +```python + if not tag_tables: + # Return an empty table preserving declared field nullability. + tag_schema, data_schema = self.output_schema( + columns=column_config + ) + tc = self._pod.data_context.type_converter + return arrow_utils.make_empty_table( + {**tag_schema, **data_schema}, tc + ) +``` + +Note: `arrow_utils` is already imported in the `as_table()` body (`from orcapod.utils import arrow_utils`). + +- [ ] **Step 4: Fix `SideEffectJobFunctionStream.as_table()` in `side_effects.py`** + +Locate the empty-branch inside `SideEffectJobFunctionStream.as_table()` (around lines 728–736). +Current code: + +```python + if not tag_tables: + tag_schema, data_schema = self.output_schema(columns=column_config) + tc = self._pod.data_context.type_converter + fields = {} + for name, py_type in {**tag_schema, **data_schema}.items(): + fields[name] = pa.array( + [], type=tc.python_type_to_arrow_type(py_type) + ) + return pa.table(fields) +``` + +Replace with: + +```python + if not tag_tables: + tag_schema, data_schema = self.output_schema(columns=column_config) + tc = self._pod.data_context.type_converter + return arrow_utils.make_empty_table( + {**tag_schema, **data_schema}, tc + ) +``` + +Note: `arrow_utils` is already imported in this method's body too. + +- [ ] **Step 5: Run the test to verify it passes** + +```bash +uv run pytest "tests/test_core/side_effect_function/test_side_effect_function_pod.py::TestSideEffectEmptyTableNullability" -v +``` + +Expected: PASS. + +- [ ] **Step 6: Run the full side-effect test suite to check for regressions** + +```bash +uv run pytest tests/test_core/side_effect_function/ tests/test_core/side_effect_pod/ -v +``` + +Expected: all PASS. + +- [ ] **Step 7: Commit** + +```bash +git add src/orcapod/side_effects.py tests/test_core/side_effect_function/test_side_effect_function_pod.py +git commit -m "fix(side_effects): use make_empty_table to preserve field nullability (ITL-563)" +``` + +--- + +## Task 7: Fix `derived_source.py` — `DerivedSource._get_stream()` + +**Files:** +- Modify: `src/orcapod/core/sources/derived_source.py` +- Modify: `tests/test_core/sources/test_derived_source.py` + +- [ ] **Step 1: Write a failing test for `DerivedSource` empty cache nullability** + +Append to `tests/test_core/sources/test_derived_source.py`: + +```python +def test_derived_source_empty_cache_preserves_nullability(): + """DerivedSource._get_stream() empty table preserves nullable=False for required fields. + + Regression for ITL-563: pa.field() without nullable=False defaulted to True. + """ + # _make_node() builds a FunctionJobNode with data output "result: int" (required). + node = _make_node() + source = node.as_source() # DerivedSource backed by unrun node → empty cache + + # as_table() triggers _get_stream() which hits the empty-cache branch + table = source.as_table() + + assert table.num_rows == 0 + # "result" is int (required), must not become int | None + result_field = table.schema.field("result") + assert result_field.nullable is False +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +uv run pytest tests/test_core/sources/test_derived_source.py::test_derived_source_empty_cache_preserves_nullability -v +``` + +Expected: FAIL — `nullable` is `True` but should be `False`. + +- [ ] **Step 3: Fix `DerivedSource._get_stream()` in `derived_source.py`** + +Locate the `records is None` branch in `_get_stream()` (lines 75–95). Current code: + +```python + if records is None: + # Build empty table with correct schema + tag_schema, data_schema = self._origin.output_schema() + tag_keys = self._origin.keys()[0] + tc = self.data_context.type_converter + fields = [ + pa.field(k, tc.python_type_to_arrow_type(tag_schema[k])) + for k in tag_keys + ] + fields += [ + pa.field(k, tc.python_type_to_arrow_type(v)) + for k, v in data_schema.items() + ] + arrow_schema = pa.schema(fields) + self._cached_table = pa.table( + {f.name: pa.array([], type=f.type) for f in arrow_schema}, + schema=arrow_schema, + ) +``` + +Replace with: + +```python + if records is None: + # Build empty table preserving declared field nullability. + from orcapod.utils.arrow_utils import make_empty_table + + tag_schema, data_schema = self._origin.output_schema() + tag_keys = self._origin.keys()[0] + tc = self.data_context.type_converter + python_schema = {k: tag_schema[k] for k in tag_keys} + python_schema.update(data_schema) + self._cached_table = make_empty_table(python_schema, tc) +``` + +- [ ] **Step 4: Run the test to verify it passes** + +```bash +uv run pytest tests/test_core/sources/test_derived_source.py::test_derived_source_empty_cache_preserves_nullability -v +``` + +Expected: PASS. + +- [ ] **Step 5: Run the full sources test suite to check for regressions** + +```bash +uv run pytest tests/test_core/sources/ -v +``` + +Expected: all PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/orcapod/core/sources/derived_source.py tests/test_core/sources/test_derived_source.py +git commit -m "fix(derived_source): use make_empty_table to preserve field nullability (ITL-563)" +``` + +--- + +## Task 8: Mark DESIGN_ISSUES.md resolved + run full test suite + +**Files:** +- Modify: `DESIGN_ISSUES.md` + +- [ ] **Step 1: Mark CC1 resolved in `DESIGN_ISSUES.md`** + +Update the CC1 entry: change `**Status:** in progress` to `**Status:** resolved` and append a **Fix:** line: + +``` +**Fix:** Extracted ``arrow_utils.make_empty_table(python_schema, type_converter)`` using +``python_schema_to_arrow_schema`` + ``pa.Table.from_batches([], schema=...)``. Replaced all +five buggy sites. Added unit tests in ``test_arrow_utils.py``, integration test in +``test_error_policy_continue.py``, and per-site regression tests. PR: ITL-563. +``` + +- [ ] **Step 2: Run the full test suite** + +```bash +uv run pytest tests/ -v --tb=short +``` + +Expected: all PASS (or pre-existing failures only — check against `git stash` baseline if any fail). + +- [ ] **Step 3: Final commit** + +```bash +git add DESIGN_ISSUES.md +git commit -m "docs(design-issues): mark CC1 resolved — empty table nullability fix (ITL-563)" +``` + +--- + +## Self-Review Checklist (completed inline) + +| Spec requirement | Covered by task | +|---|---| +| `make_empty_table` in `arrow_utils.py` | Task 3 | +| Fix `sync_orchestrator._materialize_as_stream` | Task 5 | +| Fix `operator_node._make_empty_table` | Task 4 | +| Fix `SideEffectFunctionStream.as_table` | Task 6 | +| Fix `SideEffectJobFunctionStream.as_table` | Task 6 | +| Fix `DerivedSource._get_stream` | Task 7 | +| Unit tests: required / optional / mixed / round-trip | Task 2 | +| Integration test: failing fn + Join + continue policy | Task 5 | +| Integration test: empty buffer schema preserves nullability | Task 5 | +| Regression: side-effect empty table nullability | Task 6 | +| Regression: DerivedSource empty cache nullability | Task 7 | +| Strengthen `test_as_table_empty_schema_matches_non_empty_schema` | Task 4 | +| DESIGN_ISSUES.md updated | Tasks 1 (in progress), 8 (resolved) | From 4fffc27c306ec0aca76d163b34b97fbd9fa9834b Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:04:46 +0000 Subject: [PATCH 04/15] feat(arrow_utils): add make_empty_table preserving field nullability (ITL-563) Adds make_empty_table() which uses python_schema_to_arrow_schema() to build zero-row Arrow tables with correct nullable flags (nullable=False for plain types, nullable=True for T | None), fixing the nullability loss that caused downstream Join operators to raise InputValidationError on empty outputs. Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/utils/arrow_utils.py | 19 +++++++++ tests/test_utils/test_arrow_utils.py | 61 ++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+) diff --git a/src/orcapod/utils/arrow_utils.py b/src/orcapod/utils/arrow_utils.py index 369da869..afd05576 100644 --- a/src/orcapod/utils/arrow_utils.py +++ b/src/orcapod/utils/arrow_utils.py @@ -14,6 +14,25 @@ pa = LazyModule("pyarrow") +def make_empty_table(python_schema: "Mapping[str, Any]", type_converter: Any) -> "pa.Table": + """Return a zero-row PyArrow table whose field nullability matches ``python_schema``. + + Uses ``python_schema_to_arrow_schema`` so that plain types (``str``, ``int``, …) + produce ``nullable=False`` fields and Optional types (``str | None``) produce + ``nullable=True`` fields. This preserves the bidirectional round-trip through + ``ArrowTableStream.output_schema()``. + + Args: + python_schema: Mapping of field name to Python type annotation. + type_converter: A ``UniversalTypeConverter`` instance. + + Returns: + A zero-row ``pa.Table`` with the correct Arrow schema. + """ + arrow_schema = type_converter.python_schema_to_arrow_schema(python_schema) + return pa.Table.from_batches([], schema=arrow_schema) + + def schema_select( arrow_schema: "pa.Schema", column_names: Collection[str], diff --git a/tests/test_utils/test_arrow_utils.py b/tests/test_utils/test_arrow_utils.py index e77b193a..385c399d 100644 --- a/tests/test_utils/test_arrow_utils.py +++ b/tests/test_utils/test_arrow_utils.py @@ -731,3 +731,64 @@ def test_multiple_extension_columns_all_normalized(self): # Data values correct for both assert result.column("i").to_pylist() == [1, 2] assert result.column("b").to_pylist() == [b"x", b"y"] + + +# --------------------------------------------------------------------------- +# make_empty_table — ITL-563 +# --------------------------------------------------------------------------- + + +class TestMakeEmptyTable: + """make_empty_table preserves field nullability from python_schema.""" + + def _converter(self): + from orcapod.contexts import get_default_type_converter + return get_default_type_converter() + + def test_required_fields_are_non_nullable(self): + """Plain types produce nullable=False Arrow fields.""" + from orcapod.utils.arrow_utils import make_empty_table + + schema = {"name": str, "count": int} + table = make_empty_table(schema, self._converter()) + + assert table.num_rows == 0 + assert table.schema.field("name").nullable is False + assert table.schema.field("count").nullable is False + + def test_optional_fields_are_nullable(self): + """Optional types produce nullable=True Arrow fields.""" + from orcapod.utils.arrow_utils import make_empty_table + + schema = {"name": str | None, "count": int | None} + table = make_empty_table(schema, self._converter()) + + assert table.num_rows == 0 + assert table.schema.field("name").nullable is True + assert table.schema.field("count").nullable is True + + def test_mixed_nullability_per_field(self): + """Mixed schema: required field non-nullable, optional field nullable.""" + from orcapod.utils.arrow_utils import make_empty_table + + schema = {"subject": str, "score": int | None} + table = make_empty_table(schema, self._converter()) + + assert table.num_rows == 0 + assert table.schema.field("subject").nullable is False + assert table.schema.field("score").nullable is True + + def test_round_trips_through_arrow_table_stream(self): + """Python schema → make_empty_table → ArrowTableStream.output_schema() is identity.""" + from orcapod.utils.arrow_utils import make_empty_table + from orcapod.core.streams import ArrowTableStream + + converter = self._converter() + python_schema = {"subject": str, "score": int | None} + table = make_empty_table(python_schema, converter) + + # ArrowTableStream requires at least one data column; tag=subject, data=score + stream = ArrowTableStream(table, tag_columns=["subject"]) + _, data_schema = stream.output_schema() + + assert data_schema["score"] == (int | None) From da38f9b7c7c38b755a90e3d98f4ad7581e2a6bd8 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:07:50 +0000 Subject: [PATCH 05/15] refactor(arrow_utils): type type_converter as TypeConverterProtocol in make_empty_table (ITL-563) --- src/orcapod/utils/arrow_utils.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/orcapod/utils/arrow_utils.py b/src/orcapod/utils/arrow_utils.py index afd05576..f3a1ca05 100644 --- a/src/orcapod/utils/arrow_utils.py +++ b/src/orcapod/utils/arrow_utils.py @@ -10,11 +10,12 @@ if TYPE_CHECKING: import pyarrow as pa + from orcapod.protocols.semantic_types_protocols import TypeConverterProtocol else: pa = LazyModule("pyarrow") -def make_empty_table(python_schema: "Mapping[str, Any]", type_converter: Any) -> "pa.Table": +def make_empty_table(python_schema: "Mapping[str, Any]", type_converter: "TypeConverterProtocol") -> "pa.Table": """Return a zero-row PyArrow table whose field nullability matches ``python_schema``. Uses ``python_schema_to_arrow_schema`` so that plain types (``str``, ``int``, …) From 7a37f9f2689771da79fd030a719f38a25dd74104 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:10:42 +0000 Subject: [PATCH 06/15] fix(operator_node): use make_empty_table to preserve field nullability (ITL-563) Replace the old _make_empty_table() body (python_type_to_arrow_type loop + pa.table(dict), which lost nullable=False for plain types) with a call to arrow_utils.make_empty_table(), which uses python_schema_to_arrow_schema and pa.Table.from_batches to correctly set nullable per field. Also strengthen test_as_table_empty_schema_matches_non_empty_schema to assert that field nullability matches between the empty and populated tables. --- src/orcapod/core/nodes/operator_node.py | 18 ++++++++++-------- .../nodes/test_function_node_iteration.py | 7 ++++++- 2 files changed, 16 insertions(+), 9 deletions(-) diff --git a/src/orcapod/core/nodes/operator_node.py b/src/orcapod/core/nodes/operator_node.py index 84984877..859cb5c2 100644 --- a/src/orcapod/core/nodes/operator_node.py +++ b/src/orcapod/core/nodes/operator_node.py @@ -835,18 +835,20 @@ def _store_output_stream(self, stream: StreamProtocol) -> None: def _make_empty_table(self) -> "pa.Table": """Build a zero-row PyArrow table matching this node's full output schema. - Uses ``output_schema()`` for column names/types and - ``data_context.type_converter`` for the Python → Arrow type mapping. + Uses ``arrow_utils.make_empty_table`` so field nullability is preserved: + plain types produce ``nullable=False`` fields and ``T | None`` types + produce ``nullable=True`` fields. + Requires ``self._operator is not None`` (pre-existing limitation shared with ``_replay_from_cache``). """ + from orcapod.utils.arrow_utils import make_empty_table + tag_schema, data_schema = self.output_schema() - type_converter = self.data_context.type_converter - empty_fields: dict = {} - for name, py_type in {**tag_schema, **data_schema}.items(): - arrow_type = type_converter.python_type_to_arrow_type(py_type) - empty_fields[name] = pa.array([], type=arrow_type) - return pa.table(empty_fields) + return make_empty_table( + {**tag_schema, **data_schema}, + self.data_context.type_converter, + ) def _load_cached_stream_from_db(self) -> "ArrowTableStream | None": """Read from DB in CacheMode.REPLAY only, without modifying node state. diff --git a/tests/test_core/nodes/test_function_node_iteration.py b/tests/test_core/nodes/test_function_node_iteration.py index d22e2dfa..e1a46f2c 100644 --- a/tests/test_core/nodes/test_function_node_iteration.py +++ b/tests/test_core/nodes/test_function_node_iteration.py @@ -169,7 +169,7 @@ def on_data_crash(self, node_label, tag, data, exc): def test_as_table_empty_schema_matches_non_empty_schema(): - """as_table() empty table has the same set of columns as the populated table.""" + """as_table() empty table has the same columns and nullability as the populated table.""" db = InMemoryArrowDatabase() node_after = _make_node(db=db) node_after.run() @@ -181,5 +181,10 @@ def test_as_table_empty_schema_matches_non_empty_schema(): assert empty_table.num_rows == 0 assert full_table.num_rows > 0 assert set(empty_table.column_names) == set(full_table.column_names) + # Nullability must match — this fails before the fix + assert all( + empty_table.schema.field(n).nullable == full_table.schema.field(n).nullable + for n in empty_table.column_names + ) From d698893ef1137f70fa0ed1f5a61e0bbba5f7248a Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:17:30 +0000 Subject: [PATCH 07/15] refactor(operator_node): use module-level arrow_utils import in _make_empty_table (ITL-563) --- src/orcapod/core/nodes/operator_node.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/orcapod/core/nodes/operator_node.py b/src/orcapod/core/nodes/operator_node.py index 859cb5c2..0f22774b 100644 --- a/src/orcapod/core/nodes/operator_node.py +++ b/src/orcapod/core/nodes/operator_node.py @@ -40,6 +40,7 @@ from orcapod.protocols.database_protocols import ArrowDatabaseProtocol from orcapod.system_constants import constants from orcapod.types import CacheMode, ColumnConfig, ContentHash, Schema +from orcapod.utils import arrow_utils from orcapod.utils.lazy_module import LazyModule logger = logging.getLogger(__name__) @@ -842,10 +843,8 @@ def _make_empty_table(self) -> "pa.Table": Requires ``self._operator is not None`` (pre-existing limitation shared with ``_replay_from_cache``). """ - from orcapod.utils.arrow_utils import make_empty_table - tag_schema, data_schema = self.output_schema() - return make_empty_table( + return arrow_utils.make_empty_table( {**tag_schema, **data_schema}, self.data_context.type_converter, ) From 5eeb2f289e4c2310261298adfe7aefebca96fe94 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:22:47 +0000 Subject: [PATCH 08/15] fix(sync_orchestrator): use make_empty_table to preserve field nullability (ITL-563) Replace the empty-branch of _materialize_as_stream() to call arrow_utils.make_empty_table() instead of building a pa.table() from a dict of pa.array() calls, which was marking all fields nullable=True. Also fix Join.static_process() to use an explicit int8 type when adding the _common cartesian-product column so that zero-row tables don't produce a null-typed column that Polars rejects on join. --- src/orcapod/core/operators/join.py | 7 +- src/orcapod/pipeline/sync_orchestrator.py | 10 +-- .../test_error_policy_continue.py | 83 +++++++++++++++++++ 3 files changed, 91 insertions(+), 9 deletions(-) create mode 100644 tests/test_pipeline/test_error_policy_continue.py diff --git a/src/orcapod/core/operators/join.py b/src/orcapod/core/operators/join.py index a6d628e4..35eccd2b 100644 --- a/src/orcapod/core/operators/join.py +++ b/src/orcapod/core/operators/join.py @@ -132,8 +132,9 @@ def static_process(self, *streams: StreamProtocol) -> StreamProtocol: table = stream.as_table( columns={"source": True, "system_tags": True, "meta": True} ) - # trick to get cartesian product - table = table.add_column(0, COMMON_JOIN_KEY, pa.array([0] * len(table))) + # trick to get cartesian product; use int8 explicitly so that zero-row + # tables don't produce a null-typed column that Polars rejects on join + table = table.add_column(0, COMMON_JOIN_KEY, pa.array([0] * len(table), type=pa.int8())) table = arrow_utils.append_to_system_tags( table, f"{stream.pipeline_hash().to_hex(n_char)}:0", @@ -151,7 +152,7 @@ def static_process(self, *streams: StreamProtocol) -> StreamProtocol: # trick to ensure that there will always be at least one shared key # this ensure that no overlap in keys lead to full caretesian product next_table = next_table.add_column( - 0, COMMON_JOIN_KEY, pa.array([0] * len(next_table)) + 0, COMMON_JOIN_KEY, pa.array([0] * len(next_table), type=pa.int8()) ) # Rename any non-key columns in next_table that would collide with diff --git a/src/orcapod/pipeline/sync_orchestrator.py b/src/orcapod/pipeline/sync_orchestrator.py index 0e23f938..8c671e7b 100644 --- a/src/orcapod/pipeline/sync_orchestrator.py +++ b/src/orcapod/pipeline/sync_orchestrator.py @@ -179,16 +179,14 @@ def _materialize_as_stream(buf: list[tuple[Any, Any]], upstream_node: Any) -> An pa = LazyModule("pyarrow") if not buf: - # Build an empty stream with the correct schema from the upstream node + # Build an empty stream preserving declared field nullability. tag_schema, data_schema = upstream_node.output_schema( columns={"system_tags": True, "source": True} ) type_converter = upstream_node.data_context.type_converter - empty_fields = {} - for name, py_type in {**tag_schema, **data_schema}.items(): - arrow_type = type_converter.python_type_to_arrow_type(py_type) - empty_fields[name] = pa.array([], type=arrow_type) - empty_table = pa.table(empty_fields) + empty_table = arrow_utils.make_empty_table( + {**tag_schema, **data_schema}, type_converter + ) tag_keys = upstream_node.keys()[0] return ArrowTableStream( empty_table, diff --git a/tests/test_pipeline/test_error_policy_continue.py b/tests/test_pipeline/test_error_policy_continue.py new file mode 100644 index 00000000..5254fa81 --- /dev/null +++ b/tests/test_pipeline/test_error_policy_continue.py @@ -0,0 +1,83 @@ +"""Integration tests for error_policy='continue' + Join schema compatibility. + +Regression for ITL-563: failed functions produced empty tables with wrong +nullable flags, causing Join to raise InputValidationError and abort the pipeline. +""" +from __future__ import annotations + +import pyarrow as pa +import pytest + +from orcapod.core.data_function import PythonDataFunction +from orcapod.core.function_pod import FunctionPod +from orcapod.core.operators.join import Join +from orcapod.core.sources import ArrowTableSource +from orcapod.databases import InMemoryArrowDatabase +from orcapod.pipeline.job import PipelineJob +from orcapod.pipeline.sync_orchestrator import SyncPipelineOrchestrator + + +def _make_source_with_required_field(subjects: list[str], values: list[int]) -> ArrowTableSource: + """Source with tag=subject (str, non-nullable) and data=value (int, non-nullable).""" + schema = pa.schema([ + pa.field("subject", pa.large_string(), nullable=False), + pa.field("value", pa.int64(), nullable=False), + ]) + table = pa.table( + {"subject": subjects, "value": values}, + schema=schema, + ) + return ArrowTableSource(table, tag_columns=["subject"]) + + +def test_failed_function_with_join_does_not_abort_pipeline(): + """Topology: source → failing_function → Join ← source. + + Under error_policy='continue', the failing function should be logged, + the Join should produce zero rows (empty × non-empty = empty), and + orchestration should complete without raising. + """ + src = _make_source_with_required_field(["a", "b"], [1, 2]) + + def always_fails(value: int) -> int: + raise RuntimeError("intentional failure") + + pf = PythonDataFunction(always_fails, output_keys="transformed") + failing_pod = FunctionPod(pf) + + job = PipelineJob(name="test_join_continue", store=InMemoryArrowDatabase()) + with job: + failing_out = failing_pod(src, label="failing") + Join()(failing_out, src, label="joined") + + # Should complete without raising despite the failing function. + # error_policy is passed via the SyncPipelineOrchestrator constructor. + job.run(orchestrator=SyncPipelineOrchestrator(error_policy="continue")) + + joined_records = job.nodes["joined"].get_all_records() + # Join of empty × non-empty = empty (zero rows), not an error + assert joined_records is None or joined_records.num_rows == 0 + + +def test_empty_buffer_schema_preserves_nullability(): + """_materialize_as_stream on an empty buffer preserves required field nullability.""" + src = _make_source_with_required_field(["x"], [10]) + + def identity(value: int) -> int: + return value + + pf = PythonDataFunction(identity, output_keys="result") + pod = FunctionPod(pf) + + db = InMemoryArrowDatabase() + job = PipelineJob(name="test_empty_schema", store=db) + with job: + pod(src, label="fn") + + # Don't run — buffer will be empty + node = job.nodes["fn"] + stream = SyncPipelineOrchestrator._materialize_as_stream([], node) + tag_schema, data_schema = stream.output_schema() + + # "result" is declared as int (required), must NOT become int | None + assert data_schema["result"] == int From dbf57e548d962d8249849617ec4a8b624ad0d942 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:29:32 +0000 Subject: [PATCH 09/15] refactor(sync_orchestrator): module-level arrow_utils import; remove unused pytest import (ITL-563) --- src/orcapod/pipeline/sync_orchestrator.py | 2 +- tests/test_pipeline/test_error_policy_continue.py | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/src/orcapod/pipeline/sync_orchestrator.py b/src/orcapod/pipeline/sync_orchestrator.py index 8c671e7b..eef5b082 100644 --- a/src/orcapod/pipeline/sync_orchestrator.py +++ b/src/orcapod/pipeline/sync_orchestrator.py @@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any, Literal from orcapod.pipeline.result import OrchestratorResult +from orcapod.utils import arrow_utils from orcapod.protocols.node_protocols import ( is_function_node, is_operator_node, @@ -173,7 +174,6 @@ def _materialize_as_stream(buf: list[tuple[Any, Any]], upstream_node: Any) -> An An ArrowTableStream. """ from orcapod.core.streams.arrow_table_stream import ArrowTableStream - from orcapod.utils import arrow_utils from orcapod.utils.lazy_module import LazyModule pa = LazyModule("pyarrow") diff --git a/tests/test_pipeline/test_error_policy_continue.py b/tests/test_pipeline/test_error_policy_continue.py index 5254fa81..7fc37e96 100644 --- a/tests/test_pipeline/test_error_policy_continue.py +++ b/tests/test_pipeline/test_error_policy_continue.py @@ -6,7 +6,6 @@ from __future__ import annotations import pyarrow as pa -import pytest from orcapod.core.data_function import PythonDataFunction from orcapod.core.function_pod import FunctionPod From bd031f3433694b622bb4aad240052964df4820f9 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:32:29 +0000 Subject: [PATCH 10/15] fix(side_effects): use make_empty_table to preserve field nullability (ITL-563) Replace the hand-rolled empty-branch logic in SideEffectPodStream.as_table() and SideEffectNode.as_table() with arrow_utils.make_empty_table(), which delegates to python_schema_to_arrow_schema and preserves nullable=False for plain types. Also move arrow_utils import to module level per project convention. Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/side_effects.py | 24 +++++++------------ .../test_side_effect_function_pod.py | 24 +++++++++++++++++++ 2 files changed, 33 insertions(+), 15 deletions(-) diff --git a/src/orcapod/side_effects.py b/src/orcapod/side_effects.py index 8b272cda..265312fe 100644 --- a/src/orcapod/side_effects.py +++ b/src/orcapod/side_effects.py @@ -23,6 +23,7 @@ InvocationHashConfig, ) from orcapod.system_constants import TRACKING_DB_SCHEMA_VERSION +from orcapod.utils import arrow_utils from orcapod.utils.lazy_module import LazyModule if TYPE_CHECKING: @@ -154,7 +155,6 @@ def as_table( all_info: bool = False, ) -> pa.Table: from orcapod.types import ColumnConfig as _ColumnConfig - from orcapod.utils import arrow_utils column_config = _ColumnConfig.handle_config(columns, all_info=all_info) tag_tables = [] @@ -163,17 +163,14 @@ def as_table( tag_tables.append(tag.as_table(columns=column_config)) data_tables.append(data.as_table(columns=column_config)) if not tag_tables: - # Return an empty table with the correct schema + # Return an empty table preserving declared field nullability. tag_schema, data_schema = self.output_schema( columns=column_config ) tc = self._pod.data_context.type_converter - fields = {} - for name, py_type in {**tag_schema, **data_schema}.items(): - fields[name] = pa.array( - [], type=tc.python_type_to_arrow_type(py_type) - ) - return pa.table(fields) + return arrow_utils.make_empty_table( + {**tag_schema, **data_schema}, tc + ) combined_tags = pa.concat_tables(tag_tables) combined_data = pa.concat_tables(data_tables) return arrow_utils.hstack_tables(combined_tags, combined_data) @@ -717,7 +714,6 @@ def as_table( ) -> pa.Table: """Collect all rows from ``iter_data()`` into an Arrow table.""" from orcapod.types import ColumnConfig as _ColumnConfig - from orcapod.utils import arrow_utils column_config = _ColumnConfig.handle_config(columns, all_info=all_info) tag_tables = [] @@ -726,14 +722,12 @@ def as_table( tag_tables.append(tag.as_table(columns=column_config)) data_tables.append(data.as_table(columns=column_config)) if not tag_tables: + # Return an empty table preserving declared field nullability. tag_schema, data_schema = self.output_schema(columns=column_config) tc = self._pod.data_context.type_converter - fields = {} - for name, py_type in {**tag_schema, **data_schema}.items(): - fields[name] = pa.array( - [], type=tc.python_type_to_arrow_type(py_type) - ) - return pa.table(fields) + return arrow_utils.make_empty_table( + {**tag_schema, **data_schema}, tc + ) return arrow_utils.hstack_tables( pa.concat_tables(tag_tables), pa.concat_tables(data_tables), diff --git a/tests/test_core/side_effect_function/test_side_effect_function_pod.py b/tests/test_core/side_effect_function/test_side_effect_function_pod.py index c8d9b14c..e16eaab9 100644 --- a/tests/test_core/side_effect_function/test_side_effect_function_pod.py +++ b/tests/test_core/side_effect_function/test_side_effect_function_pod.py @@ -399,6 +399,30 @@ def my_fn(value: int, ctx: InvocationContext) -> str: assert pod.__name__ == "my_fn" +class TestSideEffectEmptyTableNullability: + """Regression for ITL-563: empty as_table() must preserve field nullability.""" + + def test_empty_stream_schema_preserves_required_fields(self): + """SideEffectPodStream.as_table() with no rows preserves nullable=False.""" + from orcapod.side_effects import SideEffectPod, SideEffectPodStream + + calls = [] + + def my_fn(value: int, ctx: InvocationContext) -> None: + calls.append(value) + + pod = SideEffectPod(my_fn, ctx_arg_name="ctx") + stream = _make_stream(n=0) # empty input → empty output + + se_stream = SideEffectPodStream(side_effect_pod=pod, input_stream=stream) + table = se_stream.as_table() + + assert table.num_rows == 0 + # "id" and "value" are declared non-nullable — must not become nullable + assert table.schema.field("id").nullable is False + assert table.schema.field("value").nullable is False + + class TestSideEffectCtxCollision: """Tests for ctx_arg_name collision with data columns in SideEffectPod.""" From d64c6cdbf8be86aca50b84530de27fc67a01c0be Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 05:04:38 +0000 Subject: [PATCH 11/15] test(side_effects): add SideEffectNode nullability regression test (ITL-563) --- .../test_side_effect_function_pod.py | 21 ++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/tests/test_core/side_effect_function/test_side_effect_function_pod.py b/tests/test_core/side_effect_function/test_side_effect_function_pod.py index e16eaab9..7c6e4a06 100644 --- a/tests/test_core/side_effect_function/test_side_effect_function_pod.py +++ b/tests/test_core/side_effect_function/test_side_effect_function_pod.py @@ -406,10 +406,8 @@ def test_empty_stream_schema_preserves_required_fields(self): """SideEffectPodStream.as_table() with no rows preserves nullable=False.""" from orcapod.side_effects import SideEffectPod, SideEffectPodStream - calls = [] - def my_fn(value: int, ctx: InvocationContext) -> None: - calls.append(value) + return str(value) pod = SideEffectPod(my_fn, ctx_arg_name="ctx") stream = _make_stream(n=0) # empty input → empty output @@ -422,6 +420,23 @@ def my_fn(value: int, ctx: InvocationContext) -> None: assert table.schema.field("id").nullable is False assert table.schema.field("value").nullable is False + def test_empty_node_schema_preserves_required_fields(self): + """SideEffectNode.as_table() with no rows preserves nullable=False.""" + from orcapod.side_effects import SideEffectPod, SideEffectNode + + def my_fn(value: int, ctx: InvocationContext) -> None: + return str(value) + + pod = SideEffectPod(my_fn, ctx_arg_name="ctx") + stream = _make_stream(n=0) # empty input → empty output + + node = SideEffectNode(side_effect_pod=pod, input_stream=stream) + table = node.as_table() + + assert table.num_rows == 0 + # "id" is declared non-nullable (int) — must not become nullable + assert table.schema.field("id").nullable is False + class TestSideEffectCtxCollision: """Tests for ctx_arg_name collision with data columns in SideEffectPod.""" From 870a810f2472317b5986a578e1e649045b0c1c37 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 05:06:48 +0000 Subject: [PATCH 12/15] fix(derived_source): use make_empty_table to preserve field nullability (ITL-563) Replace hand-rolled pa.field() construction in the empty-cache branch of DerivedSource._get_stream() with arrow_utils.make_empty_table(), which delegates to python_schema_to_arrow_schema and correctly sets nullable=False for required (non-Optional) fields. Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/sources/derived_source.py | 19 +++++-------------- .../test_core/sources/test_derived_source.py | 18 ++++++++++++++++++ 2 files changed, 23 insertions(+), 14 deletions(-) diff --git a/src/orcapod/core/sources/derived_source.py b/src/orcapod/core/sources/derived_source.py index 0259e1f3..e4aa4202 100644 --- a/src/orcapod/core/sources/derived_source.py +++ b/src/orcapod/core/sources/derived_source.py @@ -5,6 +5,7 @@ from orcapod.core.sources.base import RootSource from orcapod.core.streams.arrow_table_stream import ArrowTableStream from orcapod.types import ColumnConfig, Schema +from orcapod.utils import arrow_utils from orcapod.utils.lazy_module import LazyModule if TYPE_CHECKING: @@ -76,23 +77,13 @@ def _get_stream(self) -> ArrowTableStream: if self._cached_table is None: records = self._origin.get_all_records() if records is None: - # Build empty table with correct schema + # Build empty table preserving declared field nullability. tag_schema, data_schema = self._origin.output_schema() tag_keys = self._origin.keys()[0] tc = self.data_context.type_converter - fields = [ - pa.field(k, tc.python_type_to_arrow_type(tag_schema[k])) - for k in tag_keys - ] - fields += [ - pa.field(k, tc.python_type_to_arrow_type(v)) - for k, v in data_schema.items() - ] - arrow_schema = pa.schema(fields) - self._cached_table = pa.table( - {f.name: pa.array([], type=f.type) for f in arrow_schema}, - schema=arrow_schema, - ) + python_schema = {k: tag_schema[k] for k in tag_keys} + python_schema.update(data_schema) + self._cached_table = arrow_utils.make_empty_table(python_schema, tc) else: self._cached_table = records tag_keys = self._origin.keys()[0] diff --git a/tests/test_core/sources/test_derived_source.py b/tests/test_core/sources/test_derived_source.py index 02cdf6b2..81e50e4b 100644 --- a/tests/test_core/sources/test_derived_source.py +++ b/tests/test_core/sources/test_derived_source.py @@ -441,3 +441,21 @@ def test_explicit_source_id_overrides_default(self): source_id="custom_id", ) assert src.source_id == "custom_id" + + +def test_derived_source_empty_cache_preserves_nullability(): + """DerivedSource._get_stream() empty table preserves nullable=False for required fields. + + Regression for ITL-563: pa.field() without nullable=False defaulted to True. + """ + # _make_node() builds a FunctionJobNode with data output "result: int" (required). + node = _make_node() + source = node.as_source() # DerivedSource backed by unrun node → empty cache + + # as_table() triggers _get_stream() which hits the empty-cache branch + table = source.as_table() + + assert table.num_rows == 0 + # "result" is int (required), must not become int | None + result_field = table.schema.field("result") + assert result_field.nullable is False From a453c782e51aefbfc1e427977e6109b3834b24d4 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 05:16:04 +0000 Subject: [PATCH 13/15] =?UTF-8?q?docs(design-issues):=20mark=20CC1=20resol?= =?UTF-8?q?ved=20=E2=80=94=20empty=20table=20nullability=20fix=20(ITL-563)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- DESIGN_ISSUES.md | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/DESIGN_ISSUES.md b/DESIGN_ISSUES.md index 33544633..63b090ba 100644 --- a/DESIGN_ISSUES.md +++ b/DESIGN_ISSUES.md @@ -16,7 +16,7 @@ Each item has a status: `open`, `in progress`, or `resolved`. ## Cross-cutting ### CC1 — Empty tables lose field nullability, aborting `error_policy="continue"` pipelines -**Status:** in progress +**Status:** resolved **Severity:** high **Issue:** ITL-563 @@ -35,6 +35,14 @@ Sites: `sync_orchestrator.py:_materialize_as_stream`, `operator_node.py:_make_em `python_schema_to_arrow_schema` + `pa.Table.from_batches([], schema=...)` and replace all five sites. +**Fix:** Extracted ``arrow_utils.make_empty_table(python_schema, type_converter)`` using +``python_schema_to_arrow_schema`` + ``pa.Table.from_batches([], schema=...)``. Replaced all +five buggy sites: ``sync_orchestrator._materialize_as_stream``, +``operator_node._make_empty_table``, ``SideEffectPodStream.as_table``, +``SideEffectNode.as_table``, and ``DerivedSource._get_stream``. Also fixed a latent +null-typed array bug in ``Join.static_process``. Added unit tests, integration test for the +exact bug topology, and per-site regression tests. PR: ITL-563. + --- ## `src/orcapod/core/nodes/function_node.py` From 2907b1b67cef19dcc8d1cdbaf3687032e808ce46 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 05:16:31 +0000 Subject: [PATCH 14/15] docs(design-issues): consolidate duplicate Fix note in CC1 (ITL-563) --- DESIGN_ISSUES.md | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/DESIGN_ISSUES.md b/DESIGN_ISSUES.md index 63b090ba..4aec2530 100644 --- a/DESIGN_ISSUES.md +++ b/DESIGN_ISSUES.md @@ -31,17 +31,12 @@ Sites: `sync_orchestrator.py:_materialize_as_stream`, `operator_node.py:_make_em `side_effects.py:SideEffectJobFunctionStream.as_table`, `derived_source.py:DerivedSource._get_stream`. -**Fix:** Extract `arrow_utils.make_empty_table(python_schema, type_converter)` using -`python_schema_to_arrow_schema` + `pa.Table.from_batches([], schema=...)` and replace all -five sites. - -**Fix:** Extracted ``arrow_utils.make_empty_table(python_schema, type_converter)`` using -``python_schema_to_arrow_schema`` + ``pa.Table.from_batches([], schema=...)``. Replaced all -five buggy sites: ``sync_orchestrator._materialize_as_stream``, -``operator_node._make_empty_table``, ``SideEffectPodStream.as_table``, -``SideEffectNode.as_table``, and ``DerivedSource._get_stream``. Also fixed a latent -null-typed array bug in ``Join.static_process``. Added unit tests, integration test for the -exact bug topology, and per-site regression tests. PR: ITL-563. +**Fix:** Extracted `arrow_utils.make_empty_table(python_schema, type_converter)` using +`python_schema_to_arrow_schema` + `pa.Table.from_batches([], schema=...)`. Replaced all five +buggy sites: `sync_orchestrator._materialize_as_stream`, `operator_node._make_empty_table`, +`SideEffectPodStream.as_table`, `SideEffectNode.as_table`, and `DerivedSource._get_stream`. +Also fixed a latent null-typed array bug in `Join.static_process`. Added unit tests, +integration test for the exact bug topology, and per-site regression tests. PR: ITL-563. --- From 6b16ee79afeddd2e565791d7325fe6e0bb07b887 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 06:22:04 +0000 Subject: [PATCH 15/15] fix(tests): correct side-effect test function return types and DESIGN_ISSUES class names (ITL-563) - Remove `return str(value)` from two `-> None` side-effect test functions in TestSideEffectEmptyTableNullability; side-effect functions must return None - Update CC1 entry in DESIGN_ISSUES.md to use the actual class names (SideEffectPodStream/SideEffectNode) instead of the old plan names Co-Authored-By: Claude Sonnet 4.6 --- DESIGN_ISSUES.md | 4 ++-- .../side_effect_function/test_side_effect_function_pod.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/DESIGN_ISSUES.md b/DESIGN_ISSUES.md index 4aec2530..d03cb2ee 100644 --- a/DESIGN_ISSUES.md +++ b/DESIGN_ISSUES.md @@ -27,8 +27,8 @@ When a failing function's empty output is fed into a `Join`, the mismatched sche `InputValidationError` that bypasses `error_policy="continue"` and aborts orchestration. Sites: `sync_orchestrator.py:_materialize_as_stream`, `operator_node.py:_make_empty_table`, -`side_effects.py:SideEffectFunctionStream.as_table`, -`side_effects.py:SideEffectJobFunctionStream.as_table`, +`side_effects.py:SideEffectPodStream.as_table`, +`side_effects.py:SideEffectNode.as_table`, `derived_source.py:DerivedSource._get_stream`. **Fix:** Extracted `arrow_utils.make_empty_table(python_schema, type_converter)` using diff --git a/tests/test_core/side_effect_function/test_side_effect_function_pod.py b/tests/test_core/side_effect_function/test_side_effect_function_pod.py index 7c6e4a06..aa11c6ba 100644 --- a/tests/test_core/side_effect_function/test_side_effect_function_pod.py +++ b/tests/test_core/side_effect_function/test_side_effect_function_pod.py @@ -407,7 +407,7 @@ def test_empty_stream_schema_preserves_required_fields(self): from orcapod.side_effects import SideEffectPod, SideEffectPodStream def my_fn(value: int, ctx: InvocationContext) -> None: - return str(value) + pass pod = SideEffectPod(my_fn, ctx_arg_name="ctx") stream = _make_stream(n=0) # empty input → empty output @@ -425,7 +425,7 @@ def test_empty_node_schema_preserves_required_fields(self): from orcapod.side_effects import SideEffectPod, SideEffectNode def my_fn(value: int, ctx: InvocationContext) -> None: - return str(value) + pass pod = SideEffectPod(my_fn, ctx_arg_name="ctx") stream = _make_stream(n=0) # empty input → empty output