Fix pre-aggregation materialization correctness

sidemantic/core/pre_aggregation.py

@@ -229,20 +229,36 @@ def generate_materialization_sql(self, model: Any, partition_filter: str | None
                         # PERCENTILE) are not re-aggregatable. Skip them (the matcher likewise
                         # never routes complete-expression measures to a rollup).
                         continue
-                    # Generate aggregation expression
+                    # Generate aggregation expression. Metric filters belong inside
+                    # the aggregate input, exactly as they do for live queries; a
+                    # rollup must never materialize an unfiltered value for a
+                    # filtered semantic metric.
                     agg_type = measure.agg.upper()
-                    if agg_type == "COUNT" and not measure.sql:
+                    filter_sql = " AND ".join(
+                        condition.replace("{model}.", "").replace("{model}", "")
+                        for condition in (measure.filters or [])
+                    )
+                    measure_input = measure.sql_expr
+                    if filter_sql:
+                        if agg_type == "COUNT" and not measure.sql:
+                            measure_input = f"CASE WHEN {filter_sql} THEN 1 ELSE NULL END"
+                        else:
+                            measure_input = f"CASE WHEN {filter_sql} THEN {measure_input} ELSE NULL END"
+
+                    if agg_type == "COUNT" and not measure.sql and not filter_sql:
                         # COUNT(*) case
                         select_exprs.append(f"COUNT(*) as {measure_name}_raw")
+                    elif agg_type == "COUNT" and not measure.sql:
+                        select_exprs.append(f"COUNT({measure_input}) as {measure_name}_raw")
                     elif agg_type == "AVG":
                         # Store AVG as additive sum state. A compatible count
                         # measure must also be present before query planning can
                         # roll this up safely.
-                        select_exprs.append(f"SUM({measure.sql_expr}) as {measure_name}_raw")
+                        select_exprs.append(f"SUM({measure_input}) as {measure_name}_raw")
                     elif agg_type == "COUNT_DISTINCT":
-                        select_exprs.append(f"COUNT(DISTINCT {measure.sql_expr}) as {measure_name}_raw")
+                        select_exprs.append(f"COUNT(DISTINCT {measure_input}) as {measure_name}_raw")
                     else:
-                        select_exprs.append(f"{agg_type}({measure.sql_expr}) as {measure_name}_raw")
+                        select_exprs.append(f"{agg_type}({measure_input}) as {measure_name}_raw")
 
         # A rollup that projects nothing would render "SELECT  FROM ... GROUP BY " (invalid
         # SQL). This happens when its measures are all non-materializable (agg=None: derived or
@@ -270,8 +286,9 @@ def generate_materialization_sql(self, model: Any, partition_filter: str | None
 
         sql = f"""SELECT
   {select_str}
-FROM {from_clause}{where_clause}
-GROUP BY {group_by_str}"""
+FROM {from_clause}{where_clause}"""
+        if group_by_str:
+            sql += f"\nGROUP BY {group_by_str}"
 
         return sql
 
@@ -754,6 +771,11 @@ def _refresh_incremental(
         if not table_exists:
             connection.execute(f"CREATE TABLE {table_name} AS {incremental_sql}")
         else:
+            # A lookback reprocesses an overlapping watermark range. Delete that
+            # range before inserting its replacement so repeated refreshes remain
+            # idempotent instead of accumulating duplicate rollup rows.
+            if lookback:
+                connection.execute(f"DELETE FROM {table_name} WHERE {watermark_column} >= {watermark_str}")
             connection.execute(f"INSERT INTO {table_name} {incremental_sql}")
 
         # Get new watermark

sidemantic/sql/generator.py

@@ -594,6 +594,59 @@ def _resolve_filter_dimensions(self, filters: list[str], model) -> list[str]:
                 result.append(f)
         return result
 
+    def _rewrite_filter_for_model_source(self, filter_sql: str, model) -> str:
+        """Resolve a pushed semantic filter against the model's raw source."""
+        model_name = model.name
+        source_alias = "t" if model.sql else None
+        resolved_filter = filter_sql.replace("{model}.", f"{model_name}.").replace("{model}", model_name)
+        try:
+            parsed = _parse_fragment(resolved_filter, self.dialect)
+        except Exception:
+            return resolved_filter
+
+        for column in list(parsed.find_all(exp.Column)):
+            table_name = column.table
+            if table_name and table_name.replace("_cte", "") != model_name:
+                continue
+
+            dimension_name = column.name
+            granularity = None
+            dimension = model.get_dimension(dimension_name)
+            if dimension is None and "__" in dimension_name:
+                candidate_name, candidate_granularity = dimension_name.rsplit("__", 1)
+                candidate = model.get_dimension(candidate_name)
+                if candidate is not None and candidate.type == "time":
+                    granularity = candidate_granularity
+                    dimension = candidate
+
+            if dimension is None:
+                column.set("table", exp.to_identifier(source_alias) if source_alias else None)
+                continue
+
+            self._ensure_sql_dimension(model_name, dimension)
+            replacement_sql = self._dimension_base_expr(dimension)
+            effective_granularity = granularity
+            if dimension.type == "time" and effective_granularity is None:
+                effective_granularity = dimension.granularity
+            if effective_granularity:
+                replacement_sql = self._date_trunc(effective_granularity, replacement_sql)
+            if source_alias:
+                replacement_sql = replacement_sql.replace("{model}", source_alias)
+            else:
+                replacement_sql = replacement_sql.replace("{model}.", "").replace("{model}", "")
+
+            try:
+                replacement = _parse_fragment(replacement_sql, self.dialect)
+            except Exception:
+                continue
+            for replacement_column in replacement.find_all(exp.Column):
+                replacement_table = replacement_column.table
+                if replacement_table and replacement_table.replace("_cte", "") == model_name:
+                    replacement_column.set("table", exp.to_identifier(source_alias) if source_alias else None)
+            column.replace(replacement)
+
+        return parsed.sql(dialect=self.dialect)
+
     def _quote_alias(self, name: str) -> str:
         """Quote an identifier for use as a SQL alias.
 
@@ -2370,22 +2423,7 @@ def collect_measures_from_metric(metric_ref: str, visited: set[str] | None = Non
         # Build WHERE clause for pushed-down filters
         where_clause = ""
         if filters:
-            # Process filters - replace model_cte references with direct column names using SQLGlot
-            processed_filters = []
-            for f in filters:
-                try:
-                    parsed = _parse_fragment(f, self.dialect)
-                    # Remove table qualifiers (model_name_cte. or model_name.)
-                    for col in parsed.find_all(exp.Column):
-                        if col.table:
-                            clean_table = col.table.replace("_cte", "")
-                            if clean_table == model_name:
-                                col.set("table", None)
-                    processed_filter = parsed.sql(dialect=self.dialect)
-                    processed_filters.append(processed_filter)
-                except Exception:
-                    # If parsing fails, use original filter
-                    processed_filters.append(f)
+            processed_filters = [self._rewrite_filter_for_model_source(f, model) for f in filters]
 
             where_clause = f"\n  WHERE {' AND '.join(processed_filters)}"
 

tests/metrics/test_filters.py

@@ -437,6 +437,63 @@ def test_filter_classification_and_rewrite_respect_nested_select_alias_scope():
     )
 
 
+def test_structured_filters_resolve_grained_and_computed_dimensions_before_where():
+    layer = SemanticLayer()
+    layer.add_model(
+        Model(
+            name="events",
+            table="events",
+            primary_key="id",
+            dimensions=[
+                Dimension(name="created_at", type="time", sql="occurred_at", granularity="day"),
+                Dimension(name="gross", type="numeric", sql="unit_price * quantity"),
+                Dimension(name="category", type="categorical"),
+            ],
+            metrics=[Metric(name="revenue", agg="sum", sql="amount")],
+        )
+    )
+    layer.conn.execute(
+        """
+        CREATE TABLE events (
+            id INTEGER,
+            occurred_at TIMESTAMP,
+            unit_price DOUBLE,
+            quantity INTEGER,
+            category VARCHAR,
+            amount DOUBLE
+        );
+        INSERT INTO events VALUES
+            (1, '2024-01-15 12:00:00', 5, 2, 'A', 10),
+            (2, '2024-02-10 08:00:00', 12, 2, 'A', 24),
+            (3, '2024-02-12 08:00:00', 4, 2, 'B', 8);
+        """
+    )
+    filters = [
+        "events.created_at__month = DATE '2024-02-01'",
+        "events.gross >= 20",
+    ]
+
+    postgres_sql = layer.compile(
+        metrics=["events.revenue"],
+        dimensions=["events.category"],
+        filters=filters,
+        dialect="postgres",
+    )
+    where_sql = postgres_sql.split("WHERE", 1)[1]
+    assert "created_at__month" not in where_sql
+    assert "events.gross" not in where_sql
+    assert "DATE_TRUNC('MONTH', occurred_at)" in where_sql
+    assert "unit_price * quantity >= 20" in where_sql
+
+    assert df_rows(
+        layer.query(
+            metrics=["events.revenue"],
+            dimensions=["events.category"],
+            filters=filters,
+        )
+    ) == [("A", 24.0)]
+
+
 def test_metric_level_filters_use_case_when(layer):
     """Test that Metric.filters are applied via CASE WHEN inside aggregation.
 

tests/optimizations/test_pre_aggregations.py

@@ -783,14 +783,14 @@ def test_refresh_incremental_with_lookback():
         lookback="5 days",
     )
 
-    # With append mode, we'll have duplicates
+    # Lookback refresh replaces the overlapping range instead of appending duplicates.
     total_revenue = conn.execute("""
         SELECT SUM(total_revenue)
         FROM orders_preagg_daily
         WHERE order_date = DATE '2024-01-05'
     """).fetchone()[0]
 
-    assert total_revenue == 1208  # 104 + 1104 from lookback
+    assert total_revenue == 1104
 
 
 def test_refresh_incremental_uses_update_window_as_default_lookback():
@@ -844,7 +844,7 @@ def test_refresh_incremental_uses_update_window_as_default_lookback():
         WHERE order_date = DATE '2024-01-05'
     """).fetchone()[0]
 
-    assert total_revenue == 1208  # late-arriving 1104 reprocessed because update_window covered it
+    assert total_revenue == 1104  # late-arriving value replaced because update_window covered it
 
 
 def test_explicit_lookback_overrides_update_window():
@@ -1100,6 +1100,45 @@ def test_generate_materialization_sql_no_time_dimension():
     assert "SUM(price) as avg_price_raw" in sql
 
 
+def test_total_rollup_materializes_filtered_metrics_without_empty_group_by():
+    conn = duckdb.connect(":memory:")
+    conn.execute(
+        "CREATE TABLE orders AS SELECT * FROM (VALUES "
+        "(1, 'completed', 100), (2, 'pending', 50), (3, 'completed', 25)) "
+        "t(id, status, amount)"
+    )
+    model = Model(
+        name="orders",
+        table="orders",
+        primary_key="id",
+        metrics=[
+            Metric(
+                name="completed_revenue",
+                agg="sum",
+                sql="amount",
+                filters=["{model}.status = 'completed'"],
+            ),
+            Metric(
+                name="completed_count",
+                agg="count",
+                filters=["{model}.status = 'completed'"],
+            ),
+        ],
+    )
+    preagg = PreAggregation(
+        name="totals",
+        measures=["completed_revenue", "completed_count"],
+        dimensions=[],
+    )
+
+    sql = preagg.generate_materialization_sql(model)
+
+    assert "GROUP BY" not in sql
+    assert "SUM(CASE WHEN status = 'completed' THEN amount ELSE NULL END)" in sql
+    assert "COUNT(CASE WHEN status = 'completed' THEN 1 ELSE NULL END)" in sql
+    assert conn.execute(sql).fetchall() == [(125, 2)]
+
+
 def test_avg_preaggregation_rolls_up_with_sum_count_state(layer):
     layer.use_preaggregations = True
     layer.conn.execute("""