From a90017a17d1b0d8da3bda3d057cac478c7d77a2e Mon Sep 17 00:00:00 2001 From: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> Date: Tue, 13 Oct 2026 08:36:33 -0500 Subject: [PATCH 1/3] fix(execution): audit native pandas extension dtype changes Signed-off-by: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> --- CHANGELOG.md | 5 + docs/backends.md | 9 + src/freshdata/execution/__init__.py | 2 + src/freshdata/execution/_report.py | 58 +++++++ .../test_extension_dtype_audit.py | 163 ++++++++++++++++++ 5 files changed, 237 insertions(+) create mode 100644 tests/test_execution/test_extension_dtype_audit.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 15f1f5b2..2696dc18 100644 --- a/CHANGELOG.md +++ b/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). + - 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 diff --git a/docs/backends.md b/docs/backends.md index 0db98e83..348ccddf 100644 --- a/docs/backends.md +++ b/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; diff --git a/src/freshdata/execution/__init__.py b/src/freshdata/execution/__init__.py index 88ec5edf..42b93415 100644 --- a/src/freshdata/execution/__init__.py +++ b/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 diff --git a/src/freshdata/execution/_report.py b/src/freshdata/execution/_report.py index 19f0ea0d..c2be54b4 100644 --- a/src/freshdata/execution/_report.py +++ b/src/freshdata/execution/_report.py @@ -16,9 +16,67 @@ 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 + for position, native_name in enumerate(native_names): + original_dtype = source_dtypes.iloc[position] + if not is_extension_array_dtype(original_dtype): + continue + output_label = rename_map.get(native_name, native_name) + if output_label not in result.columns: + continue # dropped columns are structural changes, not dtype changes + output_dtype = result[output_label].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 diff --git a/tests/test_execution/test_extension_dtype_audit.py b/tests/test_execution/test_extension_dtype_audit.py new file mode 100644 index 00000000..a3812482 --- /dev/null +++ b/tests/test_execution/test_extension_dtype_audit.py @@ -0,0 +1,163 @@ +"""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 + +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_registration_names_are_used_for_case_insensitive_headers(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) + assert len(out.columns) == (1 if drop_second else 2) + assert {event["column"] for event in _dtype_changes(report)} == set(out.columns) From cb8b2c8159d957424ee5d2f46a03cacafdf8638b Mon Sep 17 00:00:00 2001 From: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> Date: Tue, 13 Oct 2026 09:25:07 -0500 Subject: [PATCH 2/3] fix(audit): track dtype changes by final output position Signed-off-by: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> --- src/freshdata/execution/_report.py | 17 +++++++--- .../test_extension_dtype_audit.py | 32 +++++++++++++++++++ 2 files changed, 44 insertions(+), 5 deletions(-) diff --git a/src/freshdata/execution/_report.py b/src/freshdata/execution/_report.py index c2be54b4..8adbd563 100644 --- a/src/freshdata/execution/_report.py +++ b/src/freshdata/execution/_report.py @@ -49,14 +49,21 @@ def record_extension_dtype_changes( native_names = [str(label) for label in original_names] plan_names = native_names rename_map = PlanGenerator(config, backend=backend).plan(plan_names).rename_map - for position, native_name in enumerate(native_names): + # 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 = rename_map.get(native_name, native_name) - if output_label not in result.columns: - continue # dropped columns are structural changes, not dtype changes - output_dtype = result[output_label].dtype + 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. diff --git a/tests/test_execution/test_extension_dtype_audit.py b/tests/test_execution/test_extension_dtype_audit.py index a3812482..41a87952 100644 --- a/tests/test_execution/test_extension_dtype_audit.py +++ b/tests/test_execution/test_extension_dtype_audit.py @@ -9,6 +9,7 @@ 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") @@ -161,3 +162,34 @@ def test_duckdb_registration_names_are_used_for_case_insensitive_headers(drop_se out, report = fd.clean(frame, engine="duckdb", return_report=True, **KW) 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) + 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 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"] From 2f696ce700b20463735b4b96b4a77c2aa62fe12e Mon Sep 17 00:00:00 2001 From: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> Date: Tue, 13 Oct 2026 09:49:59 -0500 Subject: [PATCH 3/3] test(audit): accept disclosed DuckDB preservation fallback Signed-off-by: WilliamK112 <164879897+WilliamK112@users.noreply.github.com> --- .../test_extension_dtype_audit.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/tests/test_execution/test_extension_dtype_audit.py b/tests/test_execution/test_extension_dtype_audit.py index 41a87952..b7ccdd5f 100644 --- a/tests/test_execution/test_extension_dtype_audit.py +++ b/tests/test_execution/test_extension_dtype_audit.py @@ -154,12 +154,15 @@ def test_dtype_details_do_not_include_category_domain_values(engine): @pytest.mark.parametrize("drop_second", [False, True]) -def test_duckdb_registration_names_are_used_for_case_insensitive_headers(drop_second): +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) @@ -173,12 +176,26 @@ def test_duckdb_final_projection_suffix_uses_the_actual_output_column(drop_unrel 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"],