fix(engines): disclose categorical pandas fallback

CHANGELOG.md

@@ -8,6 +8,12 @@ adheres to [Semantic Versioning](https://semver.org/).
 
 ### Fixed
 
+- Polars and DuckDB now disclose a pandas fallback for categorical and Arrow
+  dictionary-backed pandas inputs. This preserves pandas text-cleaning and
+  category metadata behavior rather than silently skipping or changing it.
+  The fallback preserves category domains and ordering and honours
+  `fallback_policy="warn"` / `"error"` before ingestion.
+
 - Native Polars and DuckDB runs now audit pandas extension dtype changes in
   pandas output instead of silently returning a different dtype. Each changed
   column records an `output_dtype` backend difference; native handles stay

docs/fallback-matrix.md

@@ -41,7 +41,9 @@ FreshCore also runs its own config and data checks in
 | 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 | pandas | — | pandas | DuckDB registration and subsequent alias repairs can redirect stages to the wrong column. 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 datetime / timedelta columns | native | native | — | pandas | FreshCore v1 carries only float, bool and string columns, so these dtypes would come back as strings |
+| pandas input with categorical columns | pandas | pandas | — | pandas | native ingestion can change the category dtype and text-cleaning behavior; use the pandas reference to preserve its cleaning behavior and category metadata |
+| pandas input with Arrow dictionary columns | pandas | pandas | — | native (dtype lossy) | Polars/DuckDB disclose a pandas fallback. FreshCore v1 currently casts Arrow dictionaries to strings without a dtype fallback, losing the dictionary domain and ordering; explicitly select pandas to preserve them |
 | 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 |
 | pandas input with a nanosecond `timedelta64[ns]` or a timezone-aware datetime column | native | pandas | — | pandas | DuckDB stores `INTERVAL` in microseconds (`5ns` → `0`) and returns `TIMESTAMP WITH TIME ZONE` in the session time zone, at microsecond resolution |
 | pandas input with an integer column holding a value beyond ±2\*\*53 | native | native | — | pandas | FreshCore v1 carries numbers as float64, which cannot represent such integers exactly; other integer columns are cast back to their input dtype (or reported in `backend_differences` when the result is no longer integral) |

src/freshdata/execution/_ingest.py

@@ -26,6 +26,9 @@ def pandas_ingest_fallback_reason(source: Any, engine: str | None = None) -> str
     to one backend's type system only apply to that backend; omit it to run the
     checks every native engine shares.
     """
+    if engine == "pandas":
+        return None
+
     import pandas as pd
     from pandas.api.types import infer_dtype, is_object_dtype
 
@@ -62,6 +65,13 @@ def pandas_ingest_fallback_reason(source: Any, engine: str | None = None) -> str
 def _lossy_dtype_reason(dtype: Any, label: Any, engine: str | None) -> str | None:
     """Why *engine* cannot carry a column of *dtype* back unchanged."""
     name = str(dtype)
+    if name == "category" or name.startswith("dictionary"):
+        # Native text stages skip categorical/ENUM columns. Casting them to
+        # strings would also discard the category domain and its ordering.
+        return (
+            f"categorical column {label!r} requires the pandas reference path "
+            "to preserve pandas text-cleaning, domain, and ordering behavior"
+        )
     if name.startswith(("period[", "interval[")):
         # DuckDB rejects both dtypes outright ("Data type not recognized");
         # Polars ingests a period as its raw int64 ordinal (2020-01 -> 600) and

tests/test_execution/test_categorical_ingest.py

@@ -0,0 +1,182 @@
+"""Categorical pandas inputs retain text repairs and category metadata (#456)."""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+import freshdata as fd
+from freshdata.execution import FallbackError, FallbackWarning
+from freshdata.execution._ingest import pandas_ingest_fallback_reason
+from freshdata.execution.backends._duckdb import DuckDBEngine
+from freshdata.execution.backends._polars import PolarsEngine
+
+pytest.importorskip("polars")
+pytest.importorskip("duckdb")
+
+ENGINES = ("polars", "duckdb")
+KW = {"strategy": "conservative", "fix_dtypes": False, "verbose": False}
+
+
+def _frame() -> pd.DataFrame:
+    return pd.DataFrame({
+        "c": pd.Categorical([" beta", "alpha ", "N/A"], ordered=True),
+        "id": [1, 2, 3],
+    })
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_categorical_text_repairs_match_pandas_and_are_disclosed(engine):
+    frame = _frame()
+    original = frame.copy(deep=True)
+    expected, reference = fd.clean(frame, engine="pandas", return_report=True, **KW)
+    out, report = fd.clean(frame, engine=engine, return_report=True, **KW)
+
+    pd.testing.assert_frame_equal(pd.DataFrame(out), pd.DataFrame(expected))
+    pd.testing.assert_frame_equal(frame, original)
+    assert out["c"].iloc[:2].tolist() == ["beta", "alpha"]
+    assert pd.isna(out["c"].iloc[2])
+    assert [(a.step, a.count) for a in report.actions] == [
+        (a.step, a.count) for a in reference.actions
+    ] == [("strip_whitespace", 2), ("normalize_sentinels", 1)]
+    assert report.backend == "pandas"
+    assert report.requested_backend == engine
+    (event,) = report.fallback_events
+    assert event["fallback_step"] == "pipeline"
+    assert "categorical column 'c'" in event["fallback_reason"]
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+@pytest.mark.parametrize("values", [
+    pytest.param(pd.Categorical(["a ", "a", " N/A", None], ordered=True), id="merged-categories"),
+    pytest.param(pd.Categorical([1, " a ", "N/A", None]), id="mixed-categories"),
+    pytest.param(
+        pd.Categorical([2, 1, None], categories=[2, 1, 3], ordered=True), id="numeric-order"
+    ),
+    pytest.param(
+        pd.Categorical([None, None], categories=[" unused "], ordered=True), id="all-missing"
+    ),
+    pytest.param(pd.Categorical([], categories=[" unused "], ordered=True), id="empty"),
+])
+def test_category_domain_order_and_values_match_pandas(engine, values):
+    frame = pd.DataFrame({"c": values, "id": range(len(values))})
+    expected = fd.clean(frame, engine="pandas", drop_empty_columns=False, **KW)
+    out, report = fd.clean(
+        frame, engine=engine, drop_empty_columns=False, return_report=True, **KW
+    )
+    pd.testing.assert_frame_equal(pd.DataFrame(out), pd.DataFrame(expected))
+    assert isinstance(out["c"].dtype, pd.CategoricalDtype)
+    assert report.fallback_events
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_disabled_text_steps_preserve_the_category_domain(engine):
+    frame = _frame()
+    out, report = fd.clean(
+        frame, engine=engine, strip_whitespace=False, normalize_sentinels=False,
+        return_report=True, **KW,
+    )
+    pd.testing.assert_frame_equal(pd.DataFrame(out), frame)
+    assert report.actions == []
+    assert report.fallback_events
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_categorical_fallback_honours_warn_policy(engine):
+    with pytest.warns(FallbackWarning, match="categorical column"):
+        out = fd.clean(_frame(), engine=engine, fallback_policy="warn", **KW)
+    assert out["c"].iloc[0] == "beta"
+
+
+@pytest.mark.parametrize(("engine", "backend", "ingest"), [
+    ("polars", PolarsEngine, "_to_lazy"),
+    ("duckdb", DuckDBEngine, "_register_source"),
+])
+def test_error_policy_blocks_ingestion_and_pandas_work(monkeypatch, engine, backend, ingest):
+    def forbidden(*args, **kwargs):
+        raise AssertionError("strict fallback policy must reject before cleaning")
+
+    monkeypatch.setattr(backend, ingest, forbidden)
+    monkeypatch.setattr(backend, "_fallback", forbidden)
+    with pytest.raises(FallbackError, match="categorical column 'c'"):
+        fd.clean(_frame(), engine=engine, fallback_policy="error", **KW)
+
+
+@pytest.mark.parametrize(("engine", "output_format"), [
+    ("polars", "polars-lazy"), ("duckdb", "duckdb"),
+])
+def test_native_handle_requests_disclose_materialized_pandas_fallback(engine, output_format):
+    out, report = fd.clean(
+        _frame(), engine=engine, output_format=output_format, return_report=True, **KW
+    )
+    assert isinstance(out, pd.DataFrame)
+    assert report.materialized is True
+    assert report.backend == "pandas"
+    assert report.fallback_events
+
+
+def _arrow_frame(ordered=False) -> pd.DataFrame:
+    pa = pytest.importorskip("pyarrow")
+    array = pa.array([" beta", "alpha ", "N/A", None]).dictionary_encode().cast(
+        pa.dictionary(pa.int8(), pa.string(), ordered=ordered)
+    )
+    return pd.DataFrame({"c": pd.Series(pd.arrays.ArrowExtensionArray(array)), "id": [1, 2, 3, 4]})
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+@pytest.mark.parametrize("ordered", [False, True])
+def test_arrow_dictionary_inputs_preserve_pandas_behavior_and_dtype(engine, ordered):
+    frame = _arrow_frame(ordered)
+    original = frame.copy(deep=True)
+    expected, reference = fd.clean(frame, engine="pandas", return_report=True, **KW)
+
+    out, report = fd.clean(frame, engine=engine, return_report=True, **KW)
+
+    pd.testing.assert_frame_equal(pd.DataFrame(out), pd.DataFrame(expected))
+    pd.testing.assert_frame_equal(frame, original)
+    assert out["c"].dtype == frame["c"].dtype
+    assert [(a.step, a.count) for a in report.actions] == [
+        (a.step, a.count) for a in reference.actions
+    ]
+    assert report.backend == "pandas"
+    assert report.requested_backend == engine
+    assert "categorical column 'c'" in report.fallback_events[0]["fallback_reason"]
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_arrow_dictionary_fallback_honours_warn_policy(engine):
+    with pytest.warns(FallbackWarning, match="categorical column"):
+        fd.clean(_arrow_frame(), engine=engine, fallback_policy="warn", **KW)
+
+
+@pytest.mark.parametrize(("engine", "backend", "ingest"), [
+    ("polars", PolarsEngine, "_to_lazy"),
+    ("duckdb", DuckDBEngine, "_register_source"),
+])
+def test_arrow_dictionary_error_policy_blocks_ingestion_and_pandas_work(
+    monkeypatch, engine, backend, ingest
+):
+    def forbidden(*args, **kwargs):
+        raise AssertionError("strict fallback policy must reject before cleaning")
+
+    monkeypatch.setattr(backend, ingest, forbidden)
+    monkeypatch.setattr(backend, "_fallback", forbidden)
+    with pytest.raises(FallbackError, match="categorical column 'c'"):
+        fd.clean(_arrow_frame(), engine=engine, fallback_policy="error", **KW)
+
+
+@pytest.mark.parametrize("frame", [
+    pytest.param(_frame(), id="categorical"),
+    pytest.param(pd.DataFrame([[1, 2]], columns=["c", "c"]), id="duplicate-labels"),
+    pytest.param(pd.DataFrame({"c": [1, "a"]}), id="mixed-values"),
+    pytest.param(
+        pd.DataFrame({"c": pd.period_range("2026-01", periods=2, freq="M")}), id="period"
+    ),
+])
+def test_explicit_pandas_engine_does_not_need_an_ingest_fallback(frame):
+    assert pandas_ingest_fallback_reason(frame, engine="pandas") is None
+
+
+@pytest.mark.parametrize("ordered", [False, True])
+def test_explicit_pandas_engine_accepts_arrow_dictionary_without_fallback(ordered):
+    assert pandas_ingest_fallback_reason(_arrow_frame(ordered), engine="pandas") is None