fix(spark): atomic renames, ordered keep-aware dedup, NaN/inf handling, exact numeric type names

docs/backends.md

@@ -159,6 +159,15 @@ Notes:
   `approxQuantile`. Such divergences are recorded in `report.backend_differences`.
 - **Outlier counts** match the pandas reference where the quantile statistics
   match (Polars/DuckDB linear interpolation); Spark may flag a different count.
+  pandas, Polars, DuckDB and Spark compute the fences from finite values only;
+  `±inf` is still tested against them.
+- **Float `NaN`** is read as missing by Polars, DuckDB and Spark, as in pandas
+  (Spark keeps `NaN` as a value in float/double columns, so it is converted to
+  null on ingestion).
+- **Spark full-row dedup** keeps the first/last occurrence and the surviving
+  rows' order, like pandas. Row order is the input DataFrame's partition order,
+  which is the file order for a single read; a frame that was already shuffled
+  has no stable order to preserve.
 - A non-default pandas index (e.g. a `DatetimeIndex`) forces a pandas fallback,
   since native frames carry no index.
 - FreshCore v1 is a cleaning-first native engine, not an out-of-core engine. It

docs/fallback-matrix.md

@@ -17,7 +17,7 @@ the input checks in `pandas_ingest_fallback_reason()`
 | `strategy="balanced"` / `"aggressive"` (default) | pandas | pandas | pandas | pandas | data-dependent decision engine; porting it per backend would fork its accuracy behaviour |
 | column rename / whitespace / sentinels | native | native | native | native | — |
 | empty row/column removal | native | native | native | native | — |
-| full-row dedup (`keep="first"/"last"`) | native | native | native | native | streaming polars dedup drops row order (disclosed); `streaming_dedup=False` restores it |
+| full-row dedup (`keep="first"/"last"`) | native | native | native | native | streaming polars dedup drops row order (disclosed); `streaming_dedup=False` restores it. Spark keeps the first/last row in the input DataFrame's partition order |
 | **subset dedup** (`duplicate_subset=`) | **native** | pandas | pandas | pandas | keep semantics are order-sensitive; Polars reproduces them via order-preserving `unique` (eager, not streaming — disclosed) |
 | dedup `keep="drop"/"aggregate"` | pandas | pandas | pandas | pandas | group-wise resolution isn't expressed natively yet |
 | global impute mean/median/mode | native | native | native | native | — |

src/freshdata/execution/backends/_spark.py

@@ -43,6 +43,22 @@
 
 log = logging.getLogger("freshdata.execution.spark")
 
+#: Spark ``simpleString`` names (``DataFrame.dtypes``) of the numeric types.
+#: Matched exactly: a prefix test would take ``interval day to second`` as ``int``.
+_SPARK_INTEGER_TYPES = frozenset({"tinyint", "smallint", "int", "bigint"})
+_SPARK_FLOAT_TYPES = frozenset({"float", "double"})
+_SPARK_NUMERIC_TYPES = _SPARK_INTEGER_TYPES | _SPARK_FLOAT_TYPES | {"decimal"}
+
+
+def _spark_type_name(dtype: str) -> str:
+    """Base name of a Spark dtype string, without parameters (``decimal(10,2)``)."""
+    return dtype.split("(", 1)[0].strip().lower()
+
+
+def _renamed_columns(columns: list[str], rename_map: dict[str, str]) -> list[str]:
+    """Every column's new name, applied at once and by position."""
+    return [rename_map.get(c, c) for c in columns]
+
 
 def _is_spark_frame(source: Any) -> bool:
     if not has_pyspark():
@@ -116,6 +132,7 @@ def execute(
 
         started = time.perf_counter()
         report = CleanReport(backend="spark")
+        sdf = self._nan_as_null(sdf)
         self._init_before(sdf, report)
         sdf = self._apply_native(sdf, plan, config, report)
         self._finalize(sdf, report, started)
@@ -126,6 +143,26 @@ def _fallback(self, source: Any, config: CleanConfig) -> tuple[Any, CleanReport]
 
         return run_pipeline(materialize_to_pandas(source), config)
 
+    def _nan_as_null(self, sdf: Any) -> Any:
+        """*sdf* with float ``NaN`` read as null.
+
+        pandas treats ``NaN`` as missing, but Spark keeps it as a value in
+        float/double columns (Spark-written Parquet, CSV ``NaN`` literals,
+        non-Arrow ``createDataFrame``), so every later null test would miss it.
+        """
+        from pyspark.sql import functions as F
+
+        floats = {
+            name for name, dtype in sdf.dtypes if _spark_type_name(dtype) in _SPARK_FLOAT_TYPES
+        }
+        if not floats:
+            return sdf
+        return sdf.select(*[
+            F.when(F.isnan(F.col(c)), None).otherwise(F.col(c)).alias(c)
+            if c in floats else F.col(c)
+            for c in sdf.columns
+        ])
+
     # -- report bookkeeping -------------------------------------------------
 
     def _null_count_total(self, sdf: Any) -> int:
@@ -158,15 +195,18 @@ def _string_columns(self, sdf: Any) -> list[str]:
         return [name for name, dtype in sdf.dtypes if dtype == "string"]
 
     def _numeric_columns(self, sdf: Any) -> list[str]:
-        numeric = ("tinyint", "smallint", "int", "bigint", "float", "double", "decimal")
         return [
             name for name, dtype in sdf.dtypes
-            if dtype.startswith(numeric) and dtype != "boolean"
+            if _spark_type_name(dtype) in _SPARK_NUMERIC_TYPES
         ]
 
     def _is_integer_column(self, sdf: Any, name: str) -> bool:
         dtype = dict(sdf.dtypes)[name]
-        return dtype.startswith(("tinyint", "smallint", "int", "bigint"))
+        return _spark_type_name(dtype) in _SPARK_INTEGER_TYPES
+
+    def _is_float_column(self, sdf: Any, name: str) -> bool:
+        dtype = dict(sdf.dtypes)[name]
+        return _spark_type_name(dtype) in _SPARK_FLOAT_TYPES
 
     def _apply_native(
         self, sdf: Any, plan: NativePlan, config: CleanConfig, report: CleanReport
@@ -200,9 +240,10 @@ def _stage_rename(self, sdf: Any, plan: NativePlan, report: CleanReport) -> Any:
             preview += f", … (+{len(changes) - 4} more)"
         report.add("column_names", f"renamed {len(changes)} column(s): {preview}",
                    count=len(changes))
-        for old, new in changes:
-            sdf = sdf.withColumnRenamed(old, new)
-        return sdf
+        # One positional projection: chained withColumnRenamed calls rename
+        # every (case-insensitive) match, so a map like {"a b": "a_b",
+        # "a_b": "a_b_2"} would rename both columns in the second call.
+        return sdf.toDF(*_renamed_columns(list(sdf.columns), plan.rename_map))
 
     def _stage_clean_strings(
         self, sdf: Any, config: CleanConfig, report: CleanReport
@@ -293,8 +334,7 @@ def _stage_drop_duplicates(
         n_before = sdf.count()
         if n_before < 1:
             return sdf
-        deduped = sdf.dropDuplicates()
-        n_after = deduped.count()
+        n_after = sdf.dropDuplicates().count()
         n_dup = n_before - n_after
         if not config.drop_duplicates:
             # Detection-only default: count duplicates, report, keep every row.
@@ -318,7 +358,33 @@ def _stage_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(sdf, config.duplicate_keep)
+
+    def _dedup_in_order(self, sdf: Any, keep: str) -> Any:
+        """Full-row dedup that keeps the pandas row choice and row order.
+
+        ``dropDuplicates`` is a shuffle aggregate: surviving rows come back in
+        arbitrary order and ``keep`` is ignored. An ordinal taken before any
+        shuffle picks the first (or last) occurrence of each row, and sorting
+        by it restores the input order. The ordinal follows the input
+        DataFrame's partition order, which is the row order for a single
+        source read. ``partitionBy`` groups nulls together, as pandas does.
+        """
+        from pyspark.sql import Window
+        from pyspark.sql import functions as F
+
+        columns = list(sdf.columns)
+        ordinal = self._unique_flag(sdf, "__fd_row_ordinal__")
+        rank = self._unique_flag(sdf, "__fd_row_rank__")
+        order = F.col(ordinal).desc() if keep == "last" else F.col(ordinal).asc()
+        window = Window.partitionBy(*[F.col(c) for c in columns]).orderBy(order)
+        return (
+            sdf.withColumn(ordinal, F.monotonically_increasing_id())
+            .withColumn(rank, F.row_number().over(window))
+            .filter(F.col(rank) == 1)
+            .orderBy(F.col(ordinal).asc())
+            .drop(ordinal, rank)
+        )
 
     def _stage_impute(self, sdf: Any, config: CleanConfig, report: CleanReport) -> Any:
         from pyspark.sql import functions as F
@@ -418,6 +484,12 @@ def _outlier_bounds(
     ) -> tuple[float, float] | None:
         from pyspark.sql import functions as F
 
+        if self._is_float_column(sdf, name):
+            # Fences come from finite values only, like the pandas reference
+            # (steps.outliers.drop_infinite); inf is still tested against them.
+            sdf = sdf.filter(
+                ~F.isnan(F.col(name)) & (F.abs(F.col(name)) != F.lit(float("inf")))
+            )
         if method == "iqr":
             q = sdf.approxQuantile(name, [0.25, 0.75], 0.0)
             if len(q) < 2: