fix(polars): preserve rows in zero-column pandas inputs

CHANGELOG.md

@@ -8,6 +8,11 @@ adheres to [Semantic Versioning](https://semver.org/).
 
 ### Fixed
 
+- The Polars engine no longer silently loses rows from a zero-column pandas
+  input. It uses the recorded pandas fallback, preserving the input index and
+  row counts with pandas output; `fallback_policy="error"` rejects the run
+  before ingestion instead. Empty inputs with no rows stay native (#463).
+
 - A repeated index label no longer breaks cleaning. `fd.clean(df)` and
   `fd.suggest_plan(df)` raised on ordinary input whenever the frame's index
   carried the same label twice — the shape you get from concatenating two

docs/fallback-matrix.md

@@ -39,6 +39,7 @@ FreshCore also runs its own config and data checks in
 | semantic cleaning | native-distinct | native-distinct | pandas | pandas | polars/duckdb run it over a natively extracted distinct table; non-default semantic backends force pandas |
 | pandas input with a mixed-type object column (e.g. numbers and strings) | pandas | pandas | — | pandas | native ingestion would reject the column (polars) or cast every value to text (duckdb) |
 | pandas input with duplicate column labels | pandas | pandas | — | pandas | native frames need unique column names; the pandas pipeline deduplicates them (`"x", "x"` → `"x", "x_2"`) |
+| pandas input with zero columns and one or more rows | pandas | pandas | — | — | the pandas reference preserves the rows and index; Polars 1.36 loses these rows during ingestion, and DuckDB cannot register a zero-column frame. Use pandas output to retain the row count |
 | pandas input whose column labels collide once stringified (e.g. `1` and `"1"`) | native | native | — | pandas | FreshCore names columns by `str(label)`, so one column would overwrite the other; distinct non-string labels (e.g. `0`, `1`) stay native and come back unchanged |
 | pandas input with datetime / timedelta / categorical columns | native | native | — | pandas | FreshCore v1 carries only float, bool and string columns, so these dtypes would come back as strings |
 | pandas input with period or interval columns | pandas | pandas | — | pandas | DuckDB rejects both dtypes outright; Polars ingests a period as its raw int64 ordinal (`2020-01` → `600`) and an interval as a `{left, right}` struct |

src/freshdata/execution/_ingest.py

@@ -30,6 +30,10 @@ def pandas_ingest_fallback_reason(source: Any, engine: str | None = None) -> str
 
     if not isinstance(source, pd.DataFrame):
         return None
+    if engine == "polars" and len(source) > 0 and len(source.columns) == 0:
+        # Polars 1.36 loses these rows during ingestion. Even where ingestion
+        # preserves them, per-column metadata cannot recover their input count.
+        return "zero-column pandas source requires the pandas reference to preserve its rows"
     if source.columns.duplicated().any():
         return "duplicate input column labels require the pandas reference path"
     for i, dtype in enumerate(source.dtypes):

src/freshdata/execution/backends/_polars.py

@@ -472,6 +472,10 @@ def _stage_drop_duplicates(
         pl: Any,
         engine_config: EngineConfig | None = None,
     ) -> Any:
+        # With no columns, the pandas reference skips duplicate handling. Newer
+        # Polars versions retain height here but unique() can collapse the rows.
+        if not lf.collect_schema().names():
+            return lf
         # Counts go through the streaming engine: a plain .collect() re-evaluates
         # the whole upstream plan in memory, defeating out-of-core execution.
         n_before = self._count(lf, engine_config, pl)

tests/test_execution/test_engine_parity_edge_cases.py

@@ -108,9 +108,11 @@ def _all_empty() -> pd.DataFrame:
 
 
 @pytest.mark.parametrize("engine", ["polars", "duckdb"])
-def test_all_empty_columns_match_pandas(engine):
-    ref_out, ref = fd.clean(_all_empty(), config=_config(), engine="pandas", return_report=True)
-    out, report = fd.clean(_all_empty(), config=_config(), engine=engine, return_report=True)
+@pytest.mark.parametrize("drop_duplicates", [False, True])
+def test_all_empty_columns_match_pandas(engine, drop_duplicates):
+    config = _config(drop_duplicates=drop_duplicates)
+    ref_out, ref = fd.clean(_all_empty(), config=config, engine="pandas", return_report=True)
+    out, report = fd.clean(_all_empty(), config=config, engine=engine, return_report=True)
     assert list(out.columns) == list(ref_out.columns) == []
     assert len(out) == len(ref_out) == 2
     assert (report.rows_after, report.cols_after) == (ref.rows_after, ref.cols_after) == (2, 0)

tests/test_execution/test_polars_engine.py

@@ -54,6 +54,56 @@ def test_drop_empty_column(native_config):
     assert "empty" not in out.columns
 
 
+def test_zero_column_pandas_input_keeps_rows_and_report_counts(native_config):
+    df = pd.DataFrame(index=range(3))
+    out, report = fd.clean(df, config=native_config, engine="polars", return_report=True)
+    assert out.shape == (3, 0)
+    assert report.rows_before == report.rows_after == report.rows_materialized == 3
+    assert report.requested_backend == "polars"
+    assert report.backend == "pandas"
+    (event,) = report.fallback_events
+    assert "zero-column" in event["fallback_reason"]
+
+
+def test_zero_column_pandas_input_preserves_its_index(native_config):
+    df = pd.DataFrame(index=pd.RangeIndex(7, 10, name="record"))
+    out = fd.clean(df, config=native_config, engine="polars")
+    reference = fd.clean(df, config=native_config, engine="pandas")
+    pd.testing.assert_frame_equal(out, reference, check_frame_type=False)
+
+
+def test_zero_column_pandas_input_honours_error_policy(native_config):
+    with pytest.raises(FallbackError, match="zero-column"):
+        fd.clean(pd.DataFrame(index=range(3)), config=native_config, engine="polars",
+                 fallback_policy="error")
+
+
+def test_zero_column_pandas_input_warns_when_requested(native_config):
+    with pytest.warns(fd.FallbackWarning, match="zero-column"):
+        out = fd.clean(pd.DataFrame(index=range(3)), config=native_config, engine="polars",
+                       fallback_policy="warn")
+    assert out.shape == (3, 0)
+
+
+def test_zero_column_pandas_input_discloses_a_native_handle_fallback(native_config):
+    out, report = fd.clean(pd.DataFrame(index=range(3)), config=native_config, engine="polars",
+                           output_format="polars-lazy", return_report=True)
+    assert isinstance(out, pd.DataFrame)
+    assert out.shape == (3, 0)
+    assert report.materialized
+    assert report.fallback_events
+
+
+def test_empty_zero_column_pandas_input_stays_native(native_config):
+    out, report = fd.clean(pd.DataFrame(index=pd.RangeIndex(0)),
+                           config=native_config, engine="polars",
+                           return_report=True)
+    assert out.shape == (0, 0)
+    assert report.rows_before == report.rows_after == 0
+    assert report.backend == "polars"
+    assert report.fallback_events == []
+
+
 def test_fallback_on_knn_like_config_warns(small_df, caplog):
     """A config needing the decision engine falls back to pandas with a warning."""
     import logging