fix(duckdb,polars): fall back instead of silently changing temporal columns

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

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)) |
 

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

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:

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"

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