diff --git a/sidemantic/core/pre_aggregation.py b/sidemantic/core/pre_aggregation.py index 97d715198..361be1222 100644 --- a/sidemantic/core/pre_aggregation.py +++ b/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 diff --git a/sidemantic/sql/generator.py b/sidemantic/sql/generator.py index 281489059..79684fe50 100644 --- a/sidemantic/sql/generator.py +++ b/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)}" diff --git a/tests/metrics/test_filters.py b/tests/metrics/test_filters.py index e6fb3fcdd..cc6ed95c9 100644 --- a/tests/metrics/test_filters.py +++ b/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. diff --git a/tests/optimizations/test_pre_aggregations.py b/tests/optimizations/test_pre_aggregations.py index 0dd751035..22877cd2f 100644 --- a/tests/optimizations/test_pre_aggregations.py +++ b/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("""