Use xxHash64 for event_instance_id to prevent hash collisions

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.            |

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.
 

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),

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),

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

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

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).

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}: "