fix(polars): align zero-column handling across dependency versions

Signed-off-by: WilliamK112 <164879897+WilliamK112@users.noreply.github.com>

Author: WilliamK112

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

@@ -95,7 +95,8 @@ def test_zero_column_pandas_input_discloses_a_native_handle_fallback(native_conf
 
 
 def test_empty_zero_column_pandas_input_stays_native(native_config):
-    out, report = fd.clean(pd.DataFrame(), config=native_config, engine="polars",
+    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