diff --git a/CHANGELOG.md b/CHANGELOG.md index faad16d..ca33bfc 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/fallback-matrix.md b/docs/fallback-matrix.md index 10a656f..855bb56 100644 --- a/docs/fallback-matrix.md +++ b/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) | diff --git a/src/freshdata/execution/_ingest.py b/src/freshdata/execution/_ingest.py index 1fb4442..631f6b9 100644 --- a/src/freshdata/execution/_ingest.py +++ b/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 diff --git a/tests/test_execution/test_categorical_ingest.py b/tests/test_execution/test_categorical_ingest.py new file mode 100644 index 0000000..8bd6135 --- /dev/null +++ b/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