From 2a3ae88a06d9dcceff8c1860691a27b032885fa0 Mon Sep 17 00:00:00 2001 From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com> Date: Tue, 15 Sep 2026 01:27:01 +0530 Subject: [PATCH] fix(engines): disclose ingestion fallbacks; validate engine/output_format pairs - #206: a pandas source whose object column mixes value types crashed the Polars engine during pl.from_pandas and was silently cast to text by DuckDB, and duplicate column labels crashed Polars and came back mis-renamed from DuckDB. execution/_ingest.py now flags those inputs before ingestion, and both engines take the recorded pandas fallback (fallback_policy="error" still raises first). The Polars engine now decides fallbacks before converting the source. - #205: requesting another engine's native handle silently returned a different type, even under fallback_policy="error" (e.g. engine="duckdb" with output_format="polars-lazy" returned a DuckDBPyRelation). EngineConfig now rejects the pairing with a ValueError; engine="auto" (and the default engine) picks the engine that owns the handle format. _convert_output returns a materialized frame in place of a handle only when the report records the pandas fallback. Docs: fallback-matrix lists the two input-driven fallbacks; backends.md says a handle format needs its own engine. Closes #205 Closes #206 --- docs/backends.md | 5 + docs/fallback-matrix.md | 7 +- src/freshdata/api.py | 19 ++- src/freshdata/execution/__init__.py | 41 +++++-- src/freshdata/execution/_config.py | 9 ++ src/freshdata/execution/_ingest.py | 34 ++++++ src/freshdata/execution/backends/_duckdb.py | 7 +- src/freshdata/execution/backends/_polars.py | 33 +++-- .../test_ingest_and_output_format.py | 114 ++++++++++++++++++ 9 files changed, 239 insertions(+), 30 deletions(-) create mode 100644 src/freshdata/execution/_ingest.py create mode 100644 tests/test_execution/test_ingest_and_output_format.py diff --git a/docs/backends.md b/docs/backends.md index 5503eb04..a64aec86 100644 --- a/docs/backends.md +++ b/docs/backends.md @@ -36,6 +36,11 @@ memory: | `"duckdb"` | `DuckDBPyRelation` (un-fetched) | **No** — you call `.fetchdf()`/`.arrow()` | | `"polars-lazy"` | `pl.LazyFrame` (un-collected) | **No** — you call `.collect()` | +A native handle comes from its own engine: `"duckdb"` needs `engine="duckdb"` and +`"polars-lazy"` needs `engine="polars"`. With `engine="auto"` (or no `engine`), +freshdata picks that engine for you; any other pairing raises `ValueError` rather +than returning a different type. + Neither handle fetches/collects the *result* until you ask, but they are not equal during the pipeline: the DuckDB path keeps peak memory well below the eager equivalent, while the Polars pipeline currently collects intermediates diff --git a/docs/fallback-matrix.md b/docs/fallback-matrix.md index d46a3f35..c39385f8 100644 --- a/docs/fallback-matrix.md +++ b/docs/fallback-matrix.md @@ -8,8 +8,9 @@ engine delegates the whole pipeline to pandas** (recorded on `strategy="conservative"` with `fix_dtypes=False`. This table is transcribed from the single source of truth, -`PlanGenerator.fallback_reason()` in `src/freshdata/execution/_plan.py` — -if you change that function, change this page. +`PlanGenerator.fallback_reason()` in `src/freshdata/execution/_plan.py`, plus +the input checks in `pandas_ingest_fallback_reason()` +(`src/freshdata/execution/_ingest.py`) — if you change either, change this page. | Operation / config | polars | duckdb | spark | freshcore | Why the fallback exists | |---|---|---|---|---|---| @@ -28,6 +29,8 @@ if you change that function, change this page. | `drop_constant_columns` | pandas | pandas | pandas | pandas | needs a data scan before planning (two-phase plan not built) | | `optimize_memory` | pandas | pandas | pandas | pandas | pandas-specific downcasting — meaningless for other outputs, by design | | 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"`) | | 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/src/freshdata/api.py b/src/freshdata/api.py index 22b55f0b..e4e5d173 100644 --- a/src/freshdata/api.py +++ b/src/freshdata/api.py @@ -18,6 +18,7 @@ from .engine.context import build_contexts from .engine.model_select import EngineMode, rank_missing_models from .execution import run_with_engine +from .execution._config import NATIVE_HANDLE_FORMATS from .parsers.registry import get_parser from .plan import suggest_plan from .profile import Profile, build_profile @@ -55,6 +56,16 @@ def _is_native_engine_source(df: object) -> bool: return False +def _auto_engine_for(df: object, engine: str, output_format: str) -> str: + """Resolve the default ``engine="pandas"`` to ``"auto"`` when only a native + engine can serve the input or the requested native handle format.""" + if engine == "pandas" and ( + _is_native_engine_source(df) or output_format in NATIVE_HANDLE_FORMATS + ): + return "auto" + return engine + + def _fold_context_options( options: dict[str, object], *, @@ -349,16 +360,14 @@ def clean( if fallback_policy is not None: from .execution import EngineConfig as _EngineConfig # noqa: PLC0415 - if engine == "pandas" and engine_config is None and not _is_native_engine_source(df): + resolved_engine = _auto_engine_for(df, engine, output_format) + if resolved_engine == "pandas" and engine_config is None: raise TypeError( "fallback_policy applies to native engines; engine='pandas' " "cannot fall back (pass engine='polars'/'duckdb'/... or an " "engine_config)" ) if engine_config is None: - resolved_engine = ( - "auto" if engine == "pandas" and _is_native_engine_source(df) else engine - ) engine_config = _EngineConfig( engine=resolved_engine, output_format=output_format, @@ -379,7 +388,7 @@ def clean( df, config, options, - engine="auto" if native_source and engine == "pandas" else engine, + engine=_auto_engine_for(df, engine, output_format), output_format=output_format, engine_config=engine_config, return_report=return_report, diff --git a/src/freshdata/execution/__init__.py b/src/freshdata/execution/__init__.py index 65243084..2e318ab5 100644 --- a/src/freshdata/execution/__init__.py +++ b/src/freshdata/execution/__init__.py @@ -20,6 +20,7 @@ from ._base import ExecutionEngine from ._config import ( FALLBACK_POLICIES, + NATIVE_HANDLE_ENGINES, EngineConfig, EngineSelector, FallbackError, @@ -58,7 +59,18 @@ def _is_spark_frame(frame: Any) -> bool: return isinstance(frame, SparkDataFrame) -def _convert_output(frame: Any, output_format: str) -> Any: +def _require_recorded_fallback(frame: Any, output_format: str, report: Any) -> None: + """Allow a materialized frame in place of a native handle only after a + pandas fallback that the report discloses.""" + if report is not None and report.fallback_events: + return + raise RuntimeError( + f"output_format={output_format!r} expects a native handle, but the backend " + f"returned {type(frame).__name__} without recording a pandas fallback" + ) + + +def _convert_output(frame: Any, output_format: str, report: Any = None) -> Any: """Convert a backend-native frame to the requested output format.""" import pandas as pd @@ -68,12 +80,11 @@ def _convert_output(frame: Any, output_format: str) -> Any: # untouched. The backend is responsible for *not* having collected/fetched # it (see the DuckDB/Polars engines). We never silently materialize here. # - # A pandas frame at this point means the backend transparently fell back to - # the pandas pipeline (e.g. the balanced decision engine, which only runs on - # pandas). That fallback is already disclosed on the report - # (``fallback_events`` + ``backend="pandas"``), so we return the materialized - # frame rather than raising — the caller can read the report to see why the - # native handle wasn't available. + # EngineConfig only pairs a handle format with the engine that produces it, + # so a pandas frame here means the backend fell back to the pandas pipeline + # (e.g. the balanced decision engine). That is returned as-is only when the + # fallback is recorded on the report (``fallback_events``); anything else is + # a bug, not a silent substitution. if output_format == "duckdb": try: import duckdb @@ -81,7 +92,8 @@ def _convert_output(frame: Any, output_format: str) -> Any: duckdb = None # type: ignore[assignment] if duckdb is not None and isinstance(frame, duckdb.DuckDBPyRelation): return frame - return frame # disclosed pandas fallback + _require_recorded_fallback(frame, output_format, report) + return frame if output_format == "polars-lazy": from ._lazy import require_polars @@ -90,7 +102,8 @@ def _convert_output(frame: Any, output_format: str) -> Any: return frame if isinstance(frame, pl.DataFrame): return frame.lazy() - return frame # disclosed pandas fallback + _require_recorded_fallback(frame, output_format, report) + return frame if output_format == "spark": if is_spark: @@ -157,7 +170,11 @@ def run_with_engine( requested = engine_config.engine resolved = engine_config.engine if resolved == "auto": - resolved = EngineSelector.select(source, engine_config) + # A native handle format can only come from its own engine. + resolved = ( + NATIVE_HANDLE_ENGINES.get(engine_config.output_format) + or EngineSelector.select(source, engine_config) + ) engine_config = replace(engine_config, engine=resolved) # Semantic cleaning is scored on the pandas reference path. On a native @@ -183,7 +200,7 @@ def run_with_engine( cleaned, report = run_pipeline(frame, config) report.backend = "pandas" report.record_fallback(resolved, "semantic", reason) - result = _convert_output(cleaned, engine_config.output_format) + result = _convert_output(cleaned, engine_config.output_format, report) _finish_report(report, requested, "pandas", result) return (result, report) if return_report else result @@ -195,7 +212,7 @@ def run_with_engine( from ..semantic.native import run_semantic_native cleaned_native = run_semantic_native(cleaned_native, config, report, engine=resolved) - result = _convert_output(cleaned_native, engine_config.output_format) + result = _convert_output(cleaned_native, engine_config.output_format, report) _finish_report(report, requested, resolved, result) return (result, report) if return_report else result diff --git a/src/freshdata/execution/_config.py b/src/freshdata/execution/_config.py index 68c559be..9bacce1d 100644 --- a/src/freshdata/execution/_config.py +++ b/src/freshdata/execution/_config.py @@ -32,6 +32,8 @@ MATERIALIZING_FORMATS = frozenset({"pandas", "polars", "arrow", "spark"}) #: Output formats that hand back a native, lazy/streaming handle instead. NATIVE_HANDLE_FORMATS = frozenset({"duckdb", "polars-lazy"}) +#: The one engine that can produce each native handle format. +NATIVE_HANDLE_ENGINES = {"duckdb": "duckdb", "polars-lazy": "polars"} #: What to do when a native backend must delegate to the pandas reference: #: ``"allow"`` (record silently on the report), ``"warn"`` (also emit a #: :class:`FallbackWarning`), ``"error"`` (raise :class:`FallbackError` before @@ -117,6 +119,13 @@ def __post_init__(self) -> None: raise ValueError( f"output_format must be one of {OUTPUT_FORMATS}, got {self.output_format!r}" ) + handle_engine = NATIVE_HANDLE_ENGINES.get(self.output_format) + if handle_engine is not None and self.engine not in (handle_engine, "auto"): + raise ValueError( + f"output_format={self.output_format!r} returns a native {handle_engine} " + f"handle, which engine={self.engine!r} cannot produce; use " + f"engine={handle_engine!r} or engine='auto'" + ) if self.fallback_policy not in FALLBACK_POLICIES: raise ValueError( f"fallback_policy must be one of {FALLBACK_POLICIES}, " diff --git a/src/freshdata/execution/_ingest.py b/src/freshdata/execution/_ingest.py new file mode 100644 index 00000000..2e8185a6 --- /dev/null +++ b/src/freshdata/execution/_ingest.py @@ -0,0 +1,34 @@ +"""Input-driven reasons a pandas source must take the pandas reference path. + +: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 +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. +""" + +from __future__ import annotations + +from typing import Any + +#: ``infer_dtype`` kinds of an object column that native ingestion cannot keep as-is. +_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``.""" + import pandas as pd + from pandas.api.types import infer_dtype, is_object_dtype + + if not isinstance(source, pd.DataFrame): + return None + if source.columns.duplicated().any(): + return "duplicate input column labels require the pandas reference path" + for i, dtype in enumerate(source.dtypes): + 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 " + "and strings), which native ingestion would reject or cast to text" + ) + return None diff --git a/src/freshdata/execution/backends/_duckdb.py b/src/freshdata/execution/backends/_duckdb.py index f4dbbb5a..91cd86de 100644 --- a/src/freshdata/execution/backends/_duckdb.py +++ b/src/freshdata/execution/backends/_duckdb.py @@ -26,6 +26,7 @@ from ...steps.duplicates import check_duplicate_ratio, report_detected_duplicates from .._base import ExecutionEngine from .._config import NATIVE_HANDLE_FORMATS, enforce_fallback_policy +from .._ingest import pandas_ingest_fallback_reason from .._lazy import has_duckdb, has_polars, require_duckdb from .._metadata import MetadataScanner, is_duckdb_float from .._native_steps import ( @@ -111,8 +112,10 @@ def execute( plan_cols = self._peek_columns(source) plan = PlanGenerator(config).plan(plan_cols) - if plan.needs_fallback or self._pandas_index_forces_fallback(source): - reason = plan.fallback_reason or "pandas index semantics" + reason = plan.fallback_reason or pandas_ingest_fallback_reason(source) + if reason is None and self._pandas_index_forces_fallback(source): + reason = "pandas index semantics" + if reason is not None: enforce_fallback_policy(engine_config, "duckdb", "pipeline", reason) log.warning("freshdata DuckDBEngine: falling back to pandas (%s)", reason) cleaned, report = self._fallback(source, config) diff --git a/src/freshdata/execution/backends/_polars.py b/src/freshdata/execution/backends/_polars.py index 4a175b42..c2e308c2 100644 --- a/src/freshdata/execution/backends/_polars.py +++ b/src/freshdata/execution/backends/_polars.py @@ -21,6 +21,7 @@ from ...steps.duplicates import check_duplicate_ratio, report_detected_duplicates from .._base import ExecutionEngine from .._config import enforce_fallback_policy +from .._ingest import pandas_ingest_fallback_reason from .._lazy import require_polars from .._metadata import MetadataScanner from .._native_steps import ( @@ -111,18 +112,22 @@ def execute( self._configure_threads(engine_config) started = time.perf_counter() + # Decide config- and input-driven fallbacks before ingestion: pl.from_pandas + # raises on the inputs pandas_ingest_fallback_reason flags. + reason = ( + PlanGenerator(config, backend=self.name).fallback_reason() + or pandas_ingest_fallback_reason(source) + ) + if reason is None and self._pandas_index_forces_fallback(source): + reason = "pandas index semantics" + if reason is not None: + return self._delegate_to_pandas(source, config, engine_config, reason) + lf, memory_before = self._to_lazy(source, pl) names = list(lf.collect_schema().names()) plan = PlanGenerator(config, backend=self.name).plan(names) - - if plan.needs_fallback or self._pandas_index_forces_fallback(source): - reason = plan.fallback_reason or "pandas index semantics" - enforce_fallback_policy(engine_config, "polars", "pipeline", reason) - log.warning("freshdata PolarsEngine: falling back to pandas (%s)", reason) - cleaned, report = self._fallback(source, config) - report.backend = "pandas" - report.record_fallback("polars", "pipeline", reason) - return cleaned, report + if plan.fallback_reason is not None: + return self._delegate_to_pandas(source, config, engine_config, plan.fallback_reason) meta = MetadataScanner.from_polars_lazy(lf) report = init_report(meta, memory_before) @@ -145,6 +150,16 @@ def execute( finalize_report(report, cleaned, started) return cleaned, report + def _delegate_to_pandas( + self, source: Any, config: CleanConfig, engine_config: EngineConfig, reason: str + ) -> tuple[Any, CleanReport]: + enforce_fallback_policy(engine_config, "polars", "pipeline", reason) + log.warning("freshdata PolarsEngine: falling back to pandas (%s)", reason) + cleaned, report = self._fallback(source, config) + report.backend = "pandas" + report.record_fallback("polars", "pipeline", reason) + return cleaned, report + def _fallback(self, source: Any, config: CleanConfig) -> tuple[Any, CleanReport]: from ...cleaner import run_pipeline diff --git a/tests/test_execution/test_ingest_and_output_format.py b/tests/test_execution/test_ingest_and_output_format.py new file mode 100644 index 00000000..c8a32849 --- /dev/null +++ b/tests/test_execution/test_ingest_and_output_format.py @@ -0,0 +1,114 @@ +"""Inputs native engines cannot ingest (#206) and engine/output_format pairs (#205).""" + +from __future__ import annotations + +import pandas as pd +import pytest + +import freshdata as fd +from freshdata import CleanConfig +from freshdata.execution import EngineConfig, FallbackError + +pytest.importorskip("polars") +pytest.importorskip("duckdb") + +NATIVE = CleanConfig(strategy="conservative", fix_dtypes=False, verbose=False) +ENGINES = ("polars", "duckdb") + + +def _reasons(report): + return [event["fallback_reason"] for event in report.fallback_events] + + +# -- #206 inputs that need the pandas reference -------------------------------- + + +@pytest.mark.parametrize("engine", ENGINES) +def test_mixed_object_column_falls_back_with_pandas_values(engine): + df = pd.DataFrame({"m": [1, "a", None], "k": [1.0, 2.0, 3.0]}) + ref = fd.clean(df, config=NATIVE, engine="pandas") + out, report = fd.clean(df, config=NATIVE, engine=engine, return_report=True) + assert [type(v).__name__ for v in out["m"].dropna()] == ["int", "str"] + # engine="pandas" returns a CleanResult wrapper; compare the plain frames. + pd.testing.assert_frame_equal(pd.DataFrame(out), pd.DataFrame(ref)) + assert report.backend == "pandas" + assert any("mixes value types" in reason for reason in _reasons(report)) + + +@pytest.mark.parametrize("engine", ENGINES) +def test_duplicate_labels_fall_back_with_pandas_columns(engine): + df = pd.DataFrame([[1.0, 2.0], [3.0, 4.0]], columns=["x", "x"]) + ref = fd.clean(df, config=NATIVE, engine="pandas") + out, report = fd.clean(df, config=NATIVE, engine=engine, return_report=True) + assert list(out.columns) == list(ref.columns) == ["x", "x_2"] + assert any("duplicate input column labels" in reason for reason in _reasons(report)) + + +@pytest.mark.parametrize("engine", ENGINES) +def test_ingest_fallback_is_blocked_by_error_policy(engine): + df = pd.DataFrame({"m": [1, "a", None]}) + with pytest.raises(FallbackError, match="mixes value types"): + fd.clean(df, config=NATIVE, engine=engine, fallback_policy="error") + + +@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]}) + _, report = fd.clean(df, config=NATIVE, engine=engine, return_report=True) + assert report.backend == engine + assert report.fallback_events == [] + + +# -- #205 native handle formats belong to one engine --------------------------- + + +@pytest.mark.parametrize( + ("engine", "output_format"), + [ + ("duckdb", "polars-lazy"), + ("polars", "duckdb"), + ("pandas", "duckdb"), + ("pandas", "polars-lazy"), + ("spark", "polars-lazy"), + ("freshcore", "duckdb"), + ], +) +def test_engine_config_rejects_a_foreign_native_handle(engine, output_format): + with pytest.raises(ValueError, match="native"): + EngineConfig(engine=engine, output_format=output_format) + + +@pytest.mark.parametrize( + ("engine", "output_format"), [("duckdb", "polars-lazy"), ("polars", "duckdb")] +) +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") + + +@pytest.mark.parametrize( + ("output_format", "handle"), + [("duckdb", "DuckDBPyRelation"), ("polars-lazy", "LazyFrame")], +) +@pytest.mark.parametrize("engine", ["auto", None], ids=["auto", "default"]) +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) + assert type(out).__name__ == handle + assert report.fallback_events == [] + + +@pytest.mark.parametrize( + ("engine", "output_format"), [("duckdb", "duckdb"), ("polars", "polars-lazy")] +) +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) + assert isinstance(out, pd.DataFrame) + assert report.fallback_events