diff --git a/src/freshdata/_util.py b/src/freshdata/_util.py index ea362ef..96bfa62 100644 --- a/src/freshdata/_util.py +++ b/src/freshdata/_util.py @@ -235,3 +235,13 @@ def mask_sensitive_value(value: object) -> str: """ digest = hashlib.sha256(repr(value).encode("utf-8")).hexdigest()[:8] return f"[SENSITIVE:{digest}]" + + +#: Every character ``str.isspace`` accepts, i.e. what ``str.strip()`` removes. +#: Native engines strip exactly this set so they match the pandas reference +#: (RE2's ``\s`` is ASCII-only; Rust's whitespace excludes ``\x1c``-``\x1f``). +PY_WHITESPACE = ( + "\t\n\x0b\x0c\r\x1c\x1d\x1e\x1f \x85\xa0\u1680" + "\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a" + "\u2028\u2029\u202f\u205f\u3000" +) diff --git a/src/freshdata/execution/_metadata.py b/src/freshdata/execution/_metadata.py index 4a989fb..f0271b4 100644 --- a/src/freshdata/execution/_metadata.py +++ b/src/freshdata/execution/_metadata.py @@ -2,8 +2,8 @@ :class:`ColumnMetadata` is everything the planner and the selector need to make decisions without materialising a dataset. Each scanner uses the cheapest path -its backend offers: pandas describe on a sample, polars lazy aggregates, DuckDB -``SUMMARIZE``, or the Parquet footer. +its backend offers: pandas describe on a sample, polars lazy aggregates, one +DuckDB aggregate query, or the Parquet footer. """ from __future__ import annotations @@ -21,6 +21,15 @@ _SAMPLE_FRAC = 0.10 +def _quote_identifier(name: str) -> str: + return '"' + name.replace('"', '""') + '"' + + +def is_duckdb_float(native_dtype: str) -> bool: + """True for DuckDB floating-point types, the only ones that can hold ``NaN``.""" + return native_dtype.upper() in ("FLOAT", "DOUBLE", "REAL", "FLOAT4", "FLOAT8") + + def _canonical_dtype(kind: str) -> str: """Map an arbitrary dtype string to freshdata's canonical buckets.""" k = kind.lower() @@ -50,6 +59,8 @@ class ColumnMetadata: is_numeric: bool = False is_string: bool = False sample_values: list[Any] = field(default_factory=list) + #: The backend's own type name (e.g. DuckDB ``DOUBLE``); empty when unknown. + native_dtype: str = "" @property def is_empty(self) -> bool: @@ -135,34 +146,40 @@ def from_polars_lazy(lf: Any) -> list[ColumnMetadata]: @staticmethod def from_duckdb(conn: Any, table_name: str) -> list[ColumnMetadata]: - """Scan a registered DuckDB table/view via ``SUMMARIZE`` (no Python scan).""" - require_duckdb() - summary = conn.execute(f"SUMMARIZE {table_name}").fetchall() - cols = [d[0] for d in conn.execute(f"SUMMARIZE {table_name}").description] - idx = {name: i for i, name in enumerate(cols)} + """Scan a registered DuckDB table/view with one aggregate query (no Python scan). - (n,) = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone() - n = int(n) + Null counts are exact, and float ``NaN`` counts as missing as it does in + pandas. ``SUMMARIZE`` is not used: its ``stddev_samp`` raises on + non-finite floats and it only reports a rounded null percentage. + """ + require_duckdb() + described = conn.execute(f"DESCRIBE {table_name}").fetchall() + columns = [(str(r[0]), str(r[1])) for r in described] + aggs = ["COUNT(*)"] + for name, native in columns: + col = _quote_identifier(name) + present = f"CASE WHEN NOT isnan({col}) THEN 1 END" if is_duckdb_float(native) else col + aggs.append(f"COUNT({present})") + aggs.append(f"approx_count_distinct({col})") + row = conn.execute(f"SELECT {', '.join(aggs)} FROM {table_name}").fetchone() + n = int(row[0]) out: list[ColumnMetadata] = [] - for r in summary: - name = r[idx["column_name"]] - dtype = str(r[idx["column_type"]]) - null_pct = r[idx.get("null_percentage", -1)] if "null_percentage" in idx else None - null_ratio = float(null_pct) / 100.0 if null_pct is not None else 0.0 - approx_unique = r[idx["approx_unique"]] if "approx_unique" in idx else -1 - canonical = _canonical_dtype(dtype) - non_null = int(round(n * (1.0 - null_ratio))) + for i, (name, native) in enumerate(columns): + non_null = int(row[1 + 2 * i]) + approx_unique = row[2 + 2 * i] + canonical = _canonical_dtype(native) out.append( ColumnMetadata( - name=str(name), + name=name, dtype_str=canonical, row_count=n, - null_ratio=null_ratio, + null_ratio=0.0 if n == 0 else 1.0 - non_null / n, non_null_count=non_null, n_unique=int(approx_unique) if approx_unique is not None else -1, is_numeric=canonical in ("int64", "float64"), is_string=canonical == "string", + native_dtype=native, ) ) return out diff --git a/src/freshdata/execution/_report.py b/src/freshdata/execution/_report.py index f6eca94..19f0ea0 100644 --- a/src/freshdata/execution/_report.py +++ b/src/freshdata/execution/_report.py @@ -78,6 +78,24 @@ def finalize_report(report: CleanReport, cleaned: Any, started: float) -> CleanR return report +def zero_column_frame(backend: str, output_format: str, report: CleanReport) -> Any: + """pandas result for a native run that dropped every column as empty. + + The pandas pipeline keeps the row count on a zero-column frame, so the native + backends return the same. Polars and Arrow frames cannot hold rows without + columns; for those output formats the difference is recorded on the report. + """ + import pandas as pd + + if output_format != "pandas": + report.record_backend_difference( + backend, "drop_empty_columns", + f"every column was empty; a zero-column {output_format} result cannot carry " + f"the row count ({report.rows_before} row(s) in pandas output)", + ) + return pd.DataFrame(index=pd.RangeIndex(report.rows_before)) + + def finalize_report_native(report: CleanReport, started: float) -> CleanReport: """Finalize a report whose result was returned as a native, un-materialized handle (a DuckDB relation or a Polars ``LazyFrame``). diff --git a/src/freshdata/execution/backends/_duckdb.py b/src/freshdata/execution/backends/_duckdb.py index 3a7d7cb..f4dbbb5 100644 --- a/src/freshdata/execution/backends/_duckdb.py +++ b/src/freshdata/execution/backends/_duckdb.py @@ -16,16 +16,18 @@ from __future__ import annotations import logging +import math import os import time import weakref from typing import TYPE_CHECKING, Any +from ..._util import PY_WHITESPACE from ...steps.duplicates import check_duplicate_ratio, report_detected_duplicates from .._base import ExecutionEngine from .._config import NATIVE_HANDLE_FORMATS, enforce_fallback_policy from .._lazy import has_duckdb, has_polars, require_duckdb -from .._metadata import MetadataScanner +from .._metadata import MetadataScanner, is_duckdb_float from .._native_steps import ( impute_defined_for, integer_safe_bounds, @@ -38,7 +40,7 @@ zscore_bounds, ) from .._plan import PlanGenerator -from .._report import finalize_report, finalize_report_native, init_report +from .._report import finalize_report, finalize_report_native, init_report, zero_column_frame from ._pandas import materialize_to_pandas if TYPE_CHECKING: # pragma: no cover - typing only @@ -51,7 +53,11 @@ log = logging.getLogger("freshdata.execution.duckdb") _TABLE = "freshdata_source" +#: Scan-order ordinal used by order-preserving deduplication. +_ROW_ID = "__freshdata_row__" _NATIVE_RELATION_CONNECTIONS: dict[int, Any] = {} +#: RE2 character class of every ``str.isspace`` character (RE2's ``\s`` is ASCII-only). +_WHITESPACE_CLASS = "[" + "".join(f"\\x{{{ord(c):04X}}}" for c in PY_WHITESPACE) + "]" def _q(name: str) -> str: @@ -65,9 +71,17 @@ def _lit(value: str) -> str: def _strip_sql(col_sql: str) -> str: - """Trim leading/trailing whitespace (matches Python ``str.strip`` semantics).""" - inner = f"regexp_replace(CAST({col_sql} AS VARCHAR), '^\\s+', '', 'g')" - return f"regexp_replace({inner}, '\\s+$', '', 'g')" + """Trim leading/trailing whitespace exactly like Python ``str.strip()``.""" + return ( + f"regexp_replace(CAST({col_sql} AS VARCHAR), " + f"'^{_WHITESPACE_CLASS}+|{_WHITESPACE_CLASS}+$', '', 'g')" + ) + + +def _number_sql(value: Any) -> str: + """SQL literal for a float; ``inf``/``nan`` need a cast (bare ``inf`` is a column name).""" + number = float(value) + return f"{number}" if math.isfinite(number) else f"CAST('{number}' AS DOUBLE)" def _release_native_relation_connection(key: int) -> None: @@ -125,9 +139,13 @@ def execute( report = init_report(meta, self._memory_before(source)) report.backend = "duckdb" relation = self._run_sql_pipeline( - conn, meta, plan, config, report, materialize=False + conn, meta, plan, config, report, materialize=False, + allow_zero_columns=not native, ) - if native: + if relation is None: + # Every column was dropped as empty; SQL cannot select zero columns. + cleaned = zero_column_frame("duckdb", engine_config.output_format, report) + elif native: cleaned = relation # The relation is tied to this connection; keep it open so the # caller can stream from it. Closing here would invalidate it. @@ -278,12 +296,18 @@ def _run_sql_pipeline( report: CleanReport, *, materialize: bool = True, + allow_zero_columns: bool = True, ) -> Any: + """Build the cleaning SQL; return a relation (or a fetched frame). + + Returns ``None`` (or, with *materialize*, a zero-column frame) when every + column was dropped as empty, since SQL cannot select zero columns. + """ from ...steps.strings import active_sentinels rename = plan.rename_map string_cols = {m.name for m in meta if m.is_string} - cur = f"SELECT * FROM {_TABLE}" + cur = self._source_sql(meta) if "column_names" in plan.stages and rename: self._record_rename(rename, report) @@ -293,18 +317,26 @@ def _run_sql_pipeline( if "clean_strings" in plan.stages and string_cols: self._record_string_counts(conn, meta, rename, config, active_sentinels(config), report) - cur = self._project_clean(meta, rename, config, plan, active_sentinels(config)) + cur = self._project_clean(meta, rename, config, plan, active_sentinels(config), cur) # current column names after rename cols = [str(rename.get(m.name, m.name)) for m in meta] rows_before = report.rows_before - if "drop_empty_columns" in plan.stages and rows_before > 0: - cur, cols = self._drop_empty_columns(conn, cur, cols, report) + if "drop_empty_columns" in plan.stages and rows_before > 0 and cols: + cur, cols = self._drop_empty_columns( + conn, cur, cols, report, allow_zero_columns=allow_zero_columns + ) + if not cols: + # Like the pandas pipeline, a zero-column frame keeps its rows and + # later row stages have nothing to act on. + if materialize: + return zero_column_frame("duckdb", "pandas", report) + return None if "drop_empty_rows" in plan.stages and rows_before > 0 and cols: cur = self._drop_empty_rows(conn, cur, cols, report) - if "drop_duplicates" in plan.stages: - cur = self._drop_duplicates(conn, cur, config, report) + if "drop_duplicates" in plan.stages and cols: + cur = self._drop_duplicates(conn, cur, cols, config, report) numeric_map = {str(rename.get(m.name, m.name)): m for m in meta} if "impute" in plan.stages and cols: @@ -317,6 +349,22 @@ def _run_sql_pipeline( return conn.sql(cur) return conn.execute(cur).fetchdf() + @staticmethod + def _source_sql(meta: list) -> str: + """The registered source, with float ``NaN`` read as ``NULL``. + + pandas treats ``NaN`` as missing. DuckDB already reads a pandas ``NaN`` + as ``NULL`` but keeps real ``NaN`` from Arrow/Parquet/Polars sources, + which would otherwise count as a value in every later stage. + """ + floats = [m.name for m in meta if is_duckdb_float(m.native_dtype)] + if not floats: + return f"SELECT * FROM {_TABLE}" + replaced = ", ".join( + f"CASE WHEN isnan({_q(c)}) THEN NULL ELSE {_q(c)} END AS {_q(c)}" for c in floats + ) + return f"SELECT * REPLACE ({replaced}) FROM {_TABLE}" + def _record_rename(self, rename: dict, report: CleanReport) -> None: changes = list(rename.items()) preview = ", ".join(f"{o!r}->{n!r}" for o, n in changes[:4]) @@ -326,7 +374,8 @@ def _record_rename(self, rename: dict, report: CleanReport) -> None: count=len(changes)) def _project_clean( - self, meta: list, rename: dict, config: CleanConfig, plan: Any, sentinels: frozenset + self, meta: list, rename: dict, config: CleanConfig, plan: Any, sentinels: frozenset, + source: str, ) -> str: do_strings = "clean_strings" in plan.stages sent_list = ", ".join(_lit(s) for s in sentinels) @@ -343,7 +392,7 @@ def _project_clean( else: expr = src pieces.append(f"{expr} AS {_q(target)}") - return f"SELECT {', '.join(pieces)} FROM {_TABLE}" + return f"SELECT {', '.join(pieces)} FROM ({source}) AS _src" def _record_string_counts( self, conn: Any, meta: list, rename: dict, config: CleanConfig, @@ -387,14 +436,23 @@ def _record_string_counts( column=name, count=n_sent) def _drop_empty_columns( - self, conn: Any, cur: str, cols: list[str], report: CleanReport + self, conn: Any, cur: str, cols: list[str], report: CleanReport, + *, allow_zero_columns: bool = True, ) -> tuple[str, list[str]]: + """Drop all-missing columns; ``kept == []`` means every column was dropped.""" counts = ", ".join(f"COUNT({_q(c)}) AS c{i}" for i, c in enumerate(cols)) row = conn.execute(f"SELECT {counts} FROM ({cur}) AS _s").fetchone() dropped = [c for i, c in enumerate(cols) if int(row[i]) == 0] if not dropped: return cur, cols kept = [c for c in cols if c not in dropped] + if not kept and not allow_zero_columns: + report.record_backend_difference( + "duckdb", "drop_empty_columns", + "every column is empty and a DuckDB relation cannot have zero columns; " + "the all-missing columns were kept in the returned relation", + ) + return cur, cols report.columns_dropped.extend(dropped) report.add( "drop_empty_columns", @@ -402,7 +460,9 @@ def _drop_empty_columns( + (" …" if len(dropped) > 6 else ""), count=len(dropped), ) - select_list = ", ".join(_q(c) for c in kept) if kept else "*" + if not kept: + return cur, kept + select_list = ", ".join(_q(c) for c in kept) return f"SELECT {select_list} FROM ({cur}) AS _s", kept def _drop_empty_rows(self, conn: Any, cur: str, cols: list[str], report: CleanReport) -> str: @@ -468,7 +528,7 @@ def add_stat(expr: str, i: int, kind: str) -> None: report.add("impute", f"skipped (could not compute {resolved} for {c})", column=c) pieces.append(_q(c)) continue - lit = f"{float(value)}" if is_numeric else _lit(str(value)) + lit = _number_sql(value) if is_numeric else _lit(str(value)) pieces.append(f"COALESCE({_q(c)}, {lit}) AS {_q(c)}") shown = f"{value:.6g}" if isinstance(value, float) else repr(value) report.add("impute", @@ -499,20 +559,26 @@ def add_stat(expr: str, c: str, kind: str) -> None: stat_parts.append(f"{expr} AS x{len(stat_parts)}") for c in numeric: + # Fences come from finite values only, like the pandas reference + # (steps.outliers.drop_infinite); stddev_samp also raises on inf. + only_finite = ( + f" FILTER (WHERE isfinite({_q(c)}))" + if is_duckdb_float(numeric_map[c].native_dtype) else "" + ) if method == "iqr": - add_stat(f"quantile_cont({_q(c)}, 0.25)", c, "q1") - add_stat(f"quantile_cont({_q(c)}, 0.75)", c, "q3") + add_stat(f"quantile_cont({_q(c)}, 0.25){only_finite}", c, "q1") + add_stat(f"quantile_cont({_q(c)}, 0.75){only_finite}", c, "q3") else: - add_stat(f"AVG({_q(c)})", c, "m") - add_stat(f"stddev_samp({_q(c)})", c, "s") + add_stat(f"AVG({_q(c)}){only_finite}", c, "m") + add_stat(f"stddev_samp({_q(c)}){only_finite}", c, "s") if clip: # Skew-aware capping aggregates (see log_widened_bounds). positive_log = f"CASE WHEN {_q(c)} > 0 THEN ln({_q(c)}) END" - add_stat(f"skewness({_q(c)})", c, "sk") - add_stat(f"MIN({_q(c)})", c, "mn") - add_stat(f"COUNT({_q(c)})", c, "cnt") - add_stat(f"quantile_cont({positive_log}, 0.25)", c, "lq1") - add_stat(f"quantile_cont({positive_log}, 0.75)", c, "lq3") + add_stat(f"skewness({_q(c)}){only_finite}", c, "sk") + add_stat(f"MIN({_q(c)}){only_finite}", c, "mn") + add_stat(f"COUNT({_q(c)}){only_finite}", c, "cnt") + add_stat(f"quantile_cont({positive_log}, 0.25){only_finite}", c, "lq1") + add_stat(f"quantile_cont({positive_log}, 0.75){only_finite}", c, "lq3") srow = conn.execute(f"SELECT {', '.join(stat_parts)} FROM ({cur}) AS _s").fetchone() def stat(c: str, kind: str) -> float | None: @@ -621,7 +687,7 @@ def _unique_flag(existing: list[str], base: str) -> str: return name def _drop_duplicates( - self, conn: Any, cur: str, config: CleanConfig, report: CleanReport + self, conn: Any, cur: str, cols: list[str], config: CleanConfig, report: CleanReport ) -> str: (n_before,) = conn.execute(f"SELECT COUNT(*) FROM ({cur}) AS _s").fetchone() n_before = int(n_before) @@ -631,16 +697,17 @@ def _drop_duplicates( # Detection-only default: count duplicates, report, keep every row. subset = (list(config.duplicate_subset) if config.duplicate_subset is not None else None) - cols = ", ".join(_q(c) for c in subset) if subset else "*" + distinct = ", ".join(_q(c) for c in subset) if subset else "*" (n_unique,) = conn.execute( - f"SELECT COUNT(*) FROM (SELECT DISTINCT {cols} FROM ({cur}) AS _s) AS _u" + f"SELECT COUNT(*) FROM (SELECT DISTINCT {distinct} FROM ({cur}) AS _s) AS _u" ).fetchone() report_detected_duplicates( n_before - int(n_unique), n_before, config, report, subset=subset ) return cur - deduped = f"SELECT DISTINCT * FROM ({cur}) AS _s" - (n_after,) = conn.execute(f"SELECT COUNT(*) FROM ({deduped}) AS _d").fetchone() + (n_after,) = conn.execute( + f"SELECT COUNT(*) FROM (SELECT DISTINCT * FROM ({cur}) AS _s) AS _d" + ).fetchone() n_dup = n_before - int(n_after) if n_dup <= 0: return cur @@ -660,4 +727,23 @@ def _drop_duplicates( f"{pct:.1f}% of rows were duplicates " f"(> {100 * config.duplicate_threshold:.0f}%); confirm they are not legitimate" ) - return deduped + return self._dedup_in_order(cur, cols, config.duplicate_keep) + + @staticmethod + def _dedup_in_order(cur: str, cols: list[str], keep: str) -> str: + """Full-row dedup that keeps the pandas row choice and row order. + + ``SELECT DISTINCT`` returns rows in arbitrary order and cannot honour + ``duplicate_keep``. A scan-order ordinal picks the first (or last) + occurrence of each row, and ordering by it restores the input order. + ``PARTITION BY`` groups ``NULL``s together, as ``DataFrame.duplicated`` does. + """ + rid = _q(_ROW_ID) + direction = "DESC" if keep == "last" else "ASC" + partition = ", ".join(_q(c) for c in cols) + numbered = f"SELECT *, row_number() OVER () AS {rid} FROM ({cur}) AS _s" + return ( + f"SELECT * EXCLUDE ({rid}) FROM ({numbered}) AS _n " + f"QUALIFY row_number() OVER (PARTITION BY {partition} ORDER BY {rid} {direction}) = 1 " + f"ORDER BY {rid}" + ) diff --git a/src/freshdata/execution/backends/_polars.py b/src/freshdata/execution/backends/_polars.py index e14043e..75dc716 100644 --- a/src/freshdata/execution/backends/_polars.py +++ b/src/freshdata/execution/backends/_polars.py @@ -17,6 +17,7 @@ import time from typing import TYPE_CHECKING, Any +from ..._util import PY_WHITESPACE from ...steps.duplicates import check_duplicate_ratio, report_detected_duplicates from .._base import ExecutionEngine from .._config import enforce_fallback_policy @@ -34,7 +35,7 @@ zscore_bounds, ) from .._plan import NativePlan, PlanGenerator -from .._report import finalize_report, finalize_report_native, init_report +from .._report import finalize_report, finalize_report_native, init_report, zero_column_frame from ._pandas import materialize_to_pandas if TYPE_CHECKING: # pragma: no cover - typing only @@ -127,6 +128,10 @@ def execute( ) return lf, report cleaned = self._collect(lf, engine_config, pl) + if cleaned.width == 0 and report.columns_dropped and report.rows_before > 0: + # Every column was dropped as empty; a polars frame without columns has + # height 0, while the pandas reference keeps the rows. + cleaned = zero_column_frame("polars", engine_config.output_format, report) finalize_report(report, cleaned, started) return cleaned, report @@ -359,7 +364,7 @@ def _stage_clean_strings( count_exprs: list[Any] = [] for c in string_cols: col = pl.col(c) - stripped = col.str.strip_chars() + stripped = col.str.strip_chars(PY_WHITESPACE) base = stripped if config.strip_whitespace else col if config.strip_whitespace: count_exprs.append( @@ -376,7 +381,7 @@ def _stage_clean_strings( transforms: list[Any] = [] for c in string_cols: col = pl.col(c) - stripped = col.str.strip_chars() + stripped = col.str.strip_chars(PY_WHITESPACE) base = stripped if config.strip_whitespace else col n_strip = int(counts.get(f"__strip__{c}", 0) or 0) n_sent = int(counts.get(f"__sent__{c}", 0) or 0) diff --git a/tests/test_execution/test_engine_parity_edge_cases.py b/tests/test_execution/test_engine_parity_edge_cases.py new file mode 100644 index 0000000..c9c76b8 --- /dev/null +++ b/tests/test_execution/test_engine_parity_edge_cases.py @@ -0,0 +1,171 @@ +"""Native engines match the pandas reference on edge-case inputs. + +Covers non-finite floats and exact missing counts (#199), all-empty columns +(#201), order/keep of full-row dedup (#202) and Unicode whitespace (#204). +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +import freshdata as fd +from freshdata._util import PY_WHITESPACE +from freshdata.config import CleanConfig +from freshdata.execution import EngineConfig + +pytest.importorskip("duckdb") +pytest.importorskip("polars") +pa = pytest.importorskip("pyarrow") + +ENGINES = ("pandas", "polars", "duckdb") + + +def _config(**overrides) -> CleanConfig: + return CleanConfig(strategy="conservative", fix_dtypes=False, verbose=False, **overrides) + + +def _actions(report): + return [(a.step, a.count) for a in report.actions] + + +# -- #204 whitespace --------------------------------------------------------- + + +def test_py_whitespace_is_exactly_the_str_isspace_set(): + python_whitespace = "".join(c for c in map(chr, range(0x110000)) if c.isspace()) + assert python_whitespace == PY_WHITESPACE + + +@pytest.mark.parametrize("engine", ENGINES) +def test_strip_matches_python_str_strip(engine): + values = [f"{ch}x{i}{ch}" for i, ch in enumerate(PY_WHITESPACE)] + values += ["\xa0N/A　", "\x1f\x1c"] # a padded sentinel, whitespace-only text + out = fd.clean(pd.DataFrame({"s": values}), config=_config(drop_empty_rows=False), + engine=engine) + expected = [v.strip() for v in values[:-2]] + assert out["s"].tolist()[:-2] == expected + assert out["s"].iloc[-2:].isna().all() # "N/A" and "" are sentinels once stripped + + +# -- #199 non-finite floats and exact missing counts --------------------------- + + +@pytest.mark.parametrize("engine", ENGINES) +def test_infinite_values_do_not_crash(engine): + df = pd.DataFrame({"x": [1.0, np.inf, 2.0, -np.inf, 3.0]}) + out = fd.clean(df, config=_config(), engine=engine) + assert len(out) == 5 + + +def test_duckdb_mean_impute_with_infinite_values_matches_pandas(): + df = pd.DataFrame({"x": [1.0, np.inf, np.nan, 2.0], "k": [1, 2, 3, 4]}) + config = _config(impute="mean") + ref = fd.clean(df, config=config, engine="pandas") + out = fd.clean(df, config=config, engine="duckdb") + assert out["x"].tolist() == ref["x"].tolist() == [1.0, np.inf, np.inf, 2.0] + + +def test_duckdb_clip_uses_finite_fences(): + df = pd.DataFrame({"x": [1.0, 2.0, 3.0, 4.0, np.inf, np.inf, np.inf, np.inf]}) + config = _config(outliers="clip", outlier_method="iqr") + ref = fd.clean(df, config=config, engine="pandas") + out = fd.clean(df, config=config, engine="duckdb") + assert np.isfinite(out["x"]).all() + assert out["x"].tolist() == ref["x"].tolist() + + +def test_duckdb_arrow_nan_counts_as_missing(): + table = pa.table({ + "x": pa.array([1.0, float("nan"), None, 4.0], from_pandas=False), + "y": pa.array([float("nan")] * 4, from_pandas=False), + }) + ref_out, ref = fd.clean(table.to_pandas(), config=_config(), engine="pandas", + return_report=True) + out, report = fd.clean(table, config=_config(), engine="duckdb", return_report=True) + assert report.missing_before == ref.missing_before == 6 + assert list(out.columns) == list(ref_out.columns) == ["x"] + assert report.rows_after == ref.rows_after == 2 + assert _actions(report) == _actions(ref) + + +@pytest.mark.parametrize("engine", ENGINES) +def test_missing_before_is_exact(engine): + rng = np.random.default_rng(0) + x = rng.normal(size=100_003) + x[rng.random(100_003) < 0.2663] = np.nan + _, report = fd.clean(pd.DataFrame({"x": x}), config=_config(), engine=engine, + return_report=True) + assert report.missing_before == int(np.isnan(x).sum()) + + +# -- #201 all-empty columns ---------------------------------------------------- + + +def _all_empty() -> pd.DataFrame: + return pd.DataFrame({"a": [np.nan, np.nan], "b": [None, None]}) + + +@pytest.mark.parametrize("engine", ["polars", "duckdb"]) +def test_all_empty_columns_match_pandas(engine): + ref_out, ref = fd.clean(_all_empty(), config=_config(), engine="pandas", return_report=True) + out, report = fd.clean(_all_empty(), config=_config(), engine=engine, return_report=True) + assert list(out.columns) == list(ref_out.columns) == [] + assert len(out) == len(ref_out) == 2 + assert (report.rows_after, report.cols_after) == (ref.rows_after, ref.cols_after) == (2, 0) + assert _actions(report) == _actions(ref) + assert report.backend_differences == [] + + +@pytest.mark.parametrize("engine", ["polars", "duckdb"]) +@pytest.mark.parametrize("output_format", ["polars", "arrow"]) +def test_all_empty_columns_non_pandas_output_is_disclosed(engine, output_format): + out, report = fd.clean(_all_empty(), config=_config(), engine=engine, + output_format=output_format, return_report=True) + assert len(out.columns if output_format == "polars" else out.column_names) == 0 + assert any(d["step"] == "drop_empty_columns" for d in report.backend_differences) + + +def test_duckdb_native_handle_keeps_all_empty_columns_and_says_so(): + rel, report = fd.clean(_all_empty(), config=_config(), engine="duckdb", + output_format="duckdb", return_report=True) + assert rel.columns == ["a", "b"] + assert not any(a.step == "drop_empty_columns" for a in report.actions) + assert any(d["step"] == "drop_empty_columns" for d in report.backend_differences) + + +# -- #202 dedup order and keep ------------------------------------------------- + + +def _dup_frame(n: int) -> pd.DataFrame: + if n <= 5: + return pd.DataFrame({"a": [3, 1, 3, 2, 1]}) + rng = np.random.default_rng(1) + return pd.DataFrame({ + "a": rng.integers(0, n // 10, n), + "b": rng.choice(["x", "y", None], n), + "c": rng.choice([1.5, np.nan], n), + }) + + +@pytest.mark.parametrize("keep", ["first", "last"]) +@pytest.mark.parametrize("n", [5, 20_000]) +def test_duckdb_dedup_matches_pandas_order_and_keep(keep, n): + df = _dup_frame(n) + config = _config(drop_duplicates=True, duplicate_keep=keep) + ref = fd.clean(df, config=config, engine="pandas").reset_index(drop=True) + engine_config = EngineConfig(engine="duckdb", duckdb_threads=4) + out = fd.clean(df, config=config, engine_config=engine_config).reset_index(drop=True) + pd.testing.assert_frame_equal(out, ref, check_dtype=False) + + +def test_duckdb_dedup_keeps_order_for_parquet_sources(tmp_path): + df = _dup_frame(20_000) + path = str(tmp_path / "dups.parquet") + df.to_parquet(path, index=False) + config = _config(drop_duplicates=True, duplicate_keep="last") + ref = fd.clean(df, config=config, engine="pandas").reset_index(drop=True) + engine_config = EngineConfig(engine="duckdb", duckdb_threads=4) + out = fd.clean(path, config=config, engine_config=engine_config).reset_index(drop=True) + pd.testing.assert_frame_equal(out, ref, check_dtype=False) diff --git a/tests/test_execution/test_metadata_scanner.py b/tests/test_execution/test_metadata_scanner.py index e464ed3..2d8af9c 100644 --- a/tests/test_execution/test_metadata_scanner.py +++ b/tests/test_execution/test_metadata_scanner.py @@ -60,3 +60,20 @@ def test_from_duckdb(tmp_path): assert meta["b"].is_empty finally: conn.close() + + +def test_from_duckdb_exact_counts_with_non_finite_floats(): + duckdb = pytest.importorskip("duckdb") + conn = duckdb.connect() + try: + conn.execute( + "CREATE TABLE t AS SELECT * FROM (VALUES " + "(1.0::DOUBLE, 1), ('inf'::DOUBLE, NULL), ('nan'::DOUBLE, 3), (NULL, 4)) v(x, y)" + ) + meta = {m.name: m for m in MetadataScanner.from_duckdb(conn, "t")} + assert meta["x"].row_count == 4 + assert meta["x"].non_null_count == 2 # NaN is missing (as in pandas); inf is a value + assert meta["x"].native_dtype == "DOUBLE" + assert meta["y"].non_null_count == 3 + finally: + conn.close()