diff --git a/datajunction-server/datajunction_server/api/sql.py b/datajunction-server/datajunction_server/api/sql.py index 7b1e7f7c6..ab1737e88 100644 --- a/datajunction-server/datajunction_server/api/sql.py +++ b/datajunction-server/datajunction_server/api/sql.py @@ -189,6 +189,20 @@ async def get_measures_sql_v3( ), ), use_materialized: bool = Query(True), + aggregate: bool = Query( + True, + description=( + "Whether to apply server-side aggregation at the requested grain " + "(default: True). When False, the response SQL is a flat SELECT " + "with no GROUP BY and projects each component column under its " + "exact node column name (the v3-compatible output format, not " + "v2's `_DOT_` convention). Pre-aggregation table " + "matching is bypassed. Use this when the caller applies its own " + "aggregation downstream. The `metric_formulas[].combiner` field " + "is still populated in either mode so the caller can re-aggregate " + "the raw output. Semantic parity with v2's `preaggregate=False`." + ), + ), dialect: Dialect = Query( Dialect.SPARK, description="SQL dialect for the generated query.", @@ -243,6 +257,10 @@ async def get_measures_sql_v3( use_materialized: If True (default), use materialized tables when available. Set to False when generating SQL for materialization refresh to avoid circular references. + aggregate: If True (default), apply server-side aggregation. When False, + emit a flat SELECT with no GROUP BY, project each component column + under its exact node column name, and bypass pre-aggregation table + matching. include_temporal_filters: If True, checks if metrics+dimensions resolve to a cube with temporal partitions, and applies partition filters if so. lookback_window: Lookback window for temporal filters when applicable. @@ -277,6 +295,7 @@ async def get_measures_sql_v3( lookback_window=lookback_window, query_parameters=json.loads(query_params) or None, matched_cube=cube_node.current if cube_node else None, + aggregate=aggregate, ) response = _build_measures_response(result) diff --git a/datajunction-server/datajunction_server/construction/build_v3/builder.py b/datajunction-server/datajunction_server/construction/build_v3/builder.py index 8703c9c7a..ddb075e5b 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/builder.py +++ b/datajunction-server/datajunction_server/construction/build_v3/builder.py @@ -241,6 +241,7 @@ async def setup_build_context( include_temporal_filters: bool = False, lookback_window: str | None = None, matched_cube: Optional["NodeRevision"] = None, + aggregate: bool = True, ) -> BuildContext: """ Create and initialize a BuildContext with all setup done. @@ -261,6 +262,12 @@ async def setup_build_context( use_materialized: Whether to use materialized tables include_temporal_filters: Whether to include temporal partition filters from cube lookback_window: Lookback window for temporal filters + aggregate: Whether to apply server-side aggregation (GROUP BY). When False, + emits a flat SELECT projecting each component column under its + exact node column name (v3-compatible output format, not the v2 + ``_DOT_`` convention) and bypasses pre-aggregation + table matching. Used by callers that apply their own aggregation + downstream (semantic parity with v2's ``preaggregate=False`` mode). Returns: Fully initialized BuildContext @@ -284,6 +291,7 @@ async def setup_build_context( use_materialized=use_materialized, temporal_partition_columns=temporal_partition_columns or {}, lookback_window=lookback_window, + aggregate=aggregate, ) # Add filter-driven dimensions upfront — this only parses filter strings and @@ -346,6 +354,7 @@ async def build_measures_sql( lookback_window: str | None = None, query_parameters: dict[str, Any] | None = None, matched_cube: Optional["NodeRevision"] = None, + aggregate: bool = True, ) -> GeneratedMeasuresSQL: """ Build measures SQL for a set of metrics, dimensions, and filters. @@ -370,6 +379,15 @@ async def build_measures_sql( incremental materialization and partition pruning. lookback_window: Lookback window for temporal filters (e.g., "3 DAY"). If not provided, filters to exactly the logical timestamp partition. + aggregate: If True (default), apply server-side aggregation at the + requested grain. When False, emit a flat SELECT with no GROUP BY, + project each component column under its exact node column name + (v3-compatible output format, not the v2 ``_DOT_`` + convention), and bypass pre-aggregation table matching. Intended + for callers that apply their own aggregation downstream (semantic + parity with v2's ``preaggregate=False`` mode). The + ``metric_formulas[].combiner`` field is still populated so callers + know how to aggregate the raw output. Returns: GeneratedMeasuresSQL with one GrainGroupSQL per aggregation level, @@ -386,6 +404,7 @@ async def build_measures_sql( include_temporal_filters=include_temporal_filters, lookback_window=lookback_window, matched_cube=matched_cube, + aggregate=aggregate, ) # Build grain groups from context diff --git a/datajunction-server/datajunction_server/construction/build_v3/measures.py b/datajunction-server/datajunction_server/construction/build_v3/measures.py index d0ec0a58d..3df873067 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/measures.py +++ b/datajunction-server/datajunction_server/construction/build_v3/measures.py @@ -1887,8 +1887,14 @@ def build_grain_group_sql( """ parent_node = grain_group.parent_node - # Check for matching pre-aggregation - if ctx.use_materialized and ctx.available_preaggs: + # Check for matching pre-aggregation. + # Skip matching when ``ctx.aggregate`` is False: the caller is asking + # for raw row-level output, which by definition cannot be served from a + # pre-aggregated table. We do NOT touch ``ctx.use_materialized`` here + # because that flag is dual-purpose: it also controls whether parent + # transforms are read from their materialized table vs recomputed via + # CTE (see ``should_use_materialized_table`` in ``materialization.py``). + if ctx.aggregate and ctx.use_materialized and ctx.available_preaggs: requested_grain = [dim.original_ref for dim in resolved_dimensions] matching_preagg = find_matching_preagg( ctx, @@ -1933,6 +1939,25 @@ def build_grain_group_sql( # Collect unique components for API response unique_components.append(component) + # When the caller has opted out of server-side aggregation (the v3 + # equivalent of v2's ``preaggregate=False``), project the raw inner + # expression under its exact node column name (e.g. ``is_product_view``) + # instead of wrapping it in the component's aggregation (``SUM(...)``). + # The downstream caller applies its own aggregation using the value of + # ``metric_formulas[].combiner`` from the response, so emitting the + # aggregation server-side would double-count. Mirrors the existing + # NONE-aggregability path below: the v3 output stays v3-shaped (bare + # column names), not the v2 ``_DOT_`` convention. + if not ctx.aggregate and component.expression: + col_ast = make_column_ref(component.expression) + component_alias = component.expression + component_expressions.append((component_alias, col_ast)) + component_metadata.append( + (component_alias, component, metric_node), + ) + component_aliases[component.name] = component_alias + continue + # For NONE aggregability, output raw columns (no aggregation possible) # Note: This path is only hit for BASE metrics with NONE aggregability # (e.g., metrics with RANK() directly). Derived metrics with window functions @@ -2067,9 +2092,15 @@ def build_grain_group_sql( # refs (not pre-aggregations). A GROUP BY here would be redundant # — every projected column is part of the full grain — and the # real aggregation happens downstream after the join. - skip_agg = bool( - grain_group.non_decomposable_metrics - and grain_group.aggregability == Aggregability.NONE, + # When ``ctx.aggregate`` is False, every component above also emitted + # raw column refs, so a GROUP BY would aggregate raw rows the caller + # explicitly asked to receive ungrouped. + skip_agg = ( + bool( + grain_group.non_decomposable_metrics + and grain_group.aggregability == Aggregability.NONE, + ) + or not ctx.aggregate ) query_ast, scanned_sources = build_select_ast( ctx, diff --git a/datajunction-server/datajunction_server/construction/build_v3/types.py b/datajunction-server/datajunction_server/construction/build_v3/types.py index 73f74ddc1..3e1f19083 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/types.py +++ b/datajunction-server/datajunction_server/construction/build_v3/types.py @@ -53,6 +53,15 @@ class BuildContext: # Set to False when building SQL for materialization to avoid circular references use_materialized: bool = True + # Whether to apply server-side aggregation (default: True) + # When False, emits a flat SELECT with no GROUP BY and projects each + # component column under its exact node column name (the v3-compatible + # output format, not the v2 `_DOT_` convention). + # Pre-aggregation table matching is also bypassed in this mode, since the + # callers asking for raw rows are typically applying their own aggregation + # downstream. This is the v3 equivalent of v2's `preaggregate=False` mode. + aggregate: bool = True + # Temporal filter settings for incremental materialization # Dict mapping column refs to partition objects from cube (e.g., {"v3.date.date_id": partition_obj}) # When provided, temporal filters are pushed down to parent nodes via dimension links diff --git a/datajunction-server/tests/construction/build_v3/measures_sql_test.py b/datajunction-server/tests/construction/build_v3/measures_sql_test.py index 9be58e6c9..682aa0e12 100644 --- a/datajunction-server/tests/construction/build_v3/measures_sql_test.py +++ b/datajunction-server/tests/construction/build_v3/measures_sql_test.py @@ -10082,3 +10082,300 @@ async def test_count_distinct_with_right_outer_dim_join( assert metric_resp.json()["derived_expression"] == ( "COUNT( DISTINCT account_id)" ) + + +class TestMeasuresSQLAggregateFalse: + """ + Tests for the ``aggregate=False`` mode on /sql/measures/v3/ (the v3 + equivalent of v2's ``preaggregate=False``). + + When ``aggregate=False``: + - SQL is a flat SELECT with no GROUP BY + - Component columns project the raw inner expression under the exact + node column name (the v3-compatible output format — *not* the v2 + ``_DOT_`` convention) + - Pre-aggregation table matching is bypassed (the caller asks for raw + rows, which cannot be served from a pre-aggregated table) + - ``metric_formulas[].combiner`` is populated and references the raw + column name, so downstream callers know how to aggregate + - The default (``aggregate=True``) behavior is unchanged + + This mode targets callers that apply their own aggregation downstream + over the raw row-level output (e.g., ratio metrics computed in an outer + layer where DJ's server-side aggregation would double-count). + """ + + @pytest.mark.asyncio + async def test_binary_indicator_no_group_by(self, client_with_build_v3): + """ + Binary indicator metric (``SUM(is_product_view)`` over a 0/1 column) + with ``aggregate=False``: SQL must project the raw indicator, no + GROUP BY, and combiner must still be ``SUM()`` so the + caller can aggregate downstream. + + This is the regression test for the production over-count: with + ``aggregate=True``, server-side SUM + a downstream outer SUM would + double-count; with ``aggregate=False``, the raw 0/1 column is + projected and the caller applies SUM exactly once. + """ + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.product_view_count"], + "dimensions": ["v3.page_views_enriched.page_date"], + "aggregate": False, + }, + ) + + assert response.status_code == 200, response.json() + result = response.json() + assert len(result["grain_groups"]) == 1 + gg = result["grain_groups"][0] + + assert_sql_equal( + gg["sql"], + """ + WITH v3_page_views_enriched AS ( + SELECT page_date, + CASE WHEN page_type = 'product' THEN 1 ELSE 0 END + AS is_product_view + FROM default.v3.page_views + ) + SELECT t1.page_date, + t1.is_product_view is_product_view + FROM v3_page_views_enriched t1 + """, + ) + + # No GROUP BY anywhere in the rendered SQL. + assert "GROUP BY" not in gg["sql"].upper() + # The synthetic ``_sum_`` alias is NOT used; the raw node + # column name is projected instead. + assert "is_product_view_sum_" not in gg["sql"] + # The v3 output stays v3-shaped: no v2 ``_DOT_`` qualified aliases. + assert "_DOT_" not in gg["sql"] + assert "is_product_view" in gg["sql"] + + # Column metadata: dimension preserved, component projected under + # the raw node column name, semantic_entity preserved for internal + # identity matching. + assert gg["columns"] == [ + { + "name": "page_date", + "type": "int", + "semantic_entity": "v3.page_views_enriched.page_date", + "semantic_type": "dimension", + }, + { + "name": "is_product_view", + "type": "bigint", + "semantic_entity": ( + "v3.product_view_count:is_product_view_sum_eb3a4b41" + ), + "semantic_type": "metric_component", + }, + ] + + # Combiner must point at the raw column name so downstream callers + # know how to re-aggregate. + assert result["metric_formulas"] == [ + { + "name": "v3.product_view_count", + "short_name": "product_view_count", + "query": "SELECT SUM(is_product_view) FROM v3.page_views_enriched", + "combiner": "SUM(is_product_view)", + "components": ["is_product_view"], + "is_derived": False, + "parent_name": "v3.page_views_enriched", + }, + ] + + @pytest.mark.asyncio + async def test_additive_numeric_metric(self, client_with_build_v3): + """ + Additive numeric metric (``SUM(line_total)``) with ``aggregate=False``: + same flat-SELECT / raw-column shape as the binary indicator case. + """ + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.order_details.status"], + "aggregate": False, + }, + ) + + assert response.status_code == 200, response.json() + gg = get_first_grain_group(response.json()) + + assert_sql_equal( + gg["sql"], + """ + WITH v3_order_details AS ( + SELECT o.status, + oi.quantity * oi.unit_price AS line_total + FROM default.v3.orders o + JOIN default.v3.order_items oi + ON o.order_id = oi.order_id + ) + SELECT t1.status, + t1.line_total line_total + FROM v3_order_details t1 + """, + ) + + assert "GROUP BY" not in gg["sql"].upper() + assert "line_total_sum_" not in gg["sql"] + # v3 output stays v3-shaped: no v2 ``_DOT_`` qualified aliases. + assert "_DOT_" not in gg["sql"] + + # Combiner references the raw node column name. + formula = response.json()["metric_formulas"][0] + assert formula["combiner"] == "SUM(line_total)" + assert formula["components"] == ["line_total"] + + @pytest.mark.asyncio + async def test_multi_metric_preserves_each_combiner( + self, + client_with_build_v3, + ): + """ + Multi-metric request with ``aggregate=False``: each metric's combiner + is preserved and references its raw node column name. + """ + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": [ + "v3.total_revenue", + "v3.max_unit_price", + "v3.min_unit_price", + ], + "dimensions": [], + "aggregate": False, + }, + ) + + assert response.status_code == 200, response.json() + result = response.json() + + # All three metrics share the same parent fact, so a single grain + # group is emitted with three raw-column projections. + assert len(result["grain_groups"]) == 1 + gg = result["grain_groups"][0] + assert "GROUP BY" not in gg["sql"].upper() + # No v2 ``_DOT_`` qualified aliases in the rendered SQL. + assert "_DOT_" not in gg["sql"] + + # Each component is a raw column named after the node column. + component_names = {c["name"] for c in gg["components"]} + assert "line_total" in component_names + assert "unit_price" in component_names + + # Each metric_formula keeps its own combiner pointing at the raw + # column(s) it aggregates over. + formulas_by_name = {f["name"]: f for f in result["metric_formulas"]} + assert formulas_by_name["v3.total_revenue"]["combiner"] == ("SUM(line_total)") + assert formulas_by_name["v3.max_unit_price"]["combiner"] == ("MAX(unit_price)") + assert formulas_by_name["v3.min_unit_price"]["combiner"] == ("MIN(unit_price)") + + @pytest.mark.asyncio + async def test_default_aggregate_true_unchanged( + self, + client_with_build_v3, + ): + """ + Backward-compat guard: when ``aggregate`` is not passed (or passed + as True), the response is byte-for-byte identical to the + pre-existing behavior — GROUP BY emitted, synthetic + ``_sum_`` aliases, no v2-style ``_DOT_`` aliases. + """ + # Default (no aggregate param) + default_response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.product_view_count"], + "dimensions": ["v3.page_views_enriched.page_date"], + }, + ) + # Explicit aggregate=True + explicit_response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.product_view_count"], + "dimensions": ["v3.page_views_enriched.page_date"], + "aggregate": True, + }, + ) + + assert default_response.status_code == 200 + assert explicit_response.status_code == 200 + # Same shape either way. + assert default_response.json() == explicit_response.json() + + gg = get_first_grain_group(default_response.json()) + # Existing behavior: GROUP BY present, synthetic hash alias used, + # no v2-style alias. + assert "GROUP BY" in gg["sql"].upper() + assert "_DOT_" not in gg["sql"] + assert "is_product_view_sum_eb3a4b41" in gg["sql"] + + +class TestMeasuresSQLAggregateFalseEndpoint: + """Route-level integration test for ``aggregate=False`` on /sql/measures/v3/.""" + + @pytest.mark.asyncio + async def test_route_aggregate_false_full_response_shape( + self, + client_with_build_v3, + ): + """ + Integration test: call /sql/measures/v3/ with ``aggregate=False`` and + assert the top-level response shape is preserved (grain_groups + + metric_formulas) while every per-component field reflects the + raw-row semantics. + """ + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.product_view_count"], + "dimensions": ["v3.page_views_enriched.page_date"], + "aggregate": False, + }, + ) + + assert response.status_code == 200, response.json() + body = response.json() + + # Top-level shape preserved. + assert set(body.keys()) >= { + "grain_groups", + "metric_formulas", + "dialect", + "requested_dimensions", + } + assert body["dialect"] == "spark" + assert body["requested_dimensions"] == [ + "v3.page_views_enriched.page_date", + ] + + # Exactly one grain group, exactly one metric formula. + assert len(body["grain_groups"]) == 1 + assert len(body["metric_formulas"]) == 1 + + gg = body["grain_groups"][0] + # Raw-row semantics: no GROUP BY, the component is projected under + # its exact node column name (v3-compatible output), no synthetic + # hash alias or v2 ``_DOT_`` qualified alias appears. + assert "GROUP BY" not in gg["sql"].upper() + assert "_DOT_" not in gg["sql"] + assert gg["components"][0]["name"] == "is_product_view" + assert gg["components"][0]["expression"] == "is_product_view" + assert gg["components"][0]["aggregation"] == "SUM" + + # metric_formulas[].combiner is populated and points at the raw + # node column name for downstream re-aggregation. + formula = body["metric_formulas"][0] + assert formula["name"] == "v3.product_view_count" + assert formula["combiner"] == "SUM(is_product_view)" + assert formula["components"] == ["is_product_view"]