Fix: enforce chronological funnel order, normalize qualified filters

Author: nicosuave

sidemantic/sql/generator.py

@@ -75,6 +75,32 @@ def _build_interval(self, num: str, unit: str) -> str:
             # Standard SQL: INTERVAL '7 days'
             return f"INTERVAL '{num} {unit}'"
 
+    def _strip_model_prefixes(self, filters: list[str], model_name: str) -> list[str]:
+        """Strip model name prefixes from filter column references.
+
+        Conversion and retention CTEs query the raw table directly, so qualified
+        references like ``events.region`` need to become just ``region``.
+
+        Args:
+            filters: List of filter SQL expressions
+            model_name: Model name whose prefix should be stripped
+
+        Returns:
+            Filters with model prefixes removed
+        """
+        result = []
+        for f in filters:
+            try:
+                parsed = sqlglot.parse_one(f, dialect=self.dialect)
+                for column in parsed.find_all(exp.Column):
+                    tbl = column.table
+                    if tbl and tbl.replace("_cte", "") == model_name:
+                        column.set("table", None)
+                result.append(parsed.sql(dialect=self.dialect))
+            except Exception:
+                result.append(f)
+        return result
+
     def _quote_alias(self, name: str) -> str:
         """Quote an identifier for use as a SQL alias.
 
@@ -2413,7 +2439,9 @@ def _generate_multistep_conversion_query(
     ) -> str:
         """Generate SQL for N-step conversion funnel metrics.
 
-        Uses per-entity BOOL_OR aggregation for each step, then sums in the outer query.
+        Uses timestamp-aware chronological ordering: for each entity, tracks
+        the earliest timestamp at which each step was satisfied (via MIN/CASE),
+        then gates each step on occurring after all prior steps.
 
         Args:
             metric: The Metric object with steps defined
@@ -2454,16 +2482,37 @@ def _generate_multistep_conversion_query(
         if not model:
             raise ValueError(f"No model found for conversion metric {metric_name}")
 
+        # Find timestamp dimension: prefer model.default_time_dimension, fall back to first time dim
+        timestamp_dim = None
+        if model.default_time_dimension:
+            timestamp_dim = model.get_dimension(model.default_time_dimension)
+        if not timestamp_dim:
+            for dim in model.dimensions:
+                if dim.type == "time":
+                    timestamp_dim = dim
+                    break
+
+        if not timestamp_dim:
+            raise ValueError(
+                f"Multi-step conversion funnel requires a time dimension on model '{model.name}' "
+                "to enforce chronological step ordering"
+            )
+
+        ts_sql = timestamp_dim.sql_expr
+
         # Build FROM clause
         if model.sql:
             from_clause = f"({model.sql}) AS t"
         else:
             from_clause = model.table
 
+        # Normalize filters: strip model name prefixes so they work inside CTEs
+        normalized_filters = self._strip_model_prefixes(filters or [], model.name)
+
         # Build WHERE filter clause
         filter_clause = ""
-        if filters:
-            filter_clause = "\n  WHERE " + " AND ".join(filters)
+        if normalized_filters:
+            filter_clause = "\n  WHERE " + " AND ".join(normalized_filters)
 
         # Resolve dimension columns for GROUP BY support
         dim_entries: list[tuple[str, str]] = []
@@ -2486,10 +2535,12 @@ def _generate_multistep_conversion_query(
 
         dim_aliases = [alias for alias, _ in dim_entries]
 
-        # Build inner CTE: per-entity BOOL_OR for each step
+        # Build inner CTE: per-entity MIN timestamp for each step.
+        # MIN(CASE WHEN step_expr THEN timestamp END) captures the earliest time
+        # the step was satisfied, enabling chronological ordering in the outer query.
         step_cols = []
         for i, step_expr in enumerate(metric.steps, 1):
-            step_cols.append(f"BOOL_OR({step_expr}) AS step_{i}")
+            step_cols.append(f"MIN(CASE WHEN {step_expr} THEN {ts_sql} END) AS step_{i}_ts")
 
         inner_select_parts = [f"{metric.entity} AS entity"]
         if dim_entries:
@@ -2505,16 +2556,20 @@ def _generate_multistep_conversion_query(
         group_by_inner = ",\n    ".join(group_by_inner_parts)
 
         # Build outer SELECT: total entities (only step 1 entrants) + sum of each step.
-        # Gate each step on completion of all prior steps for monotonic funnels.
+        # Gate each step on chronological order: step_N_ts must be >= step_(N-1)_ts.
         outer_select_parts = []
         if dim_aliases:
             for alias in dim_aliases:
                 outer_select_parts.append(f"{alias}")
-        outer_select_parts.append("SUM(step_1::int) AS total_entities")
+        outer_select_parts.append("SUM(CASE WHEN step_1_ts IS NOT NULL THEN 1 ELSE 0 END) AS total_entities")
         for i in range(1, len(metric.steps) + 1):
-            # Each step requires all prior steps to be true
-            gate_expr = " AND ".join(f"step_{j}" for j in range(1, i + 1))
-            outer_select_parts.append(f"SUM(({gate_expr})::int) AS step_{i}_count")
+            # Each step requires all prior steps to have been completed in chronological order
+            conditions = ["step_1_ts IS NOT NULL"]
+            for j in range(2, i + 1):
+                conditions.append(f"step_{j}_ts IS NOT NULL")
+                conditions.append(f"step_{j}_ts >= step_{j - 1}_ts")
+            gate_expr = " AND ".join(conditions)
+            outer_select_parts.append(f"SUM(CASE WHEN {gate_expr} THEN 1 ELSE 0 END) AS step_{i}_count")
         outer_select = ",\n  ".join(outer_select_parts)
 
         # Build GROUP BY, ORDER BY, LIMIT for outer query

tests/metrics/test_advanced.py

@@ -570,25 +570,26 @@ def test_wow_percent_change_partitions_by_dimension():
 
 
 def test_multistep_conversion_funnel():
-    """Test N-step conversion funnel metric with BOOL_OR aggregation pattern."""
+    """Test N-step conversion funnel metric with timestamp-aware chronological ordering."""
     events = Model(
         name="events",
         sql="""
-            SELECT 1 AS person_id, 'Application Installed' AS event
-            UNION ALL SELECT 1, 'note.opened'
-            UNION ALL SELECT 1, 'note.created'
-            UNION ALL SELECT 1, 'git.sync'
-            UNION ALL SELECT 2, 'Application Installed'
-            UNION ALL SELECT 2, 'note.opened'
-            UNION ALL SELECT 2, 'note.created'
-            UNION ALL SELECT 3, 'Application Installed'
-            UNION ALL SELECT 3, 'note.opened'
-            UNION ALL SELECT 4, 'Application Installed'
+            SELECT 1 AS person_id, 'Application Installed' AS event, '2024-01-01'::TIMESTAMP AS ts
+            UNION ALL SELECT 1, 'note.opened', '2024-01-02'::TIMESTAMP
+            UNION ALL SELECT 1, 'note.created', '2024-01-03'::TIMESTAMP
+            UNION ALL SELECT 1, 'git.sync', '2024-01-04'::TIMESTAMP
+            UNION ALL SELECT 2, 'Application Installed', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 2, 'note.opened', '2024-01-02'::TIMESTAMP
+            UNION ALL SELECT 2, 'note.created', '2024-01-03'::TIMESTAMP
+            UNION ALL SELECT 3, 'Application Installed', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 3, 'note.opened', '2024-01-02'::TIMESTAMP
+            UNION ALL SELECT 4, 'Application Installed', '2024-01-01'::TIMESTAMP
         """,
         primary_key="person_id",
         dimensions=[
             Dimension(name="person_id", sql="person_id", type="categorical"),
             Dimension(name="event", sql="event", type="categorical"),
+            Dimension(name="ts", sql="ts", type="time"),
         ],
     )
 
@@ -638,15 +639,16 @@ def test_multistep_conversion_2_steps_via_steps_list():
     events = Model(
         name="events",
         sql="""
-            SELECT 1 AS user_id, 'signup' AS event
-            UNION ALL SELECT 1, 'purchase'
-            UNION ALL SELECT 2, 'signup'
-            UNION ALL SELECT 3, 'signup'
+            SELECT 1 AS user_id, 'signup' AS event, '2024-01-01'::TIMESTAMP AS ts
+            UNION ALL SELECT 1, 'purchase', '2024-01-02'::TIMESTAMP
+            UNION ALL SELECT 2, 'signup', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 3, 'signup', '2024-01-01'::TIMESTAMP
         """,
         primary_key="user_id",
         dimensions=[
             Dimension(name="user_id", sql="user_id", type="categorical"),
             Dimension(name="event", sql="event", type="categorical"),
+            Dimension(name="ts", sql="ts", type="time"),
         ],
     )
 
@@ -742,14 +744,15 @@ def test_conversion_metric_steps_and_base_event():
     events = Model(
         name="events",
         sql="""
-            SELECT 1 AS person_id, 'install' AS event
-            UNION ALL SELECT 1, 'activate'
-            UNION ALL SELECT 2, 'install'
+            SELECT 1 AS person_id, 'install' AS event, '2024-01-01'::TIMESTAMP AS ts
+            UNION ALL SELECT 1, 'activate', '2024-01-02'::TIMESTAMP
+            UNION ALL SELECT 2, 'install', '2024-01-01'::TIMESTAMP
         """,
         primary_key="person_id",
         dimensions=[
             Dimension(name="person_id", sql="person_id", type="categorical"),
             Dimension(name="event", sql="event", type="categorical"),
+            Dimension(name="ts", sql="ts", type="time"),
         ],
     )
 
@@ -777,7 +780,7 @@ def test_conversion_metric_steps_and_base_event():
     result = conn.execute(sql)
     rows = df_rows(result)
 
-    # steps takes precedence: uses BOOL_OR pattern
+    # steps takes precedence: uses timestamp-aware pattern
     assert rows[0][0] == 2  # total_entities
     assert rows[0][1] == 2  # step_1_count (install)
     assert rows[0][2] == 1  # step_2_count (activate)
@@ -819,18 +822,19 @@ def test_multistep_funnel_monotonic_counts():
     events = Model(
         name="events",
         sql="""
-            SELECT 1 AS person_id, 'install' AS event
-            UNION ALL SELECT 1, 'activate'
-            UNION ALL SELECT 1, 'purchase'
-            UNION ALL SELECT 2, 'install'
-            UNION ALL SELECT 2, 'activate'
-            UNION ALL SELECT 3, 'install'
-            UNION ALL SELECT 5, 'purchase'
+            SELECT 1 AS person_id, 'install' AS event, '2024-01-01'::TIMESTAMP AS ts
+            UNION ALL SELECT 1, 'activate', '2024-01-02'::TIMESTAMP
+            UNION ALL SELECT 1, 'purchase', '2024-01-03'::TIMESTAMP
+            UNION ALL SELECT 2, 'install', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 2, 'activate', '2024-01-02'::TIMESTAMP
+            UNION ALL SELECT 3, 'install', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 5, 'purchase', '2024-01-01'::TIMESTAMP
         """,
         primary_key="person_id",
         dimensions=[
             Dimension(name="person_id", sql="person_id", type="categorical"),
             Dimension(name="event", sql="event", type="categorical"),
+            Dimension(name="ts", sql="ts", type="time"),
         ],
     )
 
@@ -875,16 +879,17 @@ def test_multistep_funnel_total_excludes_non_entrants():
     events = Model(
         name="events",
         sql="""
-            SELECT 1 AS uid, 'signup' AS event
-            UNION ALL SELECT 1, 'purchase'
-            UNION ALL SELECT 2, 'signup'
-            UNION ALL SELECT 3, 'browse'
-            UNION ALL SELECT 4, 'browse'
+            SELECT 1 AS uid, 'signup' AS event, '2024-01-01'::TIMESTAMP AS ts
+            UNION ALL SELECT 1, 'purchase', '2024-01-02'::TIMESTAMP
+            UNION ALL SELECT 2, 'signup', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 3, 'browse', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 4, 'browse', '2024-01-01'::TIMESTAMP
         """,
         primary_key="uid",
         dimensions=[
             Dimension(name="uid", sql="uid", type="categorical"),
             Dimension(name="event", sql="event", type="categorical"),
+            Dimension(name="ts", sql="ts", type="time"),
         ],
     )
 
@@ -913,3 +918,115 @@ def test_multistep_funnel_total_excludes_non_entrants():
     assert rows[0][0] == 2  # total_entities = step 1 entrants
     assert rows[0][1] == 2  # step_1_count
     assert rows[0][2] == 1  # step_2_count (only user 1 purchased)
+
+
+def test_multistep_funnel_chronological_order_enforced():
+    """Test that funnel steps must occur in chronological order.
+
+    An entity that does step 2 before step 1 should NOT count as having
+    completed step 2 in the funnel, even if both events exist.
+    """
+    events = Model(
+        name="events",
+        sql="""
+            SELECT 1 AS user_id, 'signup' AS event, '2024-01-01'::TIMESTAMP AS ts
+            UNION ALL SELECT 1, 'purchase', '2024-01-05'::TIMESTAMP
+            UNION ALL SELECT 2, 'purchase', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 2, 'signup', '2024-01-10'::TIMESTAMP
+            UNION ALL SELECT 3, 'signup', '2024-01-01'::TIMESTAMP
+        """,
+        primary_key="user_id",
+        dimensions=[
+            Dimension(name="user_id", sql="user_id", type="categorical"),
+            Dimension(name="event", sql="event", type="categorical"),
+            Dimension(name="ts", sql="ts", type="time"),
+        ],
+    )
+
+    funnel = Metric(
+        name="ordered_funnel",
+        type="conversion",
+        entity="user_id",
+        steps=[
+            "event = 'signup'",
+            "event = 'purchase'",
+        ],
+    )
+
+    graph = SemanticGraph()
+    graph.add_model(events)
+    graph.add_metric(funnel)
+
+    generator = SQLGenerator(graph)
+    sql = generator.generate(metrics=["ordered_funnel"], dimensions=[])
+
+    conn = duckdb.connect(":memory:")
+    result = conn.execute(sql)
+    rows = df_rows(result)
+
+    # User 1: signup(Jan 1) -> purchase(Jan 5) = chronological, counts
+    # User 2: purchase(Jan 1) -> signup(Jan 10) = REVERSED, does NOT count for step 2
+    # User 3: signup(Jan 1) only = step 1 only
+    # total_entities = 3 (all have signup)
+    # step_1_count = 3
+    # step_2_count = 1 (only user 1 has purchase AFTER signup)
+    assert rows[0][0] == 3  # total_entities
+    assert rows[0][1] == 3  # step_1_count
+    assert rows[0][2] == 1  # step_2_count (only user 1, NOT user 2)
+
+
+def test_multistep_funnel_qualified_filters():
+    """Test that model-qualified filters work with multistep funnels.
+
+    Filters like 'events.region = ...' should be normalized to 'region = ...'
+    for SQL-backed models using FROM (<sql>) AS t.
+    """
+    events = Model(
+        name="events",
+        sql="""
+            SELECT 1 AS user_id, 'signup' AS event, 'US' AS region, '2024-01-01'::TIMESTAMP AS ts
+            UNION ALL SELECT 1, 'purchase', 'US', '2024-01-02'::TIMESTAMP
+            UNION ALL SELECT 2, 'signup', 'US', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 3, 'signup', 'EU', '2024-01-01'::TIMESTAMP
+            UNION ALL SELECT 3, 'purchase', 'EU', '2024-01-02'::TIMESTAMP
+        """,
+        primary_key="user_id",
+        dimensions=[
+            Dimension(name="user_id", sql="user_id", type="categorical"),
+            Dimension(name="event", sql="event", type="categorical"),
+            Dimension(name="region", sql="region", type="categorical"),
+            Dimension(name="ts", sql="ts", type="time"),
+        ],
+    )
+
+    funnel = Metric(
+        name="regional_funnel",
+        type="conversion",
+        entity="user_id",
+        steps=[
+            "event = 'signup'",
+            "event = 'purchase'",
+        ],
+    )
+
+    graph = SemanticGraph()
+    graph.add_model(events)
+    graph.add_metric(funnel)
+
+    generator = SQLGenerator(graph)
+    # Use model-qualified filter: events.region = 'US'
+    sql = generator.generate(
+        metrics=["regional_funnel"],
+        dimensions=[],
+        filters=["events.region = 'US'"],
+    )
+
+    conn = duckdb.connect(":memory:")
+    result = conn.execute(sql)
+    rows = df_rows(result)
+
+    # Only US users: user 1 (signup + purchase), user 2 (signup only)
+    # User 3 is EU, filtered out
+    assert rows[0][0] == 2  # total_entities (US users with signup)
+    assert rows[0][1] == 2  # step_1_count
+    assert rows[0][2] == 1  # step_2_count (only user 1 purchased)