fix: preserve field nullability in empty Arrow tables (ITL-563)

DESIGN_ISSUES.md

@@ -13,6 +13,33 @@ Each item has a status: `open`, `in progress`, or `resolved`.
 
 ---
 
+## Cross-cutting
+
+### CC1 — Empty tables lose field nullability, aborting `error_policy="continue"` pipelines
+**Status:** resolved
+**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: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
+`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`
 
 ### FN1 — `FunctionNodeBase.as_table()` returned empty schema when no data existed

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__)
@@ -847,18 +848,18 @@ 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``).
         """
         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 arrow_utils.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.

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

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]

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,22 +174,19 @@ 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")
 
         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,

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),

src/orcapod/utils/arrow_utils.py

@@ -10,10 +10,30 @@
 
 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: "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``, …)
+    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],