Add window function support to dimensions (#118)

* Add window function support to dimensions

* Auto-update JSON schema

* Fix: route window dimension filters to outer query

Window function dimensions (LEAD, LAG, etc.) are computed in the CTE
SELECT but haven't been evaluated at CTE WHERE time. The filter pushdown
logic now detects dimensions with window != None and keeps their filters
in the outer query. Also ensures columns referenced by outer-query
filters are included in CTE SELECT lists.

* Fix: exclude window-dim filters from preagg shared filters

* Fix: exclude window dimensions from pre-aggregation materialization

Window dimensions (LEAD, LAG, etc.) are incompatible with GROUP BY in
rollup materialization SQL. generate_materialization_sql now raises a
clear ValueError when a pre-aggregation references a dimension that has
a window expression set, for both regular dimensions and time dimensions.

* Fix: route multi-model window filters away from preagg shared WHERE

* Fix: window dim check takes priority over metric check in filter classification

* Fix: apply window dim filters as outer WHERE in preagg path

Window dimension filters were only pushed into the model sub-query that
owned the window dimension, leaving other models' metrics unfiltered.
Now the window dim column is projected through the owning model's preagg
CTE and the filter is applied on the outer preagg WHERE, constraining
all models' results.

* Fix: resolve {model} in filters, add metric-named output column

_strip_model_prefixes now replaces {model} with the actual model name
before sqlglot parsing, so placeholder-based filters no longer leak
literal {model} tokens into WHERE clauses.

Multistep funnel output now includes the last-step count aliased to
the metric name, so ORDER BY metric_name works without runtime errors.

* Fix: push window dim filters into model subqueries, preserve preagg grain

Remove window_dim_extra_dims logic that added extra dimensions to preagg
subqueries, changing the CTE's aggregation grain. Instead, push window
dim filters into each owning model's model_filters dict. The recursive
generate() call handles window dim filter placement correctly in its
own outer WHERE, preserving the requested dimension grain.

Also fix merge conflict markers in test_advanced.py from prior stash.

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: nicosuave

sidemantic-schema.json

@@ -169,6 +169,19 @@
           "default": null,
           "description": "Named format (e.g., 'usd', 'percent', 'decimal_2')",
           "title": "Value Format Name"
+        },
+        "window": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null,
+          "description": "Window function expression (e.g., 'LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)')",
+          "title": "Window"
         }
       },
       "required": [
@@ -1933,6 +1946,19 @@
                 "default": null,
                 "description": "Named format (e.g., 'usd', 'percent', 'decimal_2')",
                 "title": "Value Format Name"
+              },
+              "window": {
+                "anyOf": [
+                  {
+                    "type": "string"
+                  },
+                  {
+                    "type": "null"
+                  }
+                ],
+                "default": null,
+                "description": "Window function expression (e.g., 'LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)')",
+                "title": "Window"
               }
             },
             "required": [

sidemantic/adapters/sidemantic.py

@@ -265,6 +265,7 @@ def _parse_model(self, model_def: dict) -> Model | None:
                 value_format_name=dim_def.get("value_format_name"),
                 parent=dim_def.get("parent"),
                 metadata=dim_def.get("metadata"),
+                window=dim_def.get("window"),
             )
             dimensions.append(dimension)
 
@@ -541,6 +542,8 @@ def _export_model(self, model: Model) -> dict:
                     dim_def["value_format_name"] = dim.value_format_name
                 if dim.parent:
                     dim_def["parent"] = dim.parent
+                if dim.window:
+                    dim_def["window"] = dim.window
                 result["dimensions"].append(dim_def)
 
         # Export metrics (model-level aggregations)

sidemantic/core/dimension.py

@@ -32,6 +32,12 @@ class Dimension(BaseModel):
     # Arbitrary metadata (ai_context, custom_extensions, etc.)
     meta: dict[str, Any] | None = Field(None, description="Arbitrary metadata for extensions")
 
+    # Window function expression
+    window: str | None = Field(
+        None,
+        description="Window function expression (e.g., 'LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)')",
+    )
+
     # Visibility
     public: bool = Field(True, description="Whether dimension is visible in API/UI")
 
@@ -68,7 +74,22 @@ def __hash__(self) -> int:
 
     @property
     def sql_expr(self) -> str:
-        """Get SQL expression, defaulting to name if not specified."""
+        """Get the base SQL expression, defaulting to name if not specified.
+
+        Always returns the row-level expression (``sql`` or ``name``), never the
+        window function.  Use ``window_sql_expr`` when you need the window
+        expression for CTE projection.
+        """
+        return self.sql or self.name
+
+    @property
+    def window_sql_expr(self) -> str:
+        """Get the window SQL expression if set, otherwise fall back to sql_expr.
+
+        Use this in CTE SELECT lists where window functions should be projected.
+        """
+        if self.window:
+            return self.window
         return self.sql or self.name
 
     def with_granularity(self, granularity: str) -> str:

sidemantic/core/pre_aggregation.py

@@ -143,16 +143,28 @@ def generate_materialization_sql(self, model: Any) -> str:
         if self.time_dimension and self.granularity:
             time_dim = model.get_dimension(self.time_dimension)
             if time_dim:
+                if time_dim.window:
+                    raise ValueError(
+                        f"Cannot use window dimension '{self.time_dimension}' as time_dimension "
+                        f"in pre-aggregation '{self.name}': window functions are incompatible "
+                        f"with GROUP BY in rollup materialization"
+                    )
                 col_name = f"{self.time_dimension}_{self.granularity}"
                 select_exprs.append(f"DATE_TRUNC('{self.granularity}', {time_dim.sql_expr}) as {col_name}")
                 group_by_positions.append(str(pos))
                 pos += 1
 
-        # Add dimensions
+        # Add dimensions (reject window dimensions - incompatible with GROUP BY)
         if self.dimensions:
             for dim_name in self.dimensions:
                 dim = model.get_dimension(dim_name)
                 if dim:
+                    if dim.window:
+                        raise ValueError(
+                            f"Cannot use window dimension '{dim_name}' in pre-aggregation "
+                            f"'{self.name}': window functions are incompatible with "
+                            f"GROUP BY in rollup materialization"
+                        )
                     select_exprs.append(f"{dim.sql_expr} as {dim_name}")
                     group_by_positions.append(str(pos))
                     pos += 1

sidemantic/sql/generator.py

@@ -1114,10 +1114,13 @@ def replace_model_placeholder(sql_expr: str) -> str:
         for dimension in model.dimensions:
             if dimension.name in needed_dimensions and dimension.name not in columns_added:
                 # For time dimensions with granularity, apply DATE_TRUNC
+                # Use window_sql_expr for CTE projection so window functions
+                # (LEAD, LAG, etc.) are evaluated here.
+                base_expr = dimension.window_sql_expr
                 if dimension.type == "time" and dimension.granularity:
-                    dim_sql = self._date_trunc(dimension.granularity, dimension.sql_expr)
+                    dim_sql = self._date_trunc(dimension.granularity, base_expr)
                 else:
-                    dim_sql = dimension.sql_expr
+                    dim_sql = base_expr
                 # Replace {model} placeholder with actual table reference
                 dim_sql = replace_model_placeholder(dim_sql)
                 select_cols.append(f"{dim_sql} AS {self._quote_alias(dimension.name)}")
@@ -1242,7 +1245,7 @@ def collect_measures_from_metric(metric_ref: str, visited: set[str] | None = Non
                 continue
             dim = model.get_dimension(col_name)
             if dim:
-                dim_sql = replace_model_placeholder(dim.sql_expr)
+                dim_sql = replace_model_placeholder(dim.window_sql_expr)
                 select_cols.append(f"{dim_sql} AS {self._quote_alias(col_name)}")
                 columns_added.add(col_name)
                 continue
@@ -1525,10 +1528,10 @@ def _generate_with_preaggregation(
             cte_name = f"{model_name}_preagg"
             cte_names.append(cte_name)
 
-            # Only pass filters relevant to this model's sub-query.
-            # Window-dim filters are included here (not in shared_filters) because
-            # preagg CTEs only project query dimensions/metrics, not window dims.
-            # The recursive generate() call handles them correctly in its outer WHERE.
+            # Pass pushdown filters plus any window-dim filters for this model.
+            # Window-dim filters are pushed into the model's sub-query (not the
+            # outer preagg join) so the recursive generate() handles them in its
+            # own outer WHERE, preserving the requested dimension grain.
             model_filters = pushdown_by_model.get(model_name, []) + window_dim_filters.get(model_name, [])
 
             # Generate sub-query for this model's metrics at the dimension grain
@@ -1897,7 +1900,24 @@ def _build_main_select(
                 if references_metric:
                     break
 
+            # Check if filter also references a window dimension.
+            # Window dims are projected as regular columns in the CTE, so
+            # they belong in WHERE, not HAVING (they aren't aggregated).
+            references_window_dim = False
             if references_metric:
+                for model_name in [base_model_name] + other_models:
+                    model_obj = self.graph.get_model(model_name)
+                    if not model_obj:
+                        continue
+                    for field_name in re.findall(f"{model_name}\\.([a-zA-Z_][a-zA-Z0-9_]*)", filter_expr):
+                        dim = model_obj.get_dimension(field_name)
+                        if dim and getattr(dim, "window", None) is not None:
+                            references_window_dim = True
+                            break
+                    if references_window_dim:
+                        break
+
+            if references_metric and not references_window_dim:
                 having_filters.append(filter_expr)
             else:
                 where_filters.append(filter_expr)

tests/adapters/sidemantic_adapter/test_parsing.py

@@ -145,5 +145,147 @@ def test_adapter_validation():
     assert len(errors) == 0
 
 
+def test_dimension_window_field(tmp_path):
+    """Test that dimensions with window expressions are parsed and use window as sql_expr."""
+    adapter = SidemanticAdapter()
+    yaml_path = tmp_path / "events.yml"
+    yaml_path.write_text(
+        """
+models:
+  - name: events
+    table: public.events
+    primary_key: event_id
+    dimensions:
+      - name: event
+        type: categorical
+
+      - name: next_event
+        type: categorical
+        sql: event
+        window: "LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)"
+        description: The next event after this one for the same person
+
+      - name: next_timestamp
+        type: time
+        sql: timestamp
+        window: "LEAD(timestamp) OVER (PARTITION BY person_id ORDER BY timestamp)"
+        description: Timestamp of the next event
+
+      - name: plain_dim
+        type: categorical
+        sql: status
+
+    metrics:
+      - name: event_count
+        agg: count
+"""
+    )
+
+    graph = adapter.parse(yaml_path)
+    model = graph.models["events"]
+
+    # Dimension without window: sql_expr returns sql or name
+    event_dim = model.get_dimension("event")
+    assert event_dim.window is None
+    assert event_dim.sql_expr == "event"
+
+    plain_dim = model.get_dimension("plain_dim")
+    assert plain_dim.window is None
+    assert plain_dim.sql_expr == "status"
+
+    # Dimension with window: sql_expr returns the base expression,
+    # window_sql_expr returns the window function
+    next_event = model.get_dimension("next_event")
+    assert next_event.window == "LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)"
+    assert next_event.sql == "event"
+    assert next_event.sql_expr == "event"
+    assert next_event.window_sql_expr == "LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)"
+
+    next_ts = model.get_dimension("next_timestamp")
+    assert next_ts.window == "LEAD(timestamp) OVER (PARTITION BY person_id ORDER BY timestamp)"
+    assert next_ts.sql == "timestamp"
+    assert next_ts.sql_expr == "timestamp"
+    assert next_ts.window_sql_expr == next_ts.window
+
+
+def test_dimension_window_roundtrip(tmp_path):
+    """Test that window dimensions survive YAML export/import roundtrip."""
+    adapter = SidemanticAdapter()
+    yaml_path = tmp_path / "events.yml"
+    yaml_path.write_text(
+        """
+models:
+  - name: events
+    table: public.events
+    primary_key: event_id
+    dimensions:
+      - name: next_event
+        type: categorical
+        sql: event
+        window: "LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)"
+
+    metrics:
+      - name: event_count
+        agg: count
+"""
+    )
+
+    graph = adapter.parse(yaml_path)
+
+    # Export
+    export_path = tmp_path / "exported.yml"
+    adapter.export(graph, export_path)
+
+    # Re-import
+    graph2 = adapter.parse(export_path)
+    model2 = graph2.models["events"]
+    dim = model2.get_dimension("next_event")
+    assert dim.window == "LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)"
+    assert dim.sql == "event"
+    assert dim.window_sql_expr == dim.window
+    assert dim.sql_expr == "event"
+
+
+def test_dimension_window_in_sql_generation():
+    """Test that window dimensions produce correct SQL in generated queries."""
+    from sidemantic.core.dimension import Dimension
+    from sidemantic.core.metric import Metric
+    from sidemantic.core.model import Model
+    from sidemantic.core.semantic_graph import SemanticGraph
+    from sidemantic.sql.generator import SQLGenerator
+
+    graph = SemanticGraph()
+    model = Model(
+        name="events",
+        table="public.events",
+        primary_key="event_id",
+        dimensions=[
+            Dimension(
+                name="event",
+                type="categorical",
+            ),
+            Dimension(
+                name="next_event",
+                type="categorical",
+                sql="event",
+                window="LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)",
+            ),
+        ],
+        metrics=[
+            Metric(name="event_count", agg="count"),
+        ],
+    )
+    graph.add_model(model)
+
+    gen = SQLGenerator(graph, dialect="duckdb")
+    sql = gen.generate(
+        metrics=["events.event_count"],
+        dimensions=["events.next_event"],
+    )
+
+    # The window expression should appear in the generated SQL
+    assert "LEAD(event) OVER (PARTITION BY person_id ORDER BY timestamp)" in sql
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])