Fix container-event event_instance_id mismatch between stats_aggregator_fact and event_instance_fact

src/impulse_reporting/aggregations/point_value_aggregator.py

@@ -257,7 +257,7 @@ def determine_aggregations(
             .transform(StatsAggregator._add_event_name_column(aggregations))
             .transform(cls._explode_point_values)
             .transform(StatsAggregator._add_channel_name_column(aggregations))
-            .transform(StatsAggregator._add_event_instance_id_column)
+            .transform(StatsAggregator._add_event_instance_id_column(aggregations))
             .transform(StatsAggregator._add_visual_id_column(aggregations))
             .select(STATS_AGGREGATOR_FACT_SCHEMA.fieldNames())
         )

src/impulse_reporting/aggregations/stats_aggregator.py

@@ -336,7 +336,7 @@ def determine_aggregations(
             .transform(StatsAggregator._explode_stats_values)
             .transform(StatsAggregator._add_channel_name_column(aggregations))
             .transform(StatsAggregator._add_cross_channel_name_column(aggregations))
-            .transform(StatsAggregator._add_event_instance_id_column)
+            .transform(StatsAggregator._add_event_instance_id_column(aggregations))
             .transform(StatsAggregator._add_visual_id_column(aggregations))
             .select(STATS_AGGREGATOR_FACT_SCHEMA.fieldNames())
         )
@@ -656,24 +656,48 @@ def _(df: DataFrame) -> DataFrame:
         return _
 
     @staticmethod
-    def _add_event_instance_id_column(df: DataFrame) -> DataFrame:
+    def _add_event_instance_id_column(
+        aggregations: list[StatsAggregator],
+    ) -> Callable[..., DataFrame]:
         """
-        Add an event_instance_id column to the DataFrame.
+        Add an event_instance_id column, matching ``event_instance_fact``.
 
-        The event_instance_id uniquely identifies each event interval instance
-        within a container and event combination.
+        The id comes from ``generate_event_instance_id_column``: a ``ContainerEvent``
+        gets ``crc32(container_id)`` (one id per container), all other event types get
+        the timestamp-based hash. The container-event case is applied per row (keyed on
+        ``stats_name``) since a frame may mix event types.
 
         Parameters
         ----------
-        df : pyspark.sql.DataFrame
-            DataFrame containing interval_index column.
+        aggregations : list of StatsAggregator
+            List of StatsAggregator visual aggregations.
 
         Returns
         -------
-        pyspark.sql.DataFrame
-            DataFrame with event_instance_id column added.
+        function
+            Function that adds the event_instance_id column to a DataFrame.
         """
-        return df.withColumn("event_instance_id", generate_event_instance_id_column())
+        from impulse_reporting.events.container_event import ContainerEvent
+
+        def _(df: DataFrame) -> DataFrame:
+            container_event_stats_names = [
+                agg.get_name()
+                for agg in aggregations
+                if agg and isinstance(agg.get_event(), ContainerEvent)
+            ]
+
+            timestamp_based_id = generate_event_instance_id_column()
+            if container_event_stats_names:
+                event_instance_id_column = f.when(
+                    f.col("stats_name").isin(container_event_stats_names),
+                    generate_event_instance_id_column(event_type=ContainerEvent),
+                ).otherwise(timestamp_based_id)
+            else:
+                event_instance_id_column = timestamp_based_id
+
+            return df.withColumn("event_instance_id", event_instance_id_column)
+
+        return _
 
     @staticmethod
     def _add_visual_id_column(

tests/impulse_reporting/integration/statistics_test.py

@@ -408,5 +408,10 @@ def test_persist_statistics_with_events(spark):
         row.event_instance_id
         for row in event_instance_fact.select("event_instance_id").distinct().collect()
     )
-    # Stats event IDs should be a subset of (or equal to) event instance IDs
-    assert stats_event_ids.issubset(event_ids) or len(stats_event_ids) > 0
+    # Every event_instance_id in the stats fact must correspond to a real event
+    # instance in event_instance_fact — the two gold tables must stay consistent.
+    assert len(stats_event_ids) > 0
+    assert stats_event_ids.issubset(event_ids), (
+        f"stats_aggregator_fact event_instance_ids not in event_instance_fact: "
+        f"{stats_event_ids - event_ids}"
+    )

tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py

@@ -4,6 +4,7 @@
 Tests follow the same pattern as histogram_test.py.
 """
 
+import pyspark.sql.functions as f
 import pyspark.sql.types as T
 import pytest
 
@@ -15,6 +16,7 @@
 from impulse_query_engine.analyze.query.solvers.default_solver import DefaultSolver
 from impulse_reporting.aggregations.stats_aggregator import StatsAggregator
 from impulse_reporting.events.basic_event import BasicEvent
+from impulse_reporting.events.container_event import ContainerEvent
 from impulse_reporting.events.points_in_time_event import PointsInTimeEvent
 from impulse_reporting.persist.dimension_schema import STATS_AGGREGATOR_DIMENSION_SCHEMA
 
@@ -464,6 +466,70 @@ def test_determine_aggregations_multiple_stats(spark, basic_narrow_db):
     assert "Vehicle Speed" in channel_names_list
 
 
+def test_determine_aggregations_container_event_instance_id(spark, basic_narrow_db):
+    """Container-event stats rows must use crc32(container_id) as event_instance_id.
+
+    This is the value ``ContainerEvent.determine_events`` writes to
+    ``event_instance_fact`` (see ``generate_event_instance_id_column``'s
+    ``ContainerEvent`` branch), so the two gold tables stay consistent. Basic-event
+    stats in the same frame must keep the timestamp-based id instead.
+    """
+    eng_rpm = basic_narrow_db.query.channel(channel_name="Engine RPM")
+
+    container_event = ContainerEvent(name="full_container")
+    basic_event = BasicEvent(name="rpm_event", expr=eng_rpm > 500)
+
+    container_stats = StatsAggregator(
+        name="container_stats",
+        input_expressions=[eng_rpm],
+        channel_names=["Engine RPM"],
+        statistics=["min", "max", "mean"],
+        event=container_event,
+    )
+    basic_stats = StatsAggregator(
+        name="basic_stats",
+        input_expressions=[eng_rpm],
+        channel_names=["Engine RPM"],
+        statistics=["min", "max", "mean"],
+        event=basic_event,
+    )
+
+    solver = DefaultSolver(spark)
+    solved_df = basic_narrow_db.query.select(
+        container_stats.get_expression(), basic_stats.get_expression()
+    ).solve(spark, solver)
+    df = StatsAggregator.determine_aggregations(
+        spark=spark,
+        aggregations=[container_stats, basic_stats],
+        solved_df=solved_df,
+    ).withColumn("expected_container_id", f.crc32(f.col("container_id").cast("string")))
+
+    # Every container-event row uses crc32(container_id) — matching event_instance_fact.
+    container_rows = df.filter(f.col("visual_id") == container_stats.get_id()).collect()
+    assert len(container_rows) > 0
+    for row in container_rows:
+        assert row.event_instance_id == row.expected_container_id, (
+            f"container_id={row.container_id}: event_instance_id="
+            f"{row.event_instance_id} != crc32(container_id)={row.expected_container_id}"
+        )
+
+    # ... and exactly one distinct event_instance_id per container (no fragmentation).
+    per_container = (
+        df.filter(f.col("visual_id") == container_stats.get_id())
+        .groupBy("container_id")
+        .agg(f.countDistinct("event_instance_id").alias("n"))
+        .collect()
+    )
+    for row in per_container:
+        assert row.n == 1, f"container {row.container_id} fragmented into {row.n} ids"
+
+    # Basic-event rows must NOT collapse to crc32(container_id); they use the
+    # timestamp-based hash (container_id::event_name::start_ts::end_ts).
+    basic_rows = df.filter(f.col("visual_id") == basic_stats.get_id()).collect()
+    assert len(basic_rows) > 0
+    assert all(row.event_instance_id != row.expected_container_id for row in basic_rows)
+
+
 def test_determine_metadata_df(spark, basic_narrow_db):
     """Test that determine_metadata_df returns a DataFrame with expected columns."""
     eng_rpm = basic_narrow_db.query.channel(channel_name="Engine RPM")