fix(duckdb): preserve case-colliding pandas columns

CHANGELOG.md

@@ -8,6 +8,10 @@ adheres to [Semantic Versioning](https://semver.org/).
 
 ### Fixed
 
+- DuckDB now discloses a pandas fallback for input headers that collide under
+  its ASCII-insensitive identifier rules. This prevents projection aliases
+  from selecting the wrong column or dropping a nonempty one (#527).
+
 - MissForest now preserves declared identifiers and targets whose headers were
   normalized during cleaning, including headers that need deduplication. Exact
   raw declarations follow the positional rename map, and MissForest uses the

docs/fallback-matrix.md

@@ -40,7 +40,7 @@ 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 with zero columns and one or more rows | pandas | pandas | — | — | the pandas reference preserves the rows and index; Polars 1.36 loses these rows during ingestion, and DuckDB cannot register a zero-column frame. Use pandas output to retain the row count |
-| 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 whose column labels collide once stringified (e.g. `1` and `"1"`) | native | pandas | — | pandas | DuckDB registration and subsequent alias repairs can redirect stages to the wrong column. 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 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 |
@@ -51,6 +51,13 @@ FreshCore also runs its own config and data checks in
 are all-or-nothing per run, not per step) and is recorded as a
 `fallback_event` with the exact reason.
 
+DuckDB also uses the recorded pandas fallback when input labels collide under
+its ASCII-insensitive identifier rules, such as `"A"` / `"a"` or `7` / `"7"`.
+Registration suffixes can otherwise collide with subsequent column-name repairs
+and redirect later stages to the wrong column. This input check applies with
+either value of `column_names`; `fallback_policy="error"` rejects it before
+ingestion. Non-ASCII case variants such as `"Ä"` / `"ä"` stay distinct in DuckDB.
+
 ## Refusing the fallback: `fallback_policy`
 
 You never have to *discover* a fallback after the fact:

src/freshdata/execution/_ingest.py

@@ -16,6 +16,7 @@
 
 #: ``infer_dtype`` kinds of an object column that native ingestion cannot keep as-is.
 _MIXED_KINDS = frozenset({"mixed", "mixed-integer"})
+_DUCKDB_ASCII_FOLD = str.maketrans("ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz")
 
 
 def pandas_ingest_fallback_reason(source: Any, engine: str | None = None) -> str | None:
@@ -36,6 +37,15 @@ def pandas_ingest_fallback_reason(source: Any, engine: str | None = None) -> str
         return "zero-column pandas source requires the pandas reference to preserve its rows"
     if source.columns.duplicated().any():
         return "duplicate input column labels require the pandas reference path"
+    if engine == "duckdb":
+        # DuckDB identifiers fold ASCII case, not Unicode case. Registration
+        # suffixes can collide again after renaming and redirect later stages.
+        names = [str(label).translate(_DUCKDB_ASCII_FOLD) for label in source.columns]
+        if len(set(names)) != len(names):
+            return (
+                "case-insensitive input column labels require the pandas reference "
+                "path to prevent DuckDB alias collisions and preserve every column"
+            )
     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:

tests/test_execution/test_duckdb_column_collisions.py

@@ -0,0 +1,93 @@
+"""DuckDB cannot safely carry case-insensitive pandas header collisions."""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+import freshdata as fd
+from freshdata.execution import FallbackError, FallbackWarning
+from freshdata.execution.backends._duckdb import DuckDBEngine
+
+pytest.importorskip("duckdb")
+KW = {"strategy": "conservative", "fix_dtypes": False, "verbose": False}
+
+
+def _frame(empty_second=False, unrelated_empty=False):
+    frame = pd.DataFrame({
+        "A": [1, 2, 3], "a": [None] * 3 if empty_second else ["x", "y", "z"],
+        "A 1": pd.Series([4, 5, 6], dtype="Int64"),
+    })
+    if unrelated_empty:
+        frame["Unused"] = None
+    return frame
+
+
+@pytest.mark.parametrize("column_names", [True, False])
+@pytest.mark.parametrize(("empty_second", "unrelated_empty"), [
+    (False, False), (True, False), (False, True),
+])
+def test_case_colliding_headers_preserve_pandas_values_and_schema(
+    column_names, empty_second, unrelated_empty
+):
+    frame = _frame(empty_second, unrelated_empty)
+    original = frame.copy(deep=True)
+    expected, reference = fd.clean(
+        frame, engine="pandas", column_names=column_names, return_report=True, **KW
+    )
+    out, report = fd.clean(
+        frame, engine="duckdb", column_names=column_names, return_report=True, **KW
+    )
+    pd.testing.assert_frame_equal(pd.DataFrame(out), pd.DataFrame(expected))
+    pd.testing.assert_frame_equal(frame, original)
+    assert report.backend == "pandas"
+    assert report.requested_backend == "duckdb"
+    assert "case-insensitive" in report.fallback_events[0]["fallback_reason"]
+    assert [(a.step, a.count) for a in report.actions] == [
+        (a.step, a.count) for a in reference.actions
+    ]
+
+
+@pytest.mark.parametrize("column_names", [True, False])
+def test_case_collision_warn_policy_is_disclosed(column_names):
+    with pytest.warns(FallbackWarning, match="case-insensitive"):
+        out = fd.clean(_frame(True), engine="duckdb", column_names=column_names,
+                       fallback_policy="warn", **KW)
+    assert out.shape == (3, 2)
+
+
+def test_strict_policy_rejects_before_native_ingestion_or_pandas_work(monkeypatch):
+    def forbidden(*args, **kwargs):
+        raise AssertionError("strict policy must reject before cleaning")
+
+    monkeypatch.setattr(DuckDBEngine, "_register_source", forbidden)
+    monkeypatch.setattr(DuckDBEngine, "_fallback", forbidden)
+    with pytest.raises(FallbackError, match="case-insensitive"):
+        fd.clean(_frame(True), engine="duckdb", fallback_policy="error", **KW)
+
+
+def test_native_handle_request_reports_the_materialized_preservation_fallback():
+    out, report = fd.clean(_frame(True), engine="duckdb", output_format="duckdb",
+                           return_report=True, **KW)
+    assert isinstance(out, pd.DataFrame)
+    assert report.materialized is True
+    assert out.shape == (3, 2)
+    assert report.fallback_events
+
+
+@pytest.mark.parametrize("names", [("A B", "A-B"), ("Ä", "ä"), ("ß", "ss")])
+def test_non_colliding_duckdb_identifiers_remain_native(names):
+    frame = pd.DataFrame([[1, 2], [3, 4]], columns=names)
+    out, report = fd.clean(frame, engine="duckdb", return_report=True, **KW)
+    assert out.shape == (2, 2)
+    assert report.backend == "duckdb"
+    assert report.fallback_events == []
+
+
+def test_non_string_labels_that_collide_in_duckdb_are_preserved_by_pandas():
+    frame = pd.DataFrame({7: [1, 2, 3], "7": [4, 5, 6]})
+    expected = fd.clean(frame, engine="pandas", column_names=False, **KW)
+    out, report = fd.clean(frame, engine="duckdb", column_names=False,
+                           return_report=True, **KW)
+    pd.testing.assert_frame_equal(pd.DataFrame(out), pd.DataFrame(expected))
+    assert report.backend == "pandas"