fix(polling_source): eliminate per-batch schema inference, fix zero-row crash (ENG-952) (#261)

* docs(specs): add ENG-952 design spec for PollingSource zero-row batch fix

* docs(specs): update ENG-952 spec — canonical-schema approach replaces Fix B

* docs(plans): add ENG-952 implementation plan

* test(polling_source): add failing regression tests for ENG-952 zero-row batch crash

* refactor(test_polling_source): remove extraneous to_config/from_config from inline test impls

Follow the established inline-class pattern used throughout the file: test
impl classes define only identity, schema, poll, fetch, and close.  Also add
a descriptive failure message to the _accumulated_stream assertion so failures
self-describe rather than raising an AttributeError on the next line.

* fix(polling_source): establish canonical Arrow schema once, eliminating zero-row crash

* fix(polling_source): defer infer-once schema establishment until first non-empty batch

* docs(design_issues): add PS4 entry for ENG-952 zero-row batch schema crash

* docs(polling_source): improve _build_stream_from_df docstring and remove stale type: ignore

* fix(test_polling_source): update test_zero_row_batch_is_not_accumulated for _batches API

ITL-617 replaced _accumulated_stream with an append-only _batches list.
Zero-row batches produce empty ArrowTableStream entries in _batches, so the
assertion must count total rows across all batches rather than batch-list length.

Also rename DESIGN_ISSUES PS4 entry to PS5 to avoid collision with the
ITL-617 PS4 entry that was added to main.

---------

Co-authored-by: agent-kurodo[bot] <268466204+agent-kurodo[bot]@users.noreply.github.com>

Author: kurodo3[bot]

DESIGN_ISSUES.md

@@ -172,6 +172,24 @@ the async loop loses the commit race. The lock is never held across ``await``.
 
 ---
 
+### PS5 — `PollingSource` re-infers Arrow schema nullability per batch, crashing on zero-row polls
+**Status:** resolved
+**Severity:** high
+**Issue:** ENG-952
+
+`_build_stream_from_df` called `infer_schema_nullable` on every batch. A zero-row batch has
+`null_count == 0` for all columns, so every field was inferred non-nullable.
+`_validate_combining_schemas` then rejected the batch against the accumulated stream's
+nullable schema.
+
+**Fix:** `_build_stream_from_df` now establishes a `_canonical_arrow_schema` exactly once —
+from `impl.schema()` when declared (no inference, no warning), or from the first non-empty
+batch otherwise (with a `WARNING`-level log). All subsequent batches are cast to the canonical
+schema by column name. Per-batch nullability inference is eliminated. Zero-row frames before
+canonical schema establishment are skipped on the infer-once path.
+
+---
+
 ## `src/orcapod/core/nodes/function_node.py`
 
 ### FN1 — `FunctionNodeBase.as_table()` returned empty schema when no data existed

src/orcapod/core/sources/polling_source.py

@@ -258,6 +258,7 @@ def __init__(
         self._cursor: Cursor[T] | None = None
         self._batches: list[ArrowTableStream] = []
         self._state_lock: threading.Lock = threading.Lock()
+        self._canonical_arrow_schema: pa.Schema | None = None
         # Derive source_id from impl identity if not explicitly provided
         if self._source_id is None:
             self._source_id = str(self._impl.identity())
@@ -613,8 +614,13 @@ def _try_build_stream(self, data: FrameInitTypes) -> ArrowTableStream | None:
             return None
         return self._build_stream_from_df(df)
 
-    def _build_stream_from_df(self, df: pl.DataFrame) -> ArrowTableStream:
-        """Build an ``ArrowTableStream`` from a Polars DataFrame."""
+    def _build_stream_from_df(self, df: pl.DataFrame) -> ArrowTableStream | None:
+        """Build an ``ArrowTableStream`` from a Polars DataFrame.
+
+        Returns ``None`` on the infer-once path when the batch has zero rows and
+        no canonical schema has been established yet — the frame is silently
+        skipped so that a spurious all-non-nullable schema is never recorded.
+        """
         from orcapod.core.streams.arrow_table_stream import ArrowTableStream
 
         # Handle Object-dtype columns (same pattern as DataFrameSource)
@@ -628,7 +634,42 @@ def _build_stream_from_df(self, df: pl.DataFrame) -> ArrowTableStream:
         df = polars_data_utils.drop_system_columns(df)
 
         arrow_table = df.to_arrow()
-        arrow_table = arrow_table.cast(arrow_utils.infer_schema_nullable(arrow_table))
+
+        # Establish canonical schema on first call; apply it on every call.
+        if self._canonical_arrow_schema is None:
+            if self._tag_schema is not None and self._data_schema is not None:
+                # Declared-schema path: derive Arrow schema from declared Python types.
+                # T | None → nullable=True; plain T → nullable=False. No inference.
+                combined = {**dict(self._tag_schema), **dict(self._data_schema)}
+                self._canonical_arrow_schema = (
+                    self.data_context.type_converter.python_schema_to_arrow_schema(combined)
+                )
+            else:
+                # Infer-once path: first non-empty batch establishes canonical nullability.
+                # Skip zero-row tables: null_count is always 0 for empty tables, so
+                # inference would set every field nullable=False — the original ENG-952 bug.
+                if arrow_table.num_rows > 0:
+                    logger.warning(
+                        "PollingSource %r: no schema declared via impl.schema(); "
+                        "inferring nullability from first batch. Implement impl.schema() "
+                        "to avoid schema drift on zero-row polls or null-free batches.",
+                        self._source_id,
+                    )
+                    self._canonical_arrow_schema = arrow_utils.infer_schema_nullable(arrow_table)
+
+        # If _canonical_arrow_schema is still None here, this is a zero-row frame
+        # on the infer-once path before any real data has arrived.  Skip it —
+        # the caller (_try_build_stream) will return None and the frame is ignored.
+        if self._canonical_arrow_schema is None:
+            return None
+
+        # Apply canonical nullability by column name (order-safe).
+        canonical_nullable = {f.name: f.nullable for f in self._canonical_arrow_schema}
+        target_schema = pa.schema([
+            pa.field(f.name, f.type, nullable=canonical_nullable.get(f.name, f.nullable))
+            for f in arrow_table.schema
+        ])
+        arrow_table = arrow_table.cast(target_schema)
 
         builder = SourceStreamBuilder(self.data_context, self.orcapod_config)
         result = builder.build(