diff --git a/src/impulse_reporting/aggregations/point_value_aggregator.py b/src/impulse_reporting/aggregations/point_value_aggregator.py index 790796aa..1bdfa7dc 100644 --- a/src/impulse_reporting/aggregations/point_value_aggregator.py +++ b/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()) ) diff --git a/src/impulse_reporting/aggregations/stats_aggregator.py b/src/impulse_reporting/aggregations/stats_aggregator.py index 1bc38d3a..bdbb378c 100644 --- a/src/impulse_reporting/aggregations/stats_aggregator.py +++ b/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( diff --git a/tests/impulse_reporting/integration/statistics_test.py b/tests/impulse_reporting/integration/statistics_test.py index 3cf8e50e..024af3fb 100644 --- a/tests/impulse_reporting/integration/statistics_test.py +++ b/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}" + ) diff --git a/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py b/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py index b3cb22b9..d6cf7045 100644 --- a/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py +++ b/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")