Incremental persistence: single MERGE per fact table + fix stale-row & ordering bugs

docs/impulse/docs/references/report/index.md

@@ -91,7 +91,7 @@ or pass `is_incremental=True` at call time.
 1. Compare every event and aggregation against its stored `definition_hash` in the gold dimension table. Classify each as **changed** (hash differs, or it's brand new) or **unchanged** (hash matches).
 2. For unchanged definitions, process only the containers that are new or have newer silver data than gold. Skip the rest.
 3. For changed definitions, reprocess all containers that match the report's filters.
-4. Persist via Delta `MERGE` on natural keys for unchanged definitions; replace atomically via `replaceWhere` on `visual_id`, `event_id`, or `channel_id` for changed ones.
+4. Persist each fact table with a single Delta `MERGE` on its natural keys. The changed rows (all containers) and unchanged rows (reprocessed containers) are written together, and stale rows within the reprocessed scope are deleted in the same transaction (see [Operational notes](#operational-notes)).
 
 #### Mode resolution
 
@@ -128,6 +128,7 @@ Renaming an aggregation, tweaking the description, or swapping `channel_name` or
 
 #### Operational notes
 
-- A single run can be partly incremental: one event is changed (full reprocess), another is unchanged (upserted containers only), a newly added aggregation is brand new (also full reprocess). Each entity walks its own path.
-- `replaceWhere` is atomic per fact table. When a definition changes, all rows for that `visual_id`, `event_id`, or `channel_id` get deleted and rewritten in one transaction. No intermediate inconsistent state, but there is a brief rewrite window.
-- `MERGE` keeps existing rows that don't conflict, so unchanged definitions accumulate rows for new containers without rewriting the old ones.
+- A single run can be partly incremental: one event is changed (full reprocess), another is unchanged (upserted containers only), a newly added aggregation is brand new (also full reprocess). Each entity walks its own path, but all of an entity's rows land in **one** `MERGE` per fact table (entity types that share a fact table are combined), so there is no intermediate inconsistent state.
+- The `MERGE` updates matched rows and inserts new ones, and also **deletes stale rows** (`whenNotMatchedBySourceDelete`) — but only within a bounded scope: the updated containers and the changed-definition entity ids. A modified container that now produces *fewer* rows (e.g. an event that fires fewer times, or an event-scoped statistic with fewer instances) has its surplus rows removed rather than left behind as orphans.
+- Rows outside that scope are never touched: containers that weren't reprocessed, and entities whose definition didn't change, keep their existing gold rows untouched. New containers carry no delete scope (they have no prior gold rows) — their rows are simply inserted.
+- The `MERGE` runs with schema evolution, so a fact-table schema change across releases doesn't break an incremental run: a newly added column is added to the gold table (existing rows get NULL). A column dropped from the schema is *retained* in gold (new rows get NULL for it) — Delta doesn't drop columns via MERGE, so a genuine column removal or an incompatible type change still needs a full (non-incremental) rebuild.

src/impulse_reporting/core/report.py

@@ -636,6 +636,8 @@ def _persist_incremental(
         None
         """
         changed_channel_ids = changed_channel_ids or {}
+        has_processed_containers = getattr(self, "_has_processed_containers", False)
+        updated_container_ids = getattr(self, "_updated_container_ids", [])
         storage_factory = WriterFactory(self.sink)
         transformer = ReportEntityTransformer()
 
@@ -653,6 +655,8 @@ def _transform(df, schema):
             id_column="visual_id",
             merge_keys=self._get_aggregation_merge_keys,
             changed_ids=changed_aggregation_ids,
+            has_processed_containers=has_processed_containers,
+            updated_container_ids=updated_container_ids,
         )
         persist_dimensions_incremental(
             self.aggregation_metadata_dfs,
@@ -673,6 +677,8 @@ def _transform(df, schema):
             id_column="event_id",
             merge_keys=["container_id", "event_id", "event_instance_id"],
             changed_ids=changed_event_ids,
+            has_processed_containers=has_processed_containers,
+            updated_container_ids=updated_container_ids,
         )
         persist_dimensions_incremental(
             self.event_metadata_dfs,
@@ -682,7 +688,31 @@ def _transform(df, schema):
             merge_keys=["event_id"],
         )
 
-        # Persist measurement dimension (upsert by container_id)
+        # Persist calculated channel facts + dimensions
+        persist_facts_incremental(
+            self.calculated_channel_dfs,
+            ChannelType,
+            self.sink,
+            _transform,
+            id_column="channel_id",
+            merge_keys=["container_id", "channel_id", "tstart"],
+            changed_ids=changed_channel_ids,
+            has_processed_containers=has_processed_containers,
+            updated_container_ids=updated_container_ids,
+        )
+        persist_dimensions_incremental(
+            self.calculated_channel_metadata_dfs,
+            ChannelType,
+            self.sink,
+            _transform,
+            merge_keys=["channel_id"],
+        )
+
+        # Persist the measurement dimension LAST (as ``_persist_full`` does). It
+        # holds the gold timestamp that container-update detection compares
+        # against; the fact solves above are lazy, so writing it earlier would
+        # bump that timestamp before they materialize and drop the containers
+        # being reprocessed.
         if self.container_dimension_df:
             writer = storage_factory.create_container_dimension_writer()
             uri = writer.get_output_uri()
@@ -709,24 +739,6 @@ def _transform(df, schema):
                 ],
             )
 
-        # Persist calculated channel facts + dimensions
-        persist_facts_incremental(
-            self.calculated_channel_dfs,
-            ChannelType,
-            self.sink,
-            _transform,
-            id_column="channel_id",
-            merge_keys=["container_id", "channel_id", "tstart"],
-            changed_ids=changed_channel_ids,
-        )
-        persist_dimensions_incremental(
-            self.calculated_channel_metadata_dfs,
-            ChannelType,
-            self.sink,
-            _transform,
-            merge_keys=["channel_id"],
-        )
-
     def _transform_for_persistence(
         self,
         df: DataFrame,
@@ -868,11 +880,24 @@ def determine_report(self, is_incremental: bool = None):
         # Determine processing mode: config overrides signature, gold must exist
         self._is_incremental = self._resolve_is_incremental(is_incremental)
 
-        # Detect containers to process (incremental mode only)
+        # Detect containers to process (incremental mode only): new + updated.
         pre_filtered_containers_df = None
         if self._is_incremental:
             pre_filtered_containers_df = self._detect_upserted_containers()
 
+        # Two signals for persistence:
+        # - has_processed_containers (new + updated): gates whether a fact table is
+        #   written (new containers must be inserted). Only a bool is needed, so
+        #   probe emptiness with isEmpty() rather than collecting the whole id list.
+        # - updated container ids: scopes the delete-by-source, since only
+        #   containers that already have gold rows can have stale rows to prune.
+        self._has_processed_containers = (
+            pre_filtered_containers_df is not None and not pre_filtered_containers_df.isEmpty()
+        )
+        self._updated_container_ids = self._collect_container_ids(
+            self._detect_updated_containers() if self._is_incremental else None
+        )
+
         hash_comparator = DefinitionHashComparator(self.spark)
 
         # Group events and aggregations by type
@@ -1110,22 +1135,80 @@ def _detect_upserted_containers(self) -> DataFrame | None:
             DataFrame containing containers to process, or None if gold table
             doesn't exist (indicating full processing is needed).
         """
+        args = self._container_detection_args()
+        if args is None:
+            return None
+        detector, silver_containers, measurement_dim_table, silver_col, gold_col = args
+        return detector.detect_upserted_containers(
+            silver_containers,
+            measurement_dim_table,
+            silver_last_modified_col=silver_col,
+            gold_last_modified_col=gold_col,
+        )
+
+    def _detect_updated_containers(self) -> DataFrame | None:
+        """Detect only UPDATED containers (present in gold, newer silver timestamp).
+
+        Excludes new containers — see
+        ``ContainerUpsertDetector.detect_updated_containers``. Used to scope the
+        incremental delete-by-source. Returns None in sinkless mode or when the
+        gold table doesn't exist.
+
+        Returns
+        -------
+        DataFrame | None
+            Updated containers, or None.
+        """
+        args = self._container_detection_args()
+        if args is None:
+            return None
+        detector, silver_containers, measurement_dim_table, silver_col, gold_col = args
+        return detector.detect_updated_containers(
+            silver_containers,
+            measurement_dim_table,
+            silver_last_modified_col=silver_col,
+            gold_last_modified_col=gold_col,
+        )
+
+    def _container_detection_args(self):
+        """Shared inputs for container detection, or None in sinkless mode.
+
+        Returns
+        -------
+        tuple | None
+            ``(detector, silver_containers_df, measurement_dim_table, silver_col,
+            gold_col)`` — the freshness column names come from the incremental
+            config (default ``"last_modified"``). None when no sink is configured.
+        """
         if not self._has_sink:
             return None
         detector = ContainerUpsertDetector(self.spark)
         silver_containers = self.db.container_metrics(self.spark)
         measurement_dim_table = self.sink.config.get_output_uri_measurement_dimensions_table()
 
-        # Retrieve configurable column names (default: "last_modified")
         silver_col = "last_modified"
         gold_col = "last_modified"
         if hasattr(self.config, "incremental") and self.config.incremental is not None:
             silver_col = self.config.incremental.silver_last_modified_column
             gold_col = self.config.incremental.gold_last_modified_column
 
-        return detector.detect_upserted_containers(
-            silver_containers,
-            measurement_dim_table,
-            silver_last_modified_col=silver_col,
-            gold_last_modified_col=gold_col,
-        )
+        return detector, silver_containers, measurement_dim_table, silver_col, gold_col
+
+    @staticmethod
+    def _collect_container_ids(containers_df: DataFrame | None) -> list:
+        """Collect ``container_id`` values from a detected-containers DataFrame.
+
+        Parameters
+        ----------
+        containers_df : DataFrame | None
+            A detected-containers DataFrame (silver schema, includes
+            ``container_id``), or None.
+
+        Returns
+        -------
+        list
+            Container id values (empty when None).
+        """
+        if containers_df is None:
+            return []
+        return [row["container_id"] for row in containers_df.select("container_id").collect()]

src/impulse_reporting/core/report_utils.py

@@ -10,6 +10,7 @@
 from functools import reduce
 from typing import TYPE_CHECKING
 
+import pyspark.sql.functions as F
 from pyspark.sql import DataFrame, SparkSession
 
 if TYPE_CHECKING:
@@ -455,7 +456,8 @@ def cleanup_temp_tables(spark: SparkSession, catalog: str, schema: str) -> None:
 # Generic over the entity type-enum (``EventType`` / ``AggregationType`` /
 # ``ChannelType``); shared by events, aggregations, and calculated channels.
 # ``persist_facts_incremental`` groups by output table and unions shared-table
-# types before ``replace_by_ids``; ``merge_keys`` accepts a per-type callable.
+# types into a single ``merge_incremental``; ``merge_keys`` accepts a per-type
+# callable.
 # ---------------------------------------------------------------------------
 
 
@@ -631,19 +633,19 @@ def persist_facts_incremental(
     id_column: str,
     merge_keys: list[str] | Callable[[object], list[str]],
     changed_ids: dict[str, list[int]],
+    has_processed_containers: bool = False,
+    updated_container_ids: list | None = None,
+    container_id_col: str = "container_id",
 ) -> None:
-    """Incremental persist of fact DataFrames, grouped by output table.
+    """Incremental persist of fact DataFrames in a single MERGE per output table.
 
-    Per-type facts are grouped by their fact-table name, so entity types that
-    share a table (e.g. ``StatsAggregator`` + ``PointValueAggregator`` →
-    ``stats_aggregator_fact``, or mixed event types → ``event_instance_fact``)
-    are persisted together with no clobber.  For each table:
-
-    - **Changed** definitions (only types listed in *changed_ids*) are
-      ``unionByName``-combined and rewritten atomically in a single
-      ``replace_by_ids`` over all containers.
-    - **Unchanged** definitions are ``upsert``-ed (MERGE) over the incremental
-      container subset.
+    Per-type facts are grouped by fact-table name, so types sharing a table
+    (e.g. ``StatsAggregator`` + ``PointValueAggregator``, or mixed event types)
+    persist together with no clobber. Per table, changed rows (all containers)
+    and unchanged rows (reprocessed containers) are ``unionByName``-combined into
+    one ``sink.merge_incremental`` source; the delete scope prunes stale rows a
+    shrunk container leaves behind. The union is collision-free because an entity
+    is in exactly one bucket and ``merge_keys`` always includes its id.
 
     Parameters
     ----------
@@ -654,20 +656,33 @@ def persist_facts_incremental(
     type_enum : Enum
         The entity type-enum (resolves fact schema/uri + writer).
     sink : Sink
-        Target sink exposing ``replace_by_ids`` / ``upsert``.
+        Target sink exposing ``merge_incremental``.
     transform_fn : Callable[[DataFrame, StructType], DataFrame]
         Prepares a DataFrame for persistence (column projection + metadata).
     id_column : str
-        Column ``replace_by_ids`` targets for changed definitions (e.g.
+        Entity id column scoping changed-definition deletes (e.g.
         ``"channel_id"``, ``"visual_id"``, ``"event_id"``).
     merge_keys : list[str] or Callable
-        MERGE keys for unchanged upserts; a callable is resolved per entity type
-        (mirrors ``Report._get_aggregation_merge_keys``).
+        MERGE keys; a callable is resolved per entity type (mirrors
+        ``Report._get_aggregation_merge_keys``).
     changed_ids : dict[str, list[int]]
-        ``{type_name: [ids]}`` with changed definitions to replace.
+        ``{type_name: [ids]}`` with changed definitions recomputed over all
+        containers.
+    has_processed_containers : bool, optional
+        Whether any container was recomputed this run (new or updated). Gates
+        whether a table is written — new containers carry no delete scope but must
+        still be inserted. ``False`` + no changed ids → nothing to do.
+    updated_container_ids : list, optional
+        Ids of UPDATED containers only (present in gold, refreshed in silver).
+        Scopes the delete-by-source; new containers are excluded because they have
+        no gold rows to prune. Empty/None → no container-scoped delete.
+    container_id_col : str, optional
+        Gold fact-table container column, by default ``"container_id"``.
     """
     from impulse_reporting.persist.report_storage import WriterFactory
 
+    updated_container_ids = updated_container_ids or []
+
     # Group per-type facts by output table so shared-table types persist together.
     changed_by_table: dict[str, list[DataFrame]] = {}
     unchanged_by_table: dict[str, list[DataFrame]] = {}
@@ -692,24 +707,31 @@ def persist_facts_incremental(
         schema, uri = writer.extract_fact_schema_and_output_uri(entity_type)
         keys = _resolve_merge_keys(merge_keys, entity_type)
 
-        # Changed definitions: union then a single atomic replaceWhere.
-        changed_dfs = changed_by_table.get(table_name, [])
+        # Skip when there is nothing to write: no containers processed (new or
+        # updated) and no changed definitions. Keeps idempotent runs byte-identical
+        # (no no-op MERGE commit).
         table_changed_ids = changed_ids_by_table.get(table_name, [])
-        if changed_dfs and table_changed_ids:
-            combined = reduce(
-                lambda a, b: a.unionByName(b),
-                (transform_fn(cdf, schema) for cdf in changed_dfs),
-            )
-            sink.replace_by_ids(
-                df=combined,
-                uri=uri,
-                id_column=id_column,
-                ids_to_replace=table_changed_ids,
-            )
+        if not has_processed_containers and not table_changed_ids:
+            continue
 
-        # Unchanged definitions: MERGE over the incremental subset.
-        for udf in unchanged_by_table.get(table_name, []):
-            sink.upsert(transform_fn(udf, schema), uri, keys)
+        # Scope the delete-by-source: updated containers (only they can hold stale
+        # rows — new ones have none) and changed-definition entities.
+        delete_conditions = []
+        if updated_container_ids:
+            delete_conditions.append(
+                F.col(f"target.{container_id_col}").isin(updated_container_ids)
+            )
+        if table_changed_ids:
+            delete_conditions.append(F.col(f"target.{id_column}").isin(table_changed_ids))
+
+        # Single source: changed rows (all containers) + unchanged rows (processed
+        # containers), unioned by name into one MERGE.
+        source_dfs = [
+            transform_fn(df, schema)
+            for df in changed_by_table.get(table_name, []) + unchanged_by_table.get(table_name, [])
+        ]
+        source = reduce(lambda a, b: a.unionByName(b), source_dfs)
+        sink.merge_incremental(source, uri, keys, delete_conditions=delete_conditions)
 
 
 def persist_dimensions_incremental(