diff --git a/docs/backends.md b/docs/backends.md index c62c0bee..a6898e47 100644 --- a/docs/backends.md +++ b/docs/backends.md @@ -163,7 +163,11 @@ Notes: since native frames carry no index. - FreshCore also falls back, based on the data, when `outliers` is set and a float column holds `±inf`, or when `impute="mode"`/`"auto"` would fill a - nullable `boolean` column. With `drop_duplicates=False` and + nullable `boolean` column. It falls back for datetime, timedelta, + categorical, period and interval columns, for integer columns holding values + beyond ±2\*\*53, and for column labels that collide once stringified (such as + `1` and `"1"`). Other integer columns are cast back to their input dtype, and + non-string labels come back unchanged. With `drop_duplicates=False` and `duplicate_ratio_action="error"`, it falls back unless the native module reports a duplicate-row count (see the [fallback matrix](fallback-matrix.md)). - FreshCore v1 is a cleaning-first native engine, not an out-of-core engine. It diff --git a/docs/fallback-matrix.md b/docs/fallback-matrix.md index 24566094..f60256f2 100644 --- a/docs/fallback-matrix.md +++ b/docs/fallback-matrix.md @@ -37,6 +37,9 @@ FreshCore also runs its own config and data checks in | semantic cleaning | native-distinct | native-distinct | pandas | pandas | polars/duckdb run it over a natively extracted distinct table; non-default semantic backends force pandas | | pandas input with a mixed-type object column (e.g. numbers and strings) | pandas | pandas | — | pandas | native ingestion would reject the column (polars) or cast every value to text (duckdb) | | 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 whose column labels collide once stringified (e.g. `1` and `"1"`) | native | native | — | pandas | 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 / period / interval columns | native | native | — | pandas | FreshCore v1 carries only float, bool and string columns, so these dtypes would come back as strings | +| 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) | | contracts / validation / memory / profile replay | pandas | pandas | pandas | pandas | in-memory reference features (see [limitations](limitations.md)) | “pandas” means the **whole pipeline** runs on the pandas reference (fallbacks diff --git a/docs/freshcore.md b/docs/freshcore.md index a1786b4f..14a0869e 100644 --- a/docs/freshcore.md +++ b/docs/freshcore.md @@ -59,6 +59,16 @@ columns with missing values. Detection-only dedup under `duplicate_ratio_action="error"` falls back unless the native module reports `duplicates_detected`. +The native arrays carry only float, bool and string values, so FreshCore also +falls back for datetime, timedelta, categorical, period and interval columns, +for integer columns holding values beyond ±2\*\*53 (float64 cannot represent +them exactly), and for column labels that collide once stringified (such as +`1` and `"1"`). Integer columns (`int64`, nullable `Int64`, and other widths) +are cast back to their input dtype when every returned value is integral and +in range; otherwise they come back as `float64` and the change is recorded in +`report.backend_differences`. Non-string column labels such as `0` and `1` +come back unchanged. + ## Benchmarking Build the native module before benchmarking: diff --git a/src/freshdata/execution/backends/_freshcore.py b/src/freshdata/execution/backends/_freshcore.py index 26f07f1f..67e82637 100644 --- a/src/freshdata/execution/backends/_freshcore.py +++ b/src/freshdata/execution/backends/_freshcore.py @@ -13,12 +13,21 @@ from __future__ import annotations import importlib +import math import time +from collections import Counter from typing import TYPE_CHECKING, Any import numpy as np import pandas as pd -from pandas.api.types import is_bool_dtype, is_float_dtype, is_numeric_dtype, is_object_dtype +from pandas.api.types import ( + is_bool_dtype, + is_datetime64_any_dtype, + is_float_dtype, + is_integer_dtype, + is_numeric_dtype, + is_object_dtype, +) from ..._util import memory_bytes from ...config import _DEFAULT_FACTOR @@ -31,9 +40,14 @@ from ._pandas import materialize_to_pandas if TYPE_CHECKING: # pragma: no cover - typing only + from collections.abc import Sequence + from ...config import CleanConfig from .._config import EngineConfig +#: Largest integer magnitude float64 represents exactly (native numbers are f64). +_MAX_EXACT_FLOAT_INT = 2**53 + class FreshCoreEngine(ExecutionEngine): """Optional native backend backed by the ``freshdata_freshcore`` extension.""" @@ -89,8 +103,10 @@ def execute( "which this FreshCore native module does not report", ) - cleaned = self._frame_from_native(native) + cleaned, dtype_changes = self._frame_from_native(native, frame, config) report = self._report_from_native(frame, cleaned, native, started, config) + for column, detail in dtype_changes: + report.record_backend_difference("freshcore", "dtypes", detail, column=column) return cleaned, report @staticmethod @@ -150,8 +166,29 @@ def _unsupported_reason(self, frame: pd.DataFrame, config: CleanConfig) -> str | return "protected/id/target column semantics require the pandas reference path" if frame.columns.duplicated().any(): return "duplicate input column labels require the pandas reference path" + colliding = self._colliding_labels(frame, config) + if colliding: + return ( + f"column labels {self._shown(colliding)} collide once stringified: " + "FreshCore v1 names columns by str(label), so distinct labels such as " + "1 and '1' require the pandas reference path" + ) if not isinstance(frame.index, pd.RangeIndex): return "non-default pandas index semantics require the pandas reference path" + non_scalar, kinds = self._non_scalar_dtype_columns(frame) + if non_scalar: + return ( + f"{'/'.join(kinds)} column(s) {self._shown(non_scalar)}: FreshCore v1 " + "carries only float, bool and string columns, so datetime, timedelta, " + "categorical, period and interval dtypes require the pandas reference path" + ) + wide = self._wide_integer_columns(frame) + if wide: + return ( + f"integer column(s) {self._shown(wide)} hold values beyond ±2**53: " + "FreshCore v1 carries numbers as float64, which cannot represent them " + "exactly, so they require the pandas reference path" + ) if self._has_unsupported_object_values(frame): return "mixed object columns with non-string values require the pandas reference path" if config.outliers is not None: @@ -173,11 +210,75 @@ def _unsupported_reason(self, frame: pd.DataFrame, config: CleanConfig) -> str | return None @staticmethod - def _shown(columns: list[str], limit: int = 5) -> str: + def _shown(columns: Sequence[object], limit: int = 5) -> str: shown = ", ".join(repr(c) for c in columns[:limit]) extra = len(columns) - limit return f"{shown} (+{extra} more)" if extra > 0 else shown + @staticmethod + def _output_labels(frame: pd.DataFrame, config: CleanConfig) -> list[object]: + """Column labels after the (optional) column-name normalization step.""" + if config.column_names: + return normalized_column_labels(frame.columns) + return list(frame.columns) + + def _colliding_labels(self, frame: pd.DataFrame, config: CleanConfig) -> list[object]: + """Distinct labels whose ``str()`` clashes before or after renaming. + + The native module keys columns by string name, so ``1`` and ``"1"`` + (or ``1`` and ``" 1 "``, which normalizes to ``"1"``) would collapse. + """ + clashing: set[int] = set() + for labels in (list(frame.columns), self._output_labels(frame, config)): + names = [str(label) for label in labels] + counts = Counter(names) + clashing.update(i for i, name in enumerate(names) if counts[name] > 1) + return [frame.columns[i] for i in sorted(clashing)] + + @staticmethod + def _non_scalar_dtype_columns(frame: pd.DataFrame) -> tuple[list[object], list[str]]: + """Columns whose dtype the native float/bool/string arrays cannot carry.""" + found: list[object] = [] + kinds: list[str] = [] + for i, col in enumerate(frame.columns): + dtype = frame.iloc[:, i].dtype + if isinstance(dtype, pd.CategoricalDtype): + kind = "categorical" + elif isinstance(dtype, pd.PeriodDtype): + kind = "period" + elif isinstance(dtype, pd.IntervalDtype): + kind = "interval" + elif is_datetime64_any_dtype(dtype): + kind = "datetime" + elif isinstance(dtype, np.dtype) and dtype.kind == "m": + kind = "timedelta" + else: + continue + found.append(col) + if kind not in kinds: + kinds.append(kind) + return found, kinds + + @staticmethod + def _wide_integer_columns(frame: pd.DataFrame) -> list[object]: + """Integer columns holding a value float64 cannot represent exactly.""" + found: list[object] = [] + for i, col in enumerate(frame.columns): + s = frame.iloc[:, i] + if is_bool_dtype(s) or not is_integer_dtype(s): + continue + info = np.iinfo(getattr(s.dtype, "numpy_dtype", s.dtype)) + if int(info.min) >= -_MAX_EXACT_FLOAT_INT and int(info.max) <= _MAX_EXACT_FLOAT_INT: + continue # int8..int32 and their unsigned widths always fit + non_null = s.dropna() + if non_null.empty: + continue + if int(non_null.max()) > _MAX_EXACT_FLOAT_INT or ( + int(non_null.min()) < -_MAX_EXACT_FLOAT_INT + ): + found.append(col) + return found + @staticmethod def _non_finite_float_columns(frame: pd.DataFrame) -> list[str]: """Columns holding ±inf. Only float dtypes can store infinities.""" @@ -215,10 +316,7 @@ def _has_unsupported_object_values(frame: pd.DataFrame) -> bool: return False def _payload(self, frame: pd.DataFrame, config: CleanConfig) -> dict[str, Any]: - renamed = ( - normalized_column_labels(frame.columns) - if config.column_names else list(frame.columns) - ) + renamed = self._output_labels(frame, config) rename_map = [ (str(old), str(new)) for old, new in zip(frame.columns, renamed) @@ -270,18 +368,74 @@ def _column_payload(name: str, series: pd.Series) -> dict[str, Any]: ] return {"name": name, "dtype": "string", "values": string_values} - @staticmethod - def _frame_from_native(native: dict[str, Any]) -> pd.DataFrame: - data: dict[str, Any] = {} - for column in native["columns"]: + def _frame_from_native( + self, + native: dict[str, Any], + original: pd.DataFrame | None = None, + config: CleanConfig | None = None, + ) -> tuple[pd.DataFrame, list[tuple[str, str]]]: + """Rebuild the cleaned frame and list the dtype changes it could not undo. + + Native column names are ``str(label)``; with *original* and *config* + they are mapped back to the original (possibly renamed) labels, and + integer columns are cast back to their input dtype when possible. + """ + sources: dict[str, tuple[object, pd.Series]] = {} + if original is not None and config is not None: + for i, label in enumerate(self._output_labels(original, config)): + sources[str(label)] = (label, original.iloc[:, i]) + labels: list[object] = [] + data: dict[int, Any] = {} + changes: list[tuple[str, str]] = [] + for position, column in enumerate(native["columns"]): name = column["name"] values = column["values"] dtype = column.get("dtype") + label, source = sources.get(name, (name, None)) + labels.append(label) if dtype == "bool": - data[name] = pd.Series(values, dtype="boolean") + data[position] = pd.Series(values, dtype="boolean") + elif dtype == "float" and source is not None and is_integer_dtype(source): + restored, detail = self._restore_integer(values, source.dtype) + data[position] = restored + if detail is not None: + changes.append((str(label), detail)) else: - data[name] = values - return pd.DataFrame(data) + data[position] = values + cleaned = pd.DataFrame(data) + cleaned.columns = pd.Index(labels) + return cleaned, changes + + @staticmethod + def _restore_integer(values: list[Any], dtype: Any) -> tuple[Any, str | None]: + """Cast native float values back to the input integer *dtype* when exact.""" + nullable = not isinstance(dtype, np.dtype) + info = np.iinfo(getattr(dtype, "numpy_dtype", dtype)) + ints: list[int | None] = [] + problem: str | None = None + for v in values: + if v is None or (isinstance(v, float) and math.isnan(v)): + if not nullable: + problem = "missing values, which a non-nullable integer dtype cannot hold" + break + ints.append(None) + elif not (math.isfinite(v) and float(v).is_integer()): + problem = "non-integral values" + break + elif not int(info.min) <= int(v) <= int(info.max): + problem = f"values outside the {dtype} range" + break + else: + ints.append(int(v)) + if problem is None: + if nullable: + return pd.array(ints, dtype=dtype), None + return np.array(ints, dtype=dtype), None + detail = ( + f"FreshCore v1 returned {problem} for this {dtype} input column, " + "so it comes back as float64" + ) + return pd.Series(values, dtype="float64"), detail def _report_from_native( self, diff --git a/tests/test_execution/test_freshcore_dtype_fidelity.py b/tests/test_execution/test_freshcore_dtype_fidelity.py new file mode 100644 index 00000000..d2c79d80 --- /dev/null +++ b/tests/test_execution/test_freshcore_dtype_fidelity.py @@ -0,0 +1,293 @@ +"""FreshCore dtype and column-label fidelity. + +- #262: datetime/categorical/timedelta columns came back as strings, integers + as float64, and integers beyond 2**53 were silently rounded. +- #232 (part 2): labels such as ``1`` and ``"1"`` collapsed into one native + column, and non-string labels came back stringified. + +The Rust extension is optional, so a fake native module stands in for it. +""" + +from __future__ import annotations + +import re + +import numpy as np +import pandas as pd +import pytest + +import freshdata as fd +from freshdata.execution.backends._freshcore import FreshCoreEngine + +KW = {"strategy": "conservative", "fix_dtypes": False, "verbose": False} + + +class _ForbiddenNative: + @staticmethod + def execute_plan(payload): # pragma: no cover - must not be called + raise AssertionError("unexpected native call") + + +class _EchoNative: + """Applies the rename map, then returns the columns (optionally overridden).""" + + calls = 0 + payloads: list = [] + overrides: dict = {} + + @classmethod + def execute_plan(cls, payload): + cls.calls += 1 + cls.payloads.append(payload) + renames = dict(payload["config"]["rename_map"]) + columns = [] + for column in payload["columns"]: + name = renames.get(column["name"], column["name"]) + values = cls.overrides.get(name, column["values"]) + columns.append({**column, "name": name, "values": values}) + return {"columns": columns, "actions": []} + + +def _use_native(monkeypatch, module) -> None: + monkeypatch.setattr(FreshCoreEngine, "_load_native", staticmethod(lambda: module)) + + +def _echo(monkeypatch, **overrides) -> type[_EchoNative]: + native = type( + "EchoNative", (_EchoNative,), {"calls": 0, "payloads": [], "overrides": overrides} + ) + _use_native(monkeypatch, native) + return native + + +def _dtypes(frame: pd.DataFrame) -> list[str]: + return [str(t) for t in frame.dtypes] + + +FALLBACK_CASES = [ + pytest.param( + pd.DataFrame({"ts": pd.to_datetime(["2024-01-01", None, "2024-01-03"]), "k": [1.0, 2, 3]}), + "datetime column(s) 'ts'", + id="datetime", + ), + pytest.param( + pd.DataFrame( + {"tz": pd.to_datetime(["2024-01-01", "2024-01-02"]).tz_localize("UTC"), "k": [1.0, 2]} + ), + "datetime column(s) 'tz'", + id="datetime-tz", + ), + pytest.param( + pd.DataFrame({"cat": pd.Categorical(["a", "b", "a"]), "k": [1.0, 2, 3]}), + "categorical column(s) 'cat'", + id="categorical", + ), + pytest.param( + pd.DataFrame({"td": pd.to_timedelta([1, 2, 3], unit="s"), "k": [1.0, 2, 3]}), + "timedelta column(s) 'td'", + id="timedelta", + ), + pytest.param( + pd.DataFrame({"p": pd.period_range("2024-01", periods=3, freq="M"), "k": [1.0, 2, 3]}), + "period column(s) 'p'", + id="period", + ), + pytest.param( + pd.DataFrame({"iv": pd.interval_range(0, 3), "k": [1.0, 2, 3]}), + "interval column(s) 'iv'", + id="interval", + ), + pytest.param( + pd.DataFrame( + { + "ts": pd.to_datetime(["2024-01-01", "2024-01-02"]), + "cat": pd.Categorical(["a", "b"]), + "n": [1, 2], + } + ), + "datetime/categorical column(s) 'ts', 'cat'", + id="several-kinds", + ), + pytest.param( + pd.DataFrame({"big": pd.array([2**53 + 1, None, 3], dtype="Int64"), "k": [1.0, 2, 3]}), + "integer column(s) 'big' hold values beyond ±2**53", + id="wide-Int64", + ), + pytest.param( + pd.DataFrame({"big": np.array([-(2**53) - 1, 0, 1], dtype="int64")}), + "integer column(s) 'big' hold values beyond ±2**53", + id="wide-negative-int64", + ), + pytest.param( + pd.DataFrame({"big": np.array([2**64 - 1, 0, 1], dtype="uint64")}), + "integer column(s) 'big' hold values beyond ±2**53", + id="wide-uint64", + ), + pytest.param( + pd.DataFrame({1: [10.0, 20.0, 30.0], "1": ["a", "b", "c"]}), + "column labels 1, '1' collide once stringified", + id="232-colliding-labels", + ), + pytest.param( + pd.DataFrame({1: [10.0, 20.0, 30.0], " 1 ": ["a", "b", "c"]}), + "column labels 1, ' 1 ' collide once stringified", + id="232-labels-colliding-after-rename", + ), +] + + +@pytest.mark.parametrize(("df", "reason"), FALLBACK_CASES) +def test_falls_back_to_pandas_with_identical_output(monkeypatch, df, reason): + _use_native(monkeypatch, _ForbiddenNative) + expected, expected_report = fd.clean(df.copy(), engine="pandas", return_report=True, **KW) + out, report = fd.clean(df.copy(), engine="freshcore", return_report=True, **KW) + + assert report.backend == "pandas" + (event,) = report.fallback_events + assert event["backend"] == "freshcore" + assert event["fallback_step"] == "pipeline" + assert reason in event["fallback_reason"] + # engine="pandas" wraps its frame in CleanResult (a DataFrame subclass). + pd.testing.assert_frame_equal(out, pd.DataFrame(expected)) + assert list(out.columns) == list(expected.columns) + assert _dtypes(out) == _dtypes(expected) + assert [(a.step, a.column, a.count) for a in report.actions] == [ + (a.step, a.column, a.count) for a in expected_report.actions + ] + + +@pytest.mark.parametrize(("df", "reason"), FALLBACK_CASES) +def test_fallback_policy_error_refuses(monkeypatch, df, reason): + _use_native(monkeypatch, _ForbiddenNative) + with pytest.raises(fd.FallbackError, match=re.escape(reason)): + fd.clean(df, engine="freshcore", fallback_policy="error", **KW) + + +def test_issue_262_reproduction_matches_pandas(monkeypatch): + _use_native(monkeypatch, _ForbiddenNative) + df = pd.DataFrame( + { + "ts": pd.to_datetime(["2024-01-01", "2024-01-02"]), + "tz": pd.to_datetime(["2024-01-01", "2024-01-02"]).tz_localize("UTC"), + "cat": pd.Categorical(["a", "b"]), + "td": pd.to_timedelta([1, 2], unit="s"), + "n": [1, 2], + "big": pd.array([2**53 + 1, None], dtype="Int64"), + } + ) + expected = fd.clean(df.copy(), engine="pandas", **KW) + out, report = fd.clean(df.copy(), engine="freshcore", return_report=True, **KW) + + assert report.backend == "pandas" + assert _dtypes(out) == _dtypes(expected) + assert out["big"].iloc[0] == 2**53 + 1 + + +def test_integer_dtypes_round_trip_through_native(monkeypatch): + df = pd.DataFrame( + { + "n": np.array([1, 2, 3], dtype="int64"), + "big": pd.array([2**53, None, -(2**53)], dtype="Int64"), + "small": pd.array([1, None, 3], dtype="Int8"), + "u": np.array([1, 2, 3], dtype="uint16"), + "x": [1.5, 2.5, 3.5], + } + ) + native = _echo(monkeypatch) + out, report = fd.clean(df.copy(), engine="freshcore", return_report=True, **KW) + + assert native.calls == 1 + assert report.backend == "freshcore" + assert report.fallback_events == [] + assert report.backend_differences == [] + assert _dtypes(out) == ["int64", "Int64", "Int8", "uint16", "float64"] + pd.testing.assert_frame_equal(out, df) + assert out["big"].iloc[0] == 2**53 + + +def test_restored_integer_dtypes_match_pandas_engine(monkeypatch): + df = pd.DataFrame({"n": [1, 2, 3], "m": pd.array([4, None, 6], dtype="Int64")}) + expected = fd.clean(df.copy(), engine="pandas", **KW) + _echo(monkeypatch) + out = fd.clean(df.copy(), engine="freshcore", **KW) + pd.testing.assert_frame_equal(out, pd.DataFrame(expected)) + + +@pytest.mark.parametrize( + ("dtype", "native_values", "detail"), + [ + pytest.param("int64", [1.5, 2.0, 3.0], "non-integral values", id="non-integral"), + pytest.param( + "Int64", [1.0, 2.25, None], "non-integral values", id="nullable-non-integral" + ), + pytest.param( + "int64", + [1.0, None, 3.0], + "missing values, which a non-nullable integer dtype cannot hold", + id="missing-in-int64", + ), + pytest.param( + "int8", [1.0, 300.0, 3.0], "values outside the int8 range", id="out-of-range" + ), + ], +) +def test_unrestorable_integer_columns_are_reported(monkeypatch, dtype, native_values, detail): + df = pd.DataFrame({"n": pd.Series([1, 2, 3], dtype=dtype), "k": [1.0, 2.0, 3.0]}) + _echo(monkeypatch, n=native_values) + out, report = fd.clean(df, engine="freshcore", return_report=True, **KW) + + assert report.backend == "freshcore" + assert str(out["n"].dtype) == "float64" + assert out["n"].tolist()[0] == native_values[0] + (difference,) = report.backend_differences + assert difference["backend"] == "freshcore" + assert difference["step"] == "dtypes" + assert difference["column"] == "n" + assert detail in difference["detail"] + assert "float64" in difference["detail"] + + +@pytest.mark.parametrize( + "df", + [ + pytest.param(pd.DataFrame({0: [1.0, 2.0], 1: ["a", "b"]}), id="integer-labels"), + pytest.param(pd.DataFrame(np.array([[1.0, 2.0], [3.0, 4.0]])), id="range-index-labels"), + pytest.param( + pd.DataFrame({" First Name ": ["a", "b"], 2: [1, 2], 2.5: ["x", "y"]}), + id="mixed-renamed-labels", + ), + ], +) +def test_non_string_labels_round_trip(monkeypatch, df): + expected = fd.clean(df.copy(), engine="pandas", **KW) + native = _echo(monkeypatch) + out, report = fd.clean(df.copy(), engine="freshcore", return_report=True, **KW) + + assert native.calls == 1 + assert report.backend == "freshcore" + assert [c["name"] for c in native.payloads[0]["columns"]] == [str(c) for c in df.columns] + assert list(out.columns) == list(expected.columns) + assert [type(c) for c in out.columns] == [type(c) for c in expected.columns] + pd.testing.assert_frame_equal(out, pd.DataFrame(expected)) + + +@pytest.mark.parametrize( + "df", + [ + pytest.param(pd.DataFrame({"s": ["a", None, "b"], "n": [1, 2, 3]}), id="str-and-int"), + pytest.param( + pd.DataFrame({"n": np.array([2**53, -(2**53), 0], dtype="int64")}), + id="int-at-exact-limit", + ), + pytest.param( + pd.DataFrame({"n": pd.array([None, None], dtype="Int64")}), id="all-na-Int64" + ), + pytest.param(pd.DataFrame({1: [1.0, 2.0], "2": ["a", "b"]}), id="distinct-mixed-labels"), + ], +) +def test_supported_shapes_stay_native(monkeypatch, df): + native = _echo(monkeypatch) + _, report = fd.clean(df, engine="freshcore", return_report=True, **KW) + assert native.calls == 1 + assert report.backend == "freshcore" + assert report.fallback_events == []