CLUSTER and MERGE split contained intervals into separate clusters — Closes #214

src/giql/expanders/cluster.py

@@ -3,8 +3,8 @@
 CLUSTER assigns a cluster id to every interval, grouping each run of mutually
 adjacent intervals (optionally within a maximum gap, strand-aware, and gated by a
 pairwise predicate) under one id. It cannot be a single window function because it
-needs a window over a window (``SUM`` over a ``LAG``-derived flag), so the
-expansion restructures the enclosing query into a two-level form::
+needs a window over a window (``SUM`` over a running-edge-derived boundary flag),
+so the expansion restructures the enclosing query into a two-level form::
 
     SELECT *, CLUSTER(interval) AS cluster_id FROM features
 
@@ -15,15 +15,20 @@
                AS cluster_id
     FROM (
         SELECT *,
-               CASE WHEN LAG(end) OVER (...) >= start THEN 0 ELSE 1 END
+               CASE WHEN MAX(end) OVER (...) >= start THEN 0 ELSE 1 END
                     AS __giql_is_new_cluster
         FROM features
     ) AS __giql_lag_calc
 
+The boundary flag keys off the running maximum end of the preceding rows (the
+cluster's right edge so far), not the immediately preceding row's end, so a later
+interval contained within an earlier wider one is not spuriously split (#214).
+
 (The ``becomes::`` form is simplified for readability; the emitted SQL quotes
-identifiers, appends ``NULLS LAST`` to the window ORDER BY, and — for a bare or
-qualified star projection — emits the outer star as ``* EXCEPT (__giql_is_new_cluster)``
-to hide the synthesized flag from the output (#184, #185).)
+identifiers, appends ``NULLS LAST`` to the window ORDER BY and a ``ROWS BETWEEN
+UNBOUNDED PRECEDING AND 1 PRECEDING`` frame, and — for a bare or qualified star
+projection — emits the outer star as ``* EXCEPT (__giql_is_new_cluster)`` to hide
+the synthesized flag from the output (#184, #185).)
 
 This module is the AST-expansion replacement for the legacy
 ``ClusterTransformer``, which ran as a pre-pass transformer on the raw parsed AST
@@ -242,32 +247,45 @@ def _transform_for_cluster(
     # Build ORDER BY for window
     order_by = [exp.Ordered(this=exp.column(start_col, quoted=True))]
 
-    # Create LAG window spec
-    lag_window = exp.Window(
-        this=exp.Anonymous(this="LAG", expressions=[exp.column(end_col, quoted=True)]),
+    # Cluster boundaries key off the running maximum end of the preceding rows --
+    # the cluster's right edge so far -- NOT the immediately preceding row's end.
+    # LAG(end) would spuriously split a later interval that still overlaps a
+    # cluster whose previous row is a contained interval ending early (#214).
+    running_max_end = exp.Window(
+        this=exp.Max(this=exp.column(end_col, quoted=True)),
         partition_by=partition_cols,
         order=exp.Order(expressions=order_by),
+        spec=exp.WindowSpec(
+            kind="ROWS",
+            start="UNBOUNDED",
+            start_side="PRECEDING",
+            end="1",
+            end_side="PRECEDING",
+        ),
     )
 
     # Add distance offset if specified
     if distance > 0:
-        lag_with_distance = exp.Add(
-            this=lag_window, expression=exp.Literal.number(distance)
+        edge_with_distance = exp.Add(
+            this=running_max_end, expression=exp.Literal.number(distance)
         )
     else:
-        lag_with_distance = lag_window
+        edge_with_distance = running_max_end
 
-    # Build the adjacency condition (predecessor end >= current start).
+    # Build the adjacency condition (running cluster edge >= current start).
     adjacency = exp.GTE(
-        this=lag_with_distance,
+        this=edge_with_distance,
         expression=exp.column(start_col, quoted=True),
     )
 
-    # An optional predicate further restricts which adjacent intervals
-    # are kept together: a row stays in the current cluster only when it
-    # is adjacent to its predecessor AND the predicate holds between them.
-    # ``PREV(col)`` references in the predicate resolve to the predecessor
-    # row via LAG over the same partition/order as the adjacency window.
+    # An optional predicate further restricts which adjacent intervals are kept
+    # together: a row stays in the current cluster only when it is adjacent to the
+    # cluster's running-max edge AND the predicate holds against its immediate
+    # sorted predecessor. ``PREV(col)`` references in the predicate resolve to that
+    # predecessor row via LAG over the same partition/order. Note the two notions
+    # can reference different rows under containment -- adjacency comes from an
+    # earlier, wider interval while the predicate compares to the immediate
+    # predecessor -- matching the documented immediate-predecessor semantics (#214).
     predicate_expr = cluster_expr.args.get("predicate")
     if predicate_expr is not None:
         rewritten_predicate = _rewrite_predecessor_refs(

tests/expanders/test_cluster.py

@@ -131,19 +131,19 @@ def test_transpile_should_use_custom_columns_when_table_declares_them(self):
 
         # Assert
         assert 'PARTITION BY "ch", "st" ORDER BY "s"' in sql
-        assert 'LAG("e")' in sql
+        assert 'MAX("e")' in sql
         assert '"chrom"' not in sql and '"start"' not in sql
 
-    def test_transpile_should_add_distance_offset_to_lag_when_distance_positive(self):
-        """Test that a positive CLUSTER distance offsets the adjacency LAG.
+    def test_transpile_should_offset_running_edge_when_distance_positive(self):
+        """Test that a positive CLUSTER distance offsets the running cluster edge.
 
         Given:
             A CLUSTER with a positive distance.
         When:
             Transpiling the query.
         Then:
-            The adjacency should add the distance to the LAG before comparing to
-            start.
+            The adjacency should add the distance to the running max end before
+            comparing to start.
         """
         # Arrange
         query = "SELECT *, CLUSTER(interval, 100) AS cid FROM peaks"
@@ -152,18 +152,22 @@ def test_transpile_should_add_distance_offset_to_lag_when_distance_positive(self
         sql = transpile(query, tables=["peaks"])
 
         # Assert
-        window = 'OVER (PARTITION BY "chrom" ORDER BY "start" NULLS LAST)'
-        assert f'LAG("end") {window} + 100 >= "start"' in sql
+        window = (
+            'OVER (PARTITION BY "chrom" ORDER BY "start" NULLS LAST '
+            "ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING)"
+        )
+        assert f'MAX("end") {window} + 100 >= "start"' in sql
 
-    def test_transpile_should_not_offset_lag_when_no_distance(self):
+    def test_transpile_should_not_offset_running_edge_when_no_distance(self):
         """Test that a CLUSTER without distance uses a bare adjacency.
 
         Given:
             A CLUSTER with no distance argument.
         When:
             Transpiling the query.
         Then:
-            The adjacency should compare the bare LAG to start with no offset.
+            The adjacency should compare the bare running max end to start with no
+            offset.
         """
         # Arrange
         query = "SELECT *, CLUSTER(interval) AS cid FROM peaks"
@@ -172,8 +176,11 @@ def test_transpile_should_not_offset_lag_when_no_distance(self):
         sql = transpile(query, tables=["peaks"])
 
         # Assert
-        window = 'OVER (PARTITION BY "chrom" ORDER BY "start" NULLS LAST)'
-        assert f'LAG("end") {window} >= "start"' in sql
+        window = (
+            'OVER (PARTITION BY "chrom" ORDER BY "start" NULLS LAST '
+            "ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING)"
+        )
+        assert f'MAX("end") {window} >= "start"' in sql
         assert f"{window} + " not in sql
 
     def test_transpile_should_split_clauses_between_lag_calc_and_outer_query(self):
@@ -1362,7 +1369,7 @@ def test_transpile_should_keep_clustering_when_sibling_shadows_genomic(self):
         When:
             Transpiling for DuckDB and executing it.
         Then:
-            The cluster ids should stay correct — the inner ``LAG("end")`` window must
+            The cluster ids should stay correct — the inner ``MAX("end")`` window must
             bind to the real ``end`` column, not the shadowing ``0``, because the sibling
             is materialized under a reserved name rather than ``end`` (#190).
         """

tests/expanders/test_genomic_columns.py

@@ -62,7 +62,7 @@ def test_transpile_should_emit_custom_window_when_cluster_over_derived_table(
 
         # Assert
         assert 'PARTITION BY "ch" ORDER BY "s" NULLS LAST' in sql
-        assert 'LAG("e")' in sql
+        assert 'MAX("e")' in sql
         assert '"chrom"' not in sql and '"start"' not in sql and '"end"' not in sql
 
     def test_transpile_should_emit_custom_aggregation_when_merge_over_derived_table(
@@ -285,7 +285,7 @@ def test_transpile_should_prefer_cte_body_over_registered_table_of_same_name(sel
 
         # Assert
         assert 'PARTITION BY "rc" ORDER BY "rs" NULLS LAST' in sql
-        assert 'LAG("re_")' in sql
+        assert 'MAX("re_")' in sql
         assert '"ch"' not in sql
 
     def test_transpile_should_resolve_partial_custom_mapping_through_derived_table(self):
@@ -309,7 +309,7 @@ def test_transpile_should_resolve_partial_custom_mapping_through_derived_table(s
 
         # Assert
         assert 'PARTITION BY "ch" ORDER BY "start" NULLS LAST' in sql
-        assert 'LAG("end")' in sql
+        assert 'MAX("end")' in sql
 
     def test_transpile_should_partition_by_custom_strand_when_stranded_over_derived(
         self,

tests/integration/bedtools/test_cluster.py

@@ -66,6 +66,46 @@ def test_cluster_basic(duckdb_connection):
     )
 
 
+def test_cluster_contained(duckdb_connection):
+    """
+    GIVEN a wide interval that contains a later narrower one which ends before a
+    third interval still inside the wide one
+    WHEN CLUSTER operator is applied via GIQL
+    THEN all three share one cluster_id, matching the single bedtools merge region
+    (#214 -- the boundary must key off the running max end, not the previous row)
+    """
+    intervals = [
+        GenomicInterval("chr1", 0, 1000, "i1", 100, "+"),
+        GenomicInterval("chr1", 100, 200, "i2", 150, "+"),  # contained in i1
+        GenomicInterval("chr1", 300, 400, "i3", 200, "+"),  # in i1, after i2's end
+    ]
+
+    load_intervals(
+        duckdb_connection,
+        "intervals",
+        [i.to_tuple() for i in intervals],
+    )
+
+    bedtools_merged = merge([i.to_tuple() for i in intervals])
+
+    sql = transpile(
+        """
+        SELECT *, CLUSTER(interval) AS cluster_id
+        FROM intervals
+        """,
+        tables=["intervals"],
+        dialect="duckdb",
+    )
+    giql_result = duckdb_connection.execute(sql).fetchall()
+
+    assert len(giql_result) == len(intervals)
+    cluster_ids = {row[-1] for row in giql_result}
+    assert len(cluster_ids) == len(bedtools_merged) == 1, (
+        "All contained intervals should share one cluster, matching the single "
+        "bedtools merge region"
+    )
+
+
 def test_cluster_separated(duckdb_connection):
     """
     GIVEN non-overlapping intervals with gaps

tests/integration/bedtools/test_merge.py

@@ -72,6 +72,38 @@ def test_merge_overlapping_intervals(duckdb_connection):
     assert comparison.match, comparison.failure_message()
 
 
+def test_merge_contained_intervals(duckdb_connection):
+    """
+    GIVEN a wide interval that contains a later narrower one which ends before a
+    third interval still inside the wide one
+    WHEN MERGE operator is applied
+    THEN all three merge into the single wide region, matching bedtools (#214 --
+    the boundary must key off the running max end, not the previous row's end)
+    """
+    intervals = [
+        GenomicInterval("chr1", 0, 1000, "i1", 100, "+"),
+        GenomicInterval("chr1", 100, 200, "i2", 150, "+"),  # contained in i1
+        GenomicInterval("chr1", 300, 400, "i3", 200, "+"),  # in i1, after i2's end
+    ]
+
+    load_intervals(
+        duckdb_connection,
+        "intervals",
+        [i.to_tuple() for i in intervals],
+    )
+
+    bedtools_result = merge([i.to_tuple() for i in intervals])
+
+    sql = transpile(
+        "SELECT MERGE(interval) FROM intervals",
+        tables=["intervals"],
+    )
+    giql_result = duckdb_connection.execute(sql).fetchall()
+
+    comparison = compare_results(giql_result, bedtools_result)
+    assert comparison.match, comparison.failure_message()
+
+
 def test_merge_separated_intervals(duckdb_connection):
     """
     GIVEN intervals with gaps between them

tests/test_cluster_predicate_transpilation.py

@@ -31,8 +31,9 @@ def test_transpile_without_predicate_is_unchanged(self):
 
         # Assert
         assert (
-            'CASE WHEN LAG("end") OVER (PARTITION BY "chrom" ORDER BY "start" '
-            'NULLS LAST) >= "start" THEN 0 ELSE 1 END' in sql
+            'CASE WHEN MAX("end") OVER (PARTITION BY "chrom" ORDER BY "start" '
+            "NULLS LAST ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) "
+            '>= "start" THEN 0 ELSE 1 END' in sql
         )
 
     def test_transpile_with_predicate_ands_into_case(self):
@@ -179,8 +180,9 @@ def test_transpile_without_predicate_is_unchanged(self):
 
         # Assert
         assert (
-            'CASE WHEN LAG("end") OVER (PARTITION BY "chrom" ORDER BY "start" '
-            'NULLS LAST) >= "start" THEN 0 ELSE 1 END' in sql
+            'CASE WHEN MAX("end") OVER (PARTITION BY "chrom" ORDER BY "start" '
+            "NULLS LAST ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) "
+            '>= "start" THEN 0 ELSE 1 END' in sql
         )
 
     def test_transpile_predicate_inherited_through_cluster(self):

tests/test_expander.py

@@ -1708,9 +1708,8 @@ def test_transform_replaces_cluster_with_lag_calc_subquery(self):
         windows = list(result.find_all(exp.Window))
         assert any(isinstance(w.this, exp.Sum) for w in windows)  # outer cluster id
         assert any(
-            isinstance(w.this, exp.Anonymous) and w.this.name.upper() == "LAG"
-            for w in windows
-        )  # inner adjacency LAG
+            isinstance(w.this, exp.Max) for w in windows
+        )  # inner adjacency running-max edge
         assert any(
             isinstance(a, exp.Alias) and a.alias == "__giql_is_new_cluster"
             for a in result.find_all(exp.Alias)