stdlib: Add interval_self_intersect_by_group macro

This enables efficient computation of self-intersecting intervals
within partitions (e.g., per-process) without cross-partition
interference.

Change-Id: I7105c8b06740a7ad381d60e74747301489b74a06

src/trace_processor/perfetto_sql/stdlib/intervals/intersect.sql

diff --git a/src/trace_processor/perfetto_sql/stdlib/intervals/intersect.sql b/src/trace_processor/perfetto_sql/stdlib/intervals/intersect.sql
index 0dbcde0..09c6835 100644
--- a/src/trace_processor/perfetto_sql/stdlib/intervals/intersect.sql
+++ b/src/trace_processor/perfetto_sql/stdlib/intervals/intersect.sql
@@ -272,3 +272,72 @@
   JOIN _atomic_segments a ON a.ts = e.ts
   WHERE e.is_start = FALSE
 );
+
+-- Like interval_self_intersect but partitions by group_column. Intervals in
+-- different groups do not intersect with each other.
+--
+-- Runtime is O(n log n + m) per group, where n is the number of intervals in
+-- each group and m is the size of the output.
+CREATE PERFETTO MACRO interval_self_intersect_by_group(
+  -- Table or subquery containing interval data.
+  intervals TableOrSubquery,
+  -- Column containing group name for partitioning.
+  group_column ColumnName
+)
+RETURNS TableOrSubquery
+AS
+(
+  WITH
+    _all_endpoints AS (
+      SELECT id, ts, TRUE as is_start, $group_column AS group_name FROM $intervals
+      UNION
+      SELECT id, ts + dur AS ts, FALSE as is_start, $group_column AS group_name FROM $intervals
+    ),
+    _atomic_segments AS (
+      SELECT
+        ROW_NUMBER() OVER (ORDER BY group_name, ts) AS id,
+        ts,
+        IFNULL(LEAD(ts) OVER (PARTITION BY group_name ORDER BY ts) - ts, 0) AS dur,
+        group_name
+      FROM _all_endpoints
+      GROUP BY ts, group_name
+    ),
+    _ii AS (
+      SELECT
+        ii.ts,
+        ii.dur,
+        ii.id_0 AS group_id,
+        ii.id_1 AS original_id,
+        ii.group_name
+      FROM _interval_intersect!((_atomic_segments, $intervals), (group_name)) ii
+    ),
+    _original_ends AS (
+      SELECT id, ts + dur AS end_ts, $group_column AS group_name FROM $intervals
+    )
+  -- Part A: Standard segments
+  SELECT
+    ts,
+    dur,
+    group_id,
+    original_id AS id,
+    FALSE AS interval_ends_at_ts,
+    group_name
+  FROM _ii
+  WHERE dur > 0
+
+  UNION ALL
+
+  -- Part B: End markers.
+  -- We join back to _atomic_segments to get the 'next' duration
+  -- to match the original implementation's quirk.
+  SELECT
+    e.ts AS ts,
+    a.dur AS dur,
+    a.id AS group_id,
+    e.id AS id,
+    TRUE AS interval_ends_at_ts,
+    e.group_name
+  FROM _all_endpoints e
+  JOIN _atomic_segments a ON a.ts = e.ts AND a.group_name = e.group_name
+  WHERE e.is_start = FALSE
+);

test/trace_processor/diff_tests/stdlib/intervals/tests.py

diff --git a/test/trace_processor/diff_tests/stdlib/intervals/tests.py b/test/trace_processor/diff_tests/stdlib/intervals/tests.py
index 5ca0304..2d8d2d2 100644
--- a/test/trace_processor/diff_tests/stdlib/intervals/tests.py
+++ b/test/trace_processor/diff_tests/stdlib/intervals/tests.py
@@ -305,6 +305,52 @@
         310,0,11,5,1
         """))
 
+  def test_intersect_list_by_group(self):
+    return DiffTestBlueprint(
+        trace=TextProto(""),
+        query="""
+        INCLUDE PERFETTO MODULE intervals.intersect;
+
+        WITH
+          data(ts, dur, id, group_name) AS (
+            VALUES
+              -- Group A: overlapping intervals
+              (10, 100, 0, 'A'),
+              (20, 40, 1, 'A'),
+              (30, 120, 2, 'A'),
+              -- Group B: separate intervals
+              (200, 10, 3, 'B'),
+              (200, 20, 4, 'B'),
+              -- Group C: single interval
+              (300, 10, 5, 'C')
+          )
+        SELECT *
+        FROM interval_self_intersect_by_group!(data, group_name)
+        ORDER BY group_name ASC, ts ASC, id ASC;
+        """,
+        out=Csv("""
+        "ts","dur","group_id","id","interval_ends_at_ts","group_name"
+        10,10,1,0,0,"A"
+        20,10,2,0,0,"A"
+        20,10,2,1,0,"A"
+        30,30,3,0,0,"A"
+        30,30,3,1,0,"A"
+        30,30,3,2,0,"A"
+        60,50,4,0,0,"A"
+        60,50,4,1,1,"A"
+        60,50,4,2,0,"A"
+        110,40,5,0,1,"A"
+        110,40,5,2,0,"A"
+        150,0,6,2,1,"A"
+        200,10,7,3,0,"B"
+        200,10,7,4,0,"B"
+        210,10,8,3,1,"B"
+        210,10,8,4,0,"B"
+        220,0,9,4,1,"B"
+        300,10,10,5,0,"C"
+        310,0,11,5,1,"C"
+        """))
+
   def test_interval_merge_overlapping_partitioned(self):
     return DiffTestBlueprint(
         trace=TextProto(""),