From 27f9f792cef16a71bf9e7578025c5059d0fdf45a Mon Sep 17 00:00:00 2001 From: "tom.bonfert" Date: Fri, 21 Aug 2026 12:01:46 +0200 Subject: [PATCH 1/3] Switch event_instance_id hashing from crc32 to xxhash64 and widen fact schema to LongType Replaces the CRC32-based event_instance_id generation with xxHash64 to reduce collisions at scale, and updates `event_instance_id` in `EVENT_INSTANCE_FACT_SCHEMA` from `IntegerType` to `LongType` to hold the signed 64-bit hash values. ContainerEvent rows continue to hash only `container_id`, while other event types hash `container_id::event_name::start_ts::end_ts`. --- .../aggregations/stats_aggregator.py | 2 +- src/impulse_reporting/persist/fact_schema.py | 2 +- .../util/event_instance_util.py | 20 ++++++++++--------- .../aggregations/stats_aggregator_test.py | 6 +++--- .../unit/events/container_event_test.py | 4 ++-- 5 files changed, 18 insertions(+), 16 deletions(-) diff --git a/src/impulse_reporting/aggregations/stats_aggregator.py b/src/impulse_reporting/aggregations/stats_aggregator.py index bdbb378c..e5a6eb7f 100644 --- a/src/impulse_reporting/aggregations/stats_aggregator.py +++ b/src/impulse_reporting/aggregations/stats_aggregator.py @@ -663,7 +663,7 @@ def _add_event_instance_id_column( Add an event_instance_id column, matching ``event_instance_fact``. The id comes from ``generate_event_instance_id_column``: a ``ContainerEvent`` - gets ``crc32(container_id)`` (one id per container), all other event types get + gets ``xxhash64(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. diff --git a/src/impulse_reporting/persist/fact_schema.py b/src/impulse_reporting/persist/fact_schema.py index 59c8d2d0..8b6293ec 100644 --- a/src/impulse_reporting/persist/fact_schema.py +++ b/src/impulse_reporting/persist/fact_schema.py @@ -40,7 +40,7 @@ EVENT_INSTANCE_FACT_SCHEMA = StructType( [ StructField("container_id", IntegerType(), False), - StructField("event_instance_id", IntegerType(), False), + StructField("event_instance_id", LongType(), False), StructField("event_id", IntegerType(), False), StructField("start_ts", LongType(), False), StructField("end_ts", LongType(), False), diff --git a/src/impulse_reporting/util/event_instance_util.py b/src/impulse_reporting/util/event_instance_util.py index d0113270..64763986 100644 --- a/src/impulse_reporting/util/event_instance_util.py +++ b/src/impulse_reporting/util/event_instance_util.py @@ -21,17 +21,19 @@ def generate_event_instance_id_column( """ Generate an event_instance_id column. - For ``ContainerEvent`` the sentinel value ``-1`` is returned because a - container event produces exactly one instance per container. - For all other event types a CRC32 hash of - ``container_id::event_name::start_ts::end_ts`` is used. + The id is an xxHash64 of ``container_id::event_name::start_ts::end_ts``. + For ``ContainerEvent`` only ``container_id`` is hashed, since a container + event produces exactly one instance per container. The result is a signed + 64-bit long (may be negative), wide enough to keep this merge/join key + collision-free at scale. Parameters ---------- event_type : type[Event] or None, optional - The event class. When the class is ``ContainerEvent``, ``container_id`` - column is used for CRC32 hash. For any other value (including ``None`` - for backward-compatibility) the CRC32 hash column is returned. + The event class. When the class is ``ContainerEvent``, the + ``container_id`` column is hashed. For any other value (including + ``None`` for backward-compatibility) the timestamp-based hash column is + returned. container_id_col : str, optional Name of the container ID column, defaults to "container_id". event_name_col : str, optional @@ -49,9 +51,9 @@ def generate_event_instance_id_column( from impulse_reporting.events.container_event import ContainerEvent if event_type is ContainerEvent: - return f.crc32(f.col(container_id_col).cast("string")) + return f.xxhash64(f.col(container_id_col).cast("string")) - return f.crc32( + return f.xxhash64( f.concat_ws( "::", f.col(container_id_col), diff --git a/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py b/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py index d6cf7045..646d350b 100644 --- a/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py +++ b/tests/impulse_reporting/unit/aggregations/stats_aggregator_test.py @@ -502,15 +502,15 @@ def test_determine_aggregations_container_event_instance_id(spark, basic_narrow_ spark=spark, aggregations=[container_stats, basic_stats], solved_df=solved_df, - ).withColumn("expected_container_id", f.crc32(f.col("container_id").cast("string"))) + ).withColumn("expected_container_id", f.xxhash64(f.col("container_id").cast("string"))) - # Every container-event row uses crc32(container_id) — matching event_instance_fact. + # Every container-event row uses xxhash64(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}" + f"{row.event_instance_id} != xxhash64(container_id)={row.expected_container_id}" ) # ... and exactly one distinct event_instance_id per container (no fragmentation). diff --git a/tests/impulse_reporting/unit/events/container_event_test.py b/tests/impulse_reporting/unit/events/container_event_test.py index b65ab551..9a9914da 100644 --- a/tests/impulse_reporting/unit/events/container_event_test.py +++ b/tests/impulse_reporting/unit/events/container_event_test.py @@ -163,9 +163,9 @@ def test_determine_events(spark, basic_narrow_db): assert "end_ts" in df.columns assert df.count() > 0 - # Compare event_instance_id with crc32(container_id) + # Compare event_instance_id with xxhash64(container_id) - df_with_test = df.withColumn("test_values", f.crc32(f.col("container_id").cast("string"))) + df_with_test = df.withColumn("test_values", f.xxhash64(f.col("container_id").cast("string"))) for row in df_with_test.collect(): assert row.event_instance_id == row.test_values, ( f"container_id={row.container_id}: " From 792afa13d698654977de7e60025b1432f57783c2 Mon Sep 17 00:00:00 2001 From: "tom.bonfert" Date: Fri, 21 Aug 2026 12:14:18 +0200 Subject: [PATCH 2/3] docs: update event_instance_id description to reflect xxHash64 hashing --- docs/impulse/docs/references/report/event.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/impulse/docs/references/report/event.md b/docs/impulse/docs/references/report/event.md index 02142ef1..86ecd088 100644 --- a/docs/impulse/docs/references/report/event.md +++ b/docs/impulse/docs/references/report/event.md @@ -214,7 +214,7 @@ Stores materialized event occurrences (one row per event instance per container) | Column | Type | Description | |---------------------|--------|------------------------------------------| | `container_id` | `int` | Container identifier. | -| `event_instance_id` | `long` | Unique instance identifier (CRC32 hash). | +| `event_instance_id` | `long` | Unique instance identifier (xxHash64). | | `event_id` | `int` | Foreign key to `event_dimension`. | | `start_ts` | `long` | Event instance start timestamp. | | `end_ts` | `long` | Event instance end timestamp. | From 425fa9e6de0227c613bff9a6c1b6fac4aaacb3e2 Mon Sep 17 00:00:00 2001 From: "tom.bonfert" Date: Fri, 21 Aug 2026 12:46:08 +0200 Subject: [PATCH 3/3] Update BasicEvent event_instance_id assertions for signed xxHash64 ids Replaces the non-negative CRC32 assertion with checks that every BasicEvent instance has an event_instance_id and that all instance ids are distinct, matching the switch to per-instance timestamp-based xxHash64 hashing. --- .../container_and_sequence_event_test.py | 13 ++++++++----- .../integration/container_event_test.py | 13 ++++++++----- 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/tests/impulse_reporting/integration/container_and_sequence_event_test.py b/tests/impulse_reporting/integration/container_and_sequence_event_test.py index 24a27f87..1b443d38 100644 --- a/tests/impulse_reporting/integration/container_and_sequence_event_test.py +++ b/tests/impulse_reporting/integration/container_and_sequence_event_test.py @@ -195,13 +195,16 @@ def test_container_event_with_basic_event(spark, basic_narrow_db): assert "CONTAINER_EVENT" in event_dfs assert "BASIC_EVENT" in event_dfs - # BasicEvent instances: should have non-negative event_instance_id (CRC32 hashes) + # BasicEvent instances use the per-instance timestamp-based xxHash64, so each + # instance carries a distinct id (which may be negative, unlike a + # ContainerEvent's single per-container id). basic_rows = event_dfs["BASIC_EVENT"]["changed"].collect() assert len(basic_rows) > 0, "Should have at least one basic event instance" - for row in basic_rows: - assert ( - row.event_instance_id >= 0 - ), f"BasicEvent should have non-negative event_instance_id, got {row.event_instance_id}" + basic_ids = [row.event_instance_id for row in basic_rows] + assert all( + i is not None for i in basic_ids + ), "BasicEvent instances must have an event_instance_id" + assert len(set(basic_ids)) == len(basic_ids), "BasicEvent instance ids must be distinct" # Event metadata: should have 2 event definitions event_metadata_dfs = my_report.event_metadata_dfs diff --git a/tests/impulse_reporting/integration/container_event_test.py b/tests/impulse_reporting/integration/container_event_test.py index 662a77c5..8c89e55b 100644 --- a/tests/impulse_reporting/integration/container_event_test.py +++ b/tests/impulse_reporting/integration/container_event_test.py @@ -193,13 +193,16 @@ def test_container_event_with_basic_event(spark, basic_narrow_db): assert "CONTAINER_EVENT" in event_dfs assert "BASIC_EVENT" in event_dfs - # BasicEvent instances: should have non-negative event_instance_id (CRC32 hashes) + # BasicEvent instances use the per-instance timestamp-based xxHash64, so each + # instance carries a distinct id (which may be negative, unlike a + # ContainerEvent's single per-container id). basic_rows = event_dfs["BASIC_EVENT"]["changed"].collect() assert len(basic_rows) > 0, "Should have at least one basic event instance" - for row in basic_rows: - assert ( - row.event_instance_id >= 0 - ), f"BasicEvent should have non-negative event_instance_id, got {row.event_instance_id}" + basic_ids = [row.event_instance_id for row in basic_rows] + assert all( + i is not None for i in basic_ids + ), "BasicEvent instances must have an event_instance_id" + assert len(set(basic_ids)) == len(basic_ids), "BasicEvent instance ids must be distinct" # Event metadata: should have 2 event definitions event_metadata_dfs = my_report.event_metadata_dfs