diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e453be..faad16d 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). + - 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). diff --git a/docs/backends.md b/docs/backends.md index 0db98e8..348ccdd 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 88ec5ed..42b9341 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 19f0ea0..8adbd56 100644 --- a/src/freshdata/execution/_report.py +++ b/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 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 0000000..b7ccdd5 --- /dev/null +++ b/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"]