fix(execution): audit native pandas extension dtype changes

CHANGELOG.md

@@ -8,6 +8,11 @@ adheres to [Semantic Versioning](https://semver.org/).
 
 ### Fixed
 
+- 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
+  unmaterialized, and values are not recast to force the input dtype (#465).
+
 - DuckDB now discloses a pandas fallback for input headers that collide under
   its ASCII-insensitive identifier rules. This prevents projection aliases
   from selecting the wrong column or dropping a nonempty one (#527).

docs/backends.md

@@ -171,6 +171,15 @@ Notes:
   each engine uses its own quantile interpolation / tie-breaking. Polars and
   DuckDB use linear-interpolated quantiles matching pandas; Spark uses
   `approxQuantile`. Such divergences are recorded in `report.backend_differences`.
+- **Pandas extension dtypes** can change when native Polars/DuckDB results are
+  converted back to pandas: for example, nullable `Int64` without missing values
+  can become NumPy `int64`, or pandas `string` can become `object`. These changes
+  are recorded per output column in `report.backend_differences` with
+  `step="output_dtype"`, the backend, and the before/after dtype names. The audit
+  follows renamed columns and ignores dropped columns. It does not recast values
+  or undo requested type-changing operations; choose `engine="pandas"` when the
+  reference dtype behavior is required. Native handles and non-pandas outputs
+  are not materialized for this pandas-output check.
 - **Outlier counts** match the pandas reference where the quantile statistics
   match (Polars/DuckDB linear interpolation); Spark may flag a different count.
   pandas, Polars, DuckDB and Spark compute the fences from finite values only;

src/freshdata/execution/__init__.py

@@ -29,6 +29,7 @@
 )
 from ._metadata import ColumnMetadata, MetadataScanner
 from ._plan import NativePlan, PlanGenerator
+from ._report import record_extension_dtype_changes
 
 if TYPE_CHECKING:  # pragma: no cover - typing only
     from ..config import CleanConfig
@@ -217,6 +218,7 @@ def run_with_engine(
 
         cleaned_native = run_semantic_native(cleaned_native, config, report, engine=resolved)
     result = _convert_output(cleaned_native, engine_config.output_format, report)
+    record_extension_dtype_changes(source, result, config, report)
     _finish_report(report, requested, resolved, result)
     return (result, report) if return_report else result
 

src/freshdata/execution/_report.py

@@ -16,9 +16,74 @@
 from ..report import CleanReport
 
 if TYPE_CHECKING:  # pragma: no cover - typing only
+    from ..config import CleanConfig
     from ._metadata import ColumnMetadata
 
 
+def record_extension_dtype_changes(
+    source: Any, result: Any, config: CleanConfig, report: CleanReport
+) -> None:
+    """Disclose native pandas-output changes without recasting data or handles."""
+    backend = report.backend
+    if backend is None or backend not in ("polars", "duckdb"):
+        return
+    import pandas as pd
+    from pandas.api.types import is_dtype_equal, is_extension_array_dtype
+
+    if not isinstance(source, pd.DataFrame) or not isinstance(result, pd.DataFrame):
+        return
+    from ._plan import PlanGenerator
+
+    source_dtypes = source.dtypes
+    if not any(is_extension_array_dtype(dtype) for dtype in source_dtypes):
+        return
+    original_names = list(source.columns)
+    if backend == "duckdb":
+        import duckdb
+
+        # Registration deduplicates case-insensitive names before the plan's
+        # renames. Inspect relation metadata without fetching any rows.
+        native_names = duckdb.from_df(source).columns
+        plan_names = original_names
+    else:
+        native_names = [str(label) for label in original_names]
+        plan_names = native_names
+    rename_map = PlanGenerator(config, backend=backend).plan(plan_names).rename_map
+    # Native stages retain source order and append derived columns. Match each
+    # retained source position to the actual final label: a SQL projection can
+    # disambiguate registration/normalization collisions a second time.
+    retained = [
+        position for position, name in enumerate(native_names)
+        if rename_map.get(name, name) not in report.columns_dropped
+    ]
+    for output_position, position in enumerate(retained):
+        if output_position >= len(result.columns):
+            break
+        original_dtype = source_dtypes.iloc[position]
+        if not is_extension_array_dtype(original_dtype):
+            continue
+        output_label = str(result.columns[output_position])
+        output_dtype = result.iloc[:, output_position].dtype
+        if not is_dtype_equal(original_dtype, output_dtype):
+            # Categorical repr includes actual category values. Report dtype
+            # names and storage, not a domain that can contain sensitive data.
+            before = str(original_dtype)
+            after = str(output_dtype)
+            if isinstance(original_dtype, pd.StringDtype):
+                before = f"string[{original_dtype.storage}]"
+            if isinstance(output_dtype, pd.StringDtype):
+                after = f"string[{output_dtype.storage}]"
+            if isinstance(original_dtype, pd.SparseDtype):
+                before = f"Sparse[{original_dtype.subtype}]"
+            report.record_backend_difference(
+                backend,
+                "output_dtype",
+                f"pandas output changed the input extension dtype/metadata "
+                f"from {before} to {after}",
+                column=output_label,
+            )
+
+
 def init_report(metadata: list[ColumnMetadata], memory_before: int = 0) -> CleanReport:
     """Initialise a report's "before" counts from cheap column metadata."""
     rows_before = metadata[0].row_count if metadata else 0

tests/test_execution/test_extension_dtype_audit.py

@@ -0,0 +1,212 @@
+"""Native pandas output discloses extension dtype changes without recasting."""
+
+from __future__ import annotations
+
+import json
+
+import pandas as pd
+import pytest
+from pandas.api.types import is_dtype_equal
+
+import freshdata as fd
+from freshdata.execution._report import record_extension_dtype_changes
+
+pytest.importorskip("polars")
+pytest.importorskip("duckdb")
+pytest.importorskip("pyarrow")
+
+ENGINES = ("polars", "duckdb")
+KW = {"strategy": "conservative", "fix_dtypes": False, "verbose": False}
+
+
+def _dtype_changes(report):
+    return [event for event in report.backend_differences if event["step"] == "output_dtype"]
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+@pytest.mark.parametrize(("dtype", "values"), [
+    pytest.param("Int8", [1, 2, 3], id="nullable-small-int"),
+    pytest.param("UInt64", [2**63 + 1, 2**63 + 2, 2**63 + 3], id="wide-unsigned"),
+    pytest.param("Float64", [1.0, None, 3.0], id="nullable-float64"),
+    pytest.param("Float32", [1.0, None, 3.0], id="nullable-float32"),
+    pytest.param("string", ["a", None, "b"], id="nullable-text"),
+    pytest.param("string[pyarrow]", ["a", None, "b"], id="arrow-text"),
+    pytest.param("boolean", [True, None, False], id="nullable-boolean"),
+    pytest.param("Int64", [2**53 + 1, None, 2**53 + 3], id="exact-nullable-wide-int"),
+])
+def test_extension_dtype_change_is_audited_per_column(engine, dtype, values):
+    frame = pd.DataFrame({"c": pd.Series(values, dtype=dtype), "row": [0, 1, 2]})
+    original = frame.copy(deep=True)
+    out, report = fd.clean(frame, engine=engine, return_report=True, **KW)
+
+    assert out["c"].isna().tolist() == frame["c"].isna().tolist()
+    assert out.loc[frame["c"].notna(), "c"].tolist() == frame.loc[
+        frame["c"].notna(), "c"
+    ].tolist()
+    pd.testing.assert_frame_equal(frame, original)
+    changes = _dtype_changes(report)
+    if is_dtype_equal(frame["c"].dtype, out["c"].dtype):
+        assert changes == []
+    else:
+        assert len(changes) == 1
+        assert changes[0]["backend"] == engine
+        assert changes[0]["column"] == "c"
+        assert str(frame["c"].dtype) in changes[0]["detail"]
+        assert str(out["c"].dtype) in changes[0]["detail"]
+        assert json.loads(report.to_json())["backend_differences"] == report.backend_differences
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_renamed_and_disambiguated_columns_are_audited_under_output_names(engine):
+    frame = pd.DataFrame({
+        "Customer Value": pd.Series([1, 2, 3], dtype="Int64"),
+        "Customer-Value": pd.Series([4, 5, 6], dtype="Int64"),
+    })
+    out, report = fd.clean(frame, engine=engine, return_report=True, **KW)
+    assert out.columns.tolist() == ["customer_value", "customer_value_2"]
+    assert {event["column"] for event in _dtype_changes(report)} == set(out.columns)
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_non_string_column_label_maps_to_the_native_output_label(engine):
+    frame = pd.DataFrame({7: pd.Series([1, 2, 3], dtype="Int64")})
+    out, report = fd.clean(frame, engine=engine, column_names=False, return_report=True, **KW)
+    assert out.columns.tolist() == ["7"]
+    assert [event["column"] for event in _dtype_changes(report)] == ["7"]
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_dropped_extension_column_is_not_a_dtype_change(engine):
+    frame = pd.DataFrame({"empty": pd.Series([None, None], dtype="Int64"), "row": [0, 1]})
+    out, report = fd.clean(frame, engine=engine, return_report=True, **KW)
+    assert out.columns.tolist() == ["row"]
+    assert _dtype_changes(report) == []
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_regular_numpy_dtypes_do_not_create_extension_dtype_events(engine):
+    frame = pd.DataFrame({"n": [1, 2, 3], "s": ["a", "b", "c"]})
+    _, report = fd.clean(frame, engine=engine, return_report=True, **KW)
+    assert _dtype_changes(report) == []
+
+
+def test_pandas_fallback_is_not_reported_as_a_native_dtype_change():
+    frame = pd.DataFrame({"c": pd.Series([1, 2, 3], dtype="Int64")})
+    _, report = fd.clean(frame, engine="polars", strategy="balanced", return_report=True,
+                         fix_dtypes=False, verbose=False)
+    assert report.backend == "pandas"
+    assert report.fallback_events
+    assert _dtype_changes(report) == []
+
+
+@pytest.mark.parametrize(("engine", "output_format"), [
+    ("polars", "polars-lazy"), ("duckdb", "duckdb"),
+])
+def test_native_handles_are_not_materialized_for_pandas_dtype_auditing(
+    engine, output_format, monkeypatch
+):
+    def forbidden(*args, **kwargs):
+        raise AssertionError("dtype auditing must not convert a native handle to pandas")
+
+    monkeypatch.setattr("freshdata.adapters.polars.to_pandas", forbidden)
+    frame = pd.DataFrame({"c": pd.Series([1, 2, 3], dtype="Int64")})
+    _, report = fd.clean(frame, engine=engine, output_format=output_format,
+                         return_report=True, **KW)
+    assert report.materialized is False
+    assert report.rows_materialized is None
+    assert _dtype_changes(report) == []
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_arrow_output_does_not_claim_a_pandas_dtype_change(engine):
+    frame = pd.DataFrame({"c": pd.Series([1, 2, 3], dtype="Int64")})
+    _, report = fd.clean(frame, engine=engine, output_format="arrow", return_report=True, **KW)
+    assert _dtype_changes(report) == []
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_auditing_does_not_truncate_an_explicit_fractional_imputation(engine):
+    frame = pd.DataFrame({"c": pd.Series([1, 2, None], dtype="Int64"), "row": [0, 1, 2]})
+    out, report = fd.clean(frame, engine=engine, impute="mean", return_report=True, **KW)
+    assert out["c"].tolist() == [1.0, 2.0, 1.5]
+    assert len(_dtype_changes(report)) == 1
+    assert any(event["step"] == "impute" for event in report.backend_differences)
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_empty_frame_schema_changes_are_still_audited(engine):
+    frame = pd.DataFrame({"c": pd.Series([], dtype="Int64")})
+    out, report = fd.clean(frame, engine=engine, return_report=True, **KW)
+    assert out.shape == (0, 1)
+    changes = _dtype_changes(report)
+    assert len(changes) == (not is_dtype_equal(out["c"].dtype, frame["c"].dtype))
+
+
+@pytest.mark.parametrize("engine", ENGINES)
+def test_dtype_details_do_not_include_category_domain_values(engine):
+    frame = pd.DataFrame({
+        "c": pd.Categorical(["a", "b", "a"],
+                            categories=["a", "b", "unused-private-category"], ordered=True),
+        "row": [0, 1, 2],
+    })
+    _, report = fd.clean(frame, engine=engine, return_report=True, **KW)
+    assert "unused-private-category" not in report.to_json()
+
+
+@pytest.mark.parametrize("drop_second", [False, True])
+def test_duckdb_case_insensitive_headers_are_audited_or_preserved(drop_second):
+    frame = pd.DataFrame({
+        "Customer Value": pd.Series([1, 2, 3], dtype="Int64"),
+        "customer value": pd.Series([None] * 3 if drop_second else [4, 5, 6], dtype="Int64"),
+    })
+    out, report = fd.clean(frame, engine="duckdb", return_report=True, **KW)
+    if report.backend == "pandas":
+        _assert_collision_fallback_preserves_extension_dtypes(frame, out, report)
+        return
+    assert len(out.columns) == (1 if drop_second else 2)
+    assert {event["column"] for event in _dtype_changes(report)} == set(out.columns)
+
+
+@pytest.mark.parametrize("drop_unrelated", [False, True])
+def test_duckdb_final_projection_suffix_uses_the_actual_output_column(drop_unrelated):
+    frame = pd.DataFrame({
+        "A": [1, 2, 3], "a": ["x", "y", "z"],
+        "A 1": pd.Series([4, 5, 6], dtype="Int64"),
+    })
+    if drop_unrelated:
+        frame["Unused"] = None
+    out, report = fd.clean(frame, engine="duckdb", return_report=True, **KW)
+    if report.backend == "pandas":
+        _assert_collision_fallback_preserves_extension_dtypes(frame, out, report)
+        return
+    assert out.columns.tolist() == ["a", "a_1", "a_1_1"]
+    (event,) = _dtype_changes(report)
+    assert event["column"] == "a_1_1"
+    assert str(out.iloc[:, 2].dtype) in event["detail"]
+
+
+def _assert_collision_fallback_preserves_extension_dtypes(frame, out, report):
+    # The separate data-preservation fix (#528) may route these headers to
+    # pandas; an unchanged extension dtype must then have no native audit entry.
+    assert report.requested_backend == "duckdb"
+    assert any("case-insensitive" in event["fallback_reason"]
+               for event in report.fallback_events)
+    expected = fd.clean(frame, engine="pandas", **KW)
+    pd.testing.assert_frame_equal(pd.DataFrame(out), pd.DataFrame(expected))
+    assert _dtype_changes(report) == []
+
+
+def test_duplicate_output_labels_are_compared_by_position_without_raising():
+    source = pd.DataFrame({
+        "A": [1, 2, 3], "a": ["x", "y", "z"],
+        "A 1": pd.Series([4, 5, 6], dtype="Int64"),
+    })
+    result = pd.DataFrame([[1, "x", 4], [2, "y", 5], [3, "z", 6]],
+                          columns=["a", "a_1", "a_1"])
+    report = fd.CleanReport(backend="duckdb")
+    record_extension_dtype_changes(
+        source, result, fd.CleanConfig(strategy="conservative", fix_dtypes=False), report
+    )
+    (event,) = _dtype_changes(report)
+    assert event["column"] == "a_1"
+    assert "Int64" in event["detail"] and "int64" in event["detail"]