fix(engines): disclose ingestion fallbacks; validate engine/output_format pairs

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

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

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,

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,20 +80,20 @@ 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
         except ImportError:  # pragma: no cover - guarded upstream
             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
 

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}, "

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

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)

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