From 5c230d01ccd6da882a35cfe7d31831356ffb77f0 Mon Sep 17 00:00:00 2001 From: Kevin Costner Date: Wed, 16 Sep 2026 23:16:24 +0530 Subject: [PATCH] fix(duckdb,polars): fall back instead of silently changing temporal columns On the fully native path (strategy="conservative", fix_dtypes=False) native ingestion changed temporal values with nothing recorded in fallback_events or backend_differences: - DuckDB stores INTERVAL in microseconds, so a timedelta64[ns] column was truncated on the way back (1431904ns -> 1431000ns, 5ns -> 0). - DuckDB returns TIMESTAMP WITH TIME ZONE in the session time zone at microsecond resolution, so a tz-aware datetime column came back in the machine's local zone with its sub-microsecond part dropped. - Both engines mishandle period and interval dtypes: DuckDB raises NotImplementedException("Data type 'period[M]' not recognized") and Polars ingests a period as its raw int64 ordinal (2020-01 -> 600) and an interval as a {left, right} struct. pandas_ingest_fallback_reason() now takes the engine name and rejects those columns, so the run takes the disclosed pandas fallback and fallback_policy ="error" can refuse it. Polars keeps its native path for nanosecond timedeltas and tz-aware datetimes, which it carries exactly. docs/fallback-matrix.md splits the old datetime/timedelta/categorical/period/ interval row into the three rows that are now true. Closes #443 --- CHANGELOG.md | 13 ++++ docs/fallback-matrix.md | 4 +- src/freshdata/execution/_ingest.py | 51 +++++++++++++-- src/freshdata/execution/backends/_duckdb.py | 2 +- src/freshdata/execution/backends/_polars.py | 2 +- .../test_ingest_and_output_format.py | 65 +++++++++++++++++-- 6 files changed, 122 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 566dd817..d8c3553a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,19 @@ adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] ### Fixed +- `engine="duckdb"` no longer silently changes temporal values on the fully + native path (`strategy="conservative"`, `fix_dtypes=False`). A nanosecond + `timedelta64[ns]` column was truncated to DuckDB's microsecond `INTERVAL` + (`5ns` came back as `0`) and a timezone-aware datetime column came back in + the machine's session time zone at microsecond resolution — both with an + empty `fallback_events`/`backend_differences`. Such columns now take the + disclosed pandas fallback instead, so the values survive unchanged and + `fallback_policy="error"` can refuse the run. +- `period` and `interval` columns no longer break native ingestion. DuckDB + raised `NotImplementedException: Data type 'period[M]' not recognized` and + Polars returned a period as its raw int64 ordinal (`2020-01` → `600`) and an + interval as a `{left, right}` struct. Both engines now fall back to the + pandas reference for those dtypes, with the reason recorded. - `CleanReport.revert()` no longer writes a restored value into other rows that share a duplicate index label. The undo log now records positional offsets and revert restores by position, so a frame with a non-unique index diff --git a/docs/fallback-matrix.md b/docs/fallback-matrix.md index 6a0b96f2..ccd475ac 100644 --- a/docs/fallback-matrix.md +++ b/docs/fallback-matrix.md @@ -40,7 +40,9 @@ FreshCore also runs its own config and data checks in | 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 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 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) | | contracts / validation / memory / profile replay | pandas | pandas | pandas | pandas | in-memory reference features (see [limitations](limitations.md)) | diff --git a/src/freshdata/execution/_ingest.py b/src/freshdata/execution/_ingest.py index 2e8185a6..61112aab 100644 --- a/src/freshdata/execution/_ingest.py +++ b/src/freshdata/execution/_ingest.py @@ -2,10 +2,12 @@ :meth:`~freshdata.execution.PlanGenerator.fallback_reason` looks only at the config. Some *inputs* cannot enter a native engine faithfully under any config: -Arrow/Polars ingestion rejects duplicate column labels, and an object column +Arrow/Polars ingestion rejects duplicate column labels, an object column holding mixed value types is either rejected (Polars) or silently cast to text -(DuckDB). Those runs take the disclosed pandas fallback instead, which -``fallback_policy="error"`` still blocks before any pandas work. +(DuckDB), and some pandas extension dtypes survive the round trip only as a +different — sometimes silently truncated — value. Those runs take the disclosed +pandas fallback instead, which ``fallback_policy="error"`` still blocks before +any pandas work. """ from __future__ import annotations @@ -16,8 +18,13 @@ _MIXED_KINDS = frozenset({"mixed", "mixed-integer"}) -def pandas_ingest_fallback_reason(source: Any) -> str | None: - """Return why *source* must be cleaned by the pandas reference, or ``None``.""" +def pandas_ingest_fallback_reason(source: Any, engine: str | None = None) -> str | None: + """Return why *source* must be cleaned by the pandas reference, or ``None``. + + *engine* names the backend that would ingest *source*. Checks that belong + to one backend's type system only apply to that backend; omit it to run the + checks every native engine shares. + """ import pandas as pd from pandas.api.types import infer_dtype, is_object_dtype @@ -26,9 +33,41 @@ def pandas_ingest_fallback_reason(source: Any) -> str | None: if source.columns.duplicated().any(): return "duplicate input column labels require the pandas reference path" for i, dtype in enumerate(source.dtypes): + label = source.columns[i] if is_object_dtype(dtype) and infer_dtype(source.iloc[:, i], skipna=True) in _MIXED_KINDS: return ( - f"object column {source.columns[i]!r} mixes value types (e.g. numbers " + f"object column {label!r} mixes value types (e.g. numbers " "and strings), which native ingestion would reject or cast to text" ) + lossy = _lossy_dtype_reason(dtype, label, engine) + if lossy is not None: + return lossy + return None + + +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.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 + # an interval as a {left, right} struct. + return ( + f"column {label!r} has dtype {name}: native ingestion does not carry " + "period and interval dtypes, so they require the pandas reference path" + ) + if engine != "duckdb": + return None + if name == "timedelta64[ns]": + return ( + f"timedelta column {label!r} is nanosecond-resolution: DuckDB stores " + "INTERVAL in microseconds and would truncate it, so it requires the " + "pandas reference path" + ) + if name.startswith("datetime64[") and getattr(dtype, "tz", None) is not None: + return ( + f"datetime column {label!r} is timezone-aware ({dtype.tz}): DuckDB " + "returns TIMESTAMP WITH TIME ZONE in the session time zone and at " + "microsecond resolution, so it requires the pandas reference path" + ) return None diff --git a/src/freshdata/execution/backends/_duckdb.py b/src/freshdata/execution/backends/_duckdb.py index 0b35236e..169200e0 100644 --- a/src/freshdata/execution/backends/_duckdb.py +++ b/src/freshdata/execution/backends/_duckdb.py @@ -147,7 +147,7 @@ def execute( plan_cols = self._peek_columns(source) plan = PlanGenerator(config).plan(plan_cols) - reason = plan.fallback_reason or pandas_ingest_fallback_reason(source) + reason = plan.fallback_reason or pandas_ingest_fallback_reason(source, self.name) if reason is None and self._pandas_index_forces_fallback(source): reason = "pandas index semantics" if reason is not None: diff --git a/src/freshdata/execution/backends/_polars.py b/src/freshdata/execution/backends/_polars.py index 7944d6ff..7c39b577 100644 --- a/src/freshdata/execution/backends/_polars.py +++ b/src/freshdata/execution/backends/_polars.py @@ -116,7 +116,7 @@ def execute( # raises on the inputs pandas_ingest_fallback_reason flags. reason = ( PlanGenerator(config, backend=self.name).fallback_reason() - or pandas_ingest_fallback_reason(source) + or pandas_ingest_fallback_reason(source, self.name) ) if reason is None and self._pandas_index_forces_fallback(source): reason = "pandas index semantics" diff --git a/tests/test_execution/test_ingest_and_output_format.py b/tests/test_execution/test_ingest_and_output_format.py index c8a32849..dad29ce5 100644 --- a/tests/test_execution/test_ingest_and_output_format.py +++ b/tests/test_execution/test_ingest_and_output_format.py @@ -51,6 +51,56 @@ def test_ingest_fallback_is_blocked_by_error_policy(engine): fd.clean(df, config=NATIVE, engine=engine, fallback_policy="error") +@pytest.mark.parametrize( + ("label", "values"), + [ + ("period", pd.period_range("2020-01", periods=2, freq="M")), + ("interval", pd.arrays.IntervalArray.from_breaks([0, 1, 2])), + ], +) +@pytest.mark.parametrize("engine", ENGINES) +def test_period_and_interval_columns_fall_back(engine, label, values): + """DuckDB rejects both dtypes; Polars turns them into ordinals/structs (#443).""" + df = pd.DataFrame({label: values}) + ref = fd.clean(df, config=NATIVE, engine="pandas") + out, report = fd.clean(df, config=NATIVE, engine=engine, return_report=True) + pd.testing.assert_frame_equal(pd.DataFrame(out), pd.DataFrame(ref)) + assert any("period and interval dtypes" in reason for reason in _reasons(report)) + + +def test_duckdb_nanosecond_timedelta_falls_back_instead_of_truncating(): + """DuckDB INTERVAL is microseconds, so 5ns would come back as 0 (#443).""" + df = pd.DataFrame({"td": pd.to_timedelta([1431904, 5], unit="ns")}) + out, report = fd.clean(df, config=NATIVE, engine="duckdb", return_report=True) + assert [v.value for v in out["td"]] == [1431904, 5] + assert any("would truncate it" in reason for reason in _reasons(report)) + + +def test_duckdb_tz_aware_datetime_falls_back_instead_of_rewriting_the_zone(): + """DuckDB returns TIMESTAMPTZ in the session zone, at microseconds (#443).""" + df = pd.DataFrame({"t": pd.to_datetime(["2020-01-01 00:00:00.000000123"] * 2, utc=True)}) + out, report = fd.clean(df, config=NATIVE, engine="duckdb", return_report=True) + assert str(out["t"].dtype) == "datetime64[ns, UTC]" + assert [v.value for v in out["t"]] == [df["t"].iloc[0].value] * 2 + assert any("timezone-aware" in reason for reason in _reasons(report)) + + +@pytest.mark.parametrize( + "values", + [ + pd.to_timedelta([1431904, 5], unit="ns"), + pd.to_datetime(["2020-01-01 00:00:00.000000123"] * 2, utc=True), + ], +) +def test_polars_keeps_nanosecond_temporal_columns_native(values): + """The DuckDB-only checks must not cost Polars its native path (#443).""" + df = pd.DataFrame({"t": values, "k": [1.0, 2.0]}) + out, report = fd.clean(df, config=NATIVE, engine="polars", return_report=True) + assert report.backend == "polars" + assert report.fallback_events == [] + assert list(out["t"]) == list(values) + + @pytest.mark.parametrize("engine", ENGINES) def test_single_type_object_columns_stay_native(engine): df = pd.DataFrame({"s": ["a", None, "b"], "k": [1.0, 2.0, 3.0]}) @@ -84,8 +134,9 @@ def test_engine_config_rejects_a_foreign_native_handle(engine, output_format): def test_clean_rejects_a_foreign_native_handle_under_error_policy(engine, output_format): df = pd.DataFrame({"a": [1.0, None, 3.0]}) with pytest.raises(ValueError, match="native"): - fd.clean(df, config=NATIVE, engine=engine, output_format=output_format, - fallback_policy="error") + fd.clean( + df, config=NATIVE, engine=engine, output_format=output_format, fallback_policy="error" + ) @pytest.mark.parametrize( @@ -96,8 +147,9 @@ def test_clean_rejects_a_foreign_native_handle_under_error_policy(engine, output def test_handle_format_selects_its_own_engine(output_format, handle, engine): df = pd.DataFrame({"a": [1.0, None, 3.0]}) kwargs = {} if engine is None else {"engine": engine} - out, report = fd.clean(df, config=NATIVE, output_format=output_format, - return_report=True, **kwargs) + out, report = fd.clean( + df, config=NATIVE, output_format=output_format, return_report=True, **kwargs + ) assert type(out).__name__ == handle assert report.fallback_events == [] @@ -108,7 +160,8 @@ def test_handle_format_selects_its_own_engine(output_format, handle, engine): def test_handle_request_with_a_recorded_fallback_returns_pandas(engine, output_format): df = pd.DataFrame({"a": [1.0, None, 3.0]}) # The default balanced strategy runs on the pandas reference. - out, report = fd.clean(df, engine=engine, output_format=output_format, - return_report=True, verbose=False) + out, report = fd.clean( + df, engine=engine, output_format=output_format, return_report=True, verbose=False + ) assert isinstance(out, pd.DataFrame) assert report.fallback_events