From 42e42e6dbba90c62ff038f3aa5a766a3080c86b5 Mon Sep 17 00:00:00 2001 From: Yian Shang Date: Tue, 1 Sep 2026 21:36:30 -0700 Subject: [PATCH 1/4] Don't merge components a grain group never accumulated Fixes #2489. A non-decomposable metric (a percentile, MAX_BY, ...) drags every metric sharing its parent down to a raw-grain grain group, so the CTE projects raw columns instead of applying each component's accumulate function. The final SELECT still applied the components' merge functions, which is only correct where accumulate and merge coincide: AVG(x) -> SUM(x) / SUM(x), silently 1.0 COUNT(x) -> SUM(x), adds up the values COUNT(*) -> SUM(alias.*), invalid SQL APPROX_COUNT_DISTINCT(x) -> sketch union over a raw column SUM/MIN/MAX -> unaffected Route those metrics down the branch that already serves genuinely non-decomposable metrics: emit the metric's own expression over the raw columns. The invariant is now carried explicitly as GrainGroupSQL.components_accumulated instead of being inferred from a component_aliases miss, which used to fall back to a raw column name and produce wrong SQL rather than failing. Also project each raw column once. Two components can read the same column (AVG needs SUM and COUNT), and a component can read a column the grain group already projects, both of which made references to it ambiguous. COUNT(*)'s "*" component is skipped entirely -- it reads no column, and projecting it re-projected every column of the parent. Adds a regression test per accumulate != merge shape, each paired with a control asserting the pre-aggregated form when the metric is requested alone, plus SUM alongside a percentile to pin the shape that was always correct. --- .../construction/build_v3/measures.py | 46 +- .../construction/build_v3/metrics.py | 124 +++++- .../construction/build_v3/types.py | 9 + .../tests/api/namespaces_test.py | 2 +- datajunction-server/tests/api/nodes_test.py | 3 + .../construction/build_v3/metrics_sql_test.py | 408 ++++++++++++++++++ datajunction-server/tests/examples.py | 36 +- 7 files changed, 591 insertions(+), 37 deletions(-) diff --git a/datajunction-server/datajunction_server/construction/build_v3/measures.py b/datajunction-server/datajunction_server/construction/build_v3/measures.py index f6912654c..c269df7d3 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/measures.py +++ b/datajunction-server/datajunction_server/construction/build_v3/measures.py @@ -2013,6 +2013,20 @@ def build_grain_group_sql( # This is needed for metrics SQL to correctly reference component columns component_aliases: dict[str, str] = {} + # Raw columns already projected by the NONE-aggregability path below. A NONE + # group projects its grain columns as themselves, so those are taken already. + # A COUNT(*) component reads no column at all, and projecting its "*" would + # re-project every column of the parent, so treat it as taken too. + projected_raw_columns: set[str] = ( + set(grain_group.grain_columns) | {"*"} + if grain_group.aggregability == Aggregability.NONE + else set() + ) + + # A NONE-aggregability group projects raw rows instead of applying each + # component's accumulate function, so the final SELECT can't merge them. + components_accumulated = grain_group.aggregability != Aggregability.NONE + for metric_node, component in grain_group.components: metrics_covered.add(metric_node.name) @@ -2024,18 +2038,27 @@ def build_grain_group_sql( # Collect unique components for API response unique_components.append(component) - # 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 - # don't go through this path - they're computed in generate_metrics_sql. - if grain_group.aggregability == Aggregability.NONE: # pragma: no cover - if component.expression: - col_ast = make_column_ref(component.expression) + # For NONE aggregability, output raw columns (no aggregation possible). + # Hit for BASE metrics with NONE aggregability (e.g., metrics with RANK() + # directly) and for a merged group dragged down to NONE by a + # non-decomposable metric (a percentile, MAX_BY, ...) on the same parent. + # Derived metrics with window functions don't go through this path - + # they're computed in generate_metrics_sql. + if grain_group.aggregability == Aggregability.NONE: + if component.expression: # pragma: no branch component_alias = component.expression - component_expressions.append((component_alias, col_ast)) - component_metadata.append( - (component_alias, component, metric_node), - ) + # Project the raw column once: components can share one (AVG(x) + # needs SUM(x) and COUNT(x), both reading x) and it may already + # be projected as a grain column. A second projection under the + # same name makes every reference to it ambiguous. + if component_alias not in projected_raw_columns: + projected_raw_columns.add(component_alias) + component_expressions.append( + (component_alias, make_column_ref(component.expression)), + ) + component_metadata.append( + (component_alias, component, metric_node), + ) component_aliases[component.name] = component_alias continue @@ -2339,6 +2362,7 @@ def build_grain_group_sql( metrics=list(metrics_covered), parent_name=grain_group.parent_node.name, component_aliases=component_aliases, + components_accumulated=components_accumulated, is_merged=grain_group.is_merged, component_aggregabilities=grain_group.component_aggregabilities, components=unique_components, diff --git a/datajunction-server/datajunction_server/construction/build_v3/metrics.py b/datajunction-server/datajunction_server/construction/build_v3/metrics.py index fb8cb3b28..e924fa888 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/metrics.py +++ b/datajunction-server/datajunction_server/construction/build_v3/metrics.py @@ -109,20 +109,42 @@ def classify_filters( return dimension_filters, metric_filters -def build_base_metric_expression( +def original_metric_expression( + ctx: BuildContext, decomposed: DecomposedMetricInfo, - cte_alias: str, - gg: GrainGroupSQL, -) -> tuple[ast.Expression, dict[str, tuple[str, str]]]: +) -> ast.Expression: """ - Build an expression AST for a base metric from a grain group. + Get a metric's own expression, as written in its query. - Always applies re-aggregation in the final SELECT using the component's - merge function (e.g., SUM for sums/counts, MIN for min, hll_union for HLL). + For a non-decomposable metric this is exactly its combiner (the decomposition + left the expression alone). For a decomposable one the combiner is written in + terms of merged components, so read the expression off the metric's query. - This is correct whether the CTE is at the exact requested grain or finer: - - If CTE is at requested grain: re-aggregation is a no-op (SUM of one = that one) - - If CTE is at finer grain: re-aggregation does actual work + Args: + ctx: Build context, used for its parsed query cache + decomposed: Decomposed metric info + + Returns: + Expression AST referencing the parent node's columns + """ + if not decomposed.components: + return deepcopy(decomposed.combiner_ast) + + projection = ctx.get_parsed_query(decomposed.metric_node).select.projection[0] + expr_ast = deepcopy(projection) + if isinstance(expr_ast, ast.Alias): # pragma: no cover + expr_ast = expr_ast.child + expr_ast.clear_parent() + return cast(ast.Expression, expr_ast) + + +def build_component_mappings( + decomposed: DecomposedMetricInfo, + cte_alias: str, + gg: GrainGroupSQL, +) -> dict[str, tuple[str, str]]: + """ + Map each of a metric's components to the grain group column holding it. Args: decomposed: Decomposed metric info with components and combiner @@ -130,12 +152,10 @@ def build_base_metric_expression( gg: The grain group SQL containing component metadata Returns: - Tuple of (expression AST, component column mappings) - The component mappings are {component_name: (cte_alias, column_name)} + {component_name: (cte_alias, column_name)} """ comp_mappings: dict[str, tuple[str, str]] = {} - # Build component -> column mappings # For merged: use component_aggregabilities to determine column source # For non-merged: use decomposed.aggregability for comp in decomposed.components: @@ -155,9 +175,48 @@ def build_base_metric_expression( grain_col = comp.rule.level[0] if comp.rule.level else comp.expression comp_mappings[comp.name] = (cte_alias, grain_col) else: - # FULL/NONE: use pre-aggregated column - actual_col = gg.component_aliases.get(comp.name, comp.name) - comp_mappings[comp.name] = (cte_alias, actual_col) # type: ignore + # FULL/NONE: use the column the grain group emitted for this component + actual_col = gg.component_aliases.get(comp.name) + if actual_col is None: # pragma: no cover + raise DJInvalidInputException( + f"Metric component {comp.name} has no column in grain group " + f"{gg.parent_name}, so it cannot be referenced.", + ) + comp_mappings[comp.name] = (cte_alias, actual_col) + + return comp_mappings + + +def build_base_metric_expression( + decomposed: DecomposedMetricInfo, + cte_alias: str, + gg: GrainGroupSQL, +) -> tuple[ast.Expression, dict[str, tuple[str, str]]]: + """ + Build an expression AST for a base metric from a pre-aggregated grain group. + + Applies re-aggregation in the final SELECT using the component's merge + function (e.g., SUM for sums/counts, MIN for min, hll_union for HLL). + + This is correct whether the CTE is at the exact requested grain or finer: + - If CTE is at requested grain: re-aggregation is a no-op (SUM of one = that one) + - If CTE is at finer grain: re-aggregation does actual work + + It requires the CTE to have applied each component's accumulate function, + i.e. ``gg.components_accumulated``. Callers must use the metric's own + expression otherwise -- merging unaccumulated columns is wrong wherever + accumulate and merge differ (AVG, COUNT, HLL sketches). + + Args: + decomposed: Decomposed metric info with components and combiner + cte_alias: Alias of the grain group CTE + gg: The grain group SQL containing component metadata + + Returns: + Tuple of (expression AST, component column mappings) + The component mappings are {component_name: (cte_alias, column_name)} + """ + comp_mappings = build_component_mappings(decomposed, cte_alias, gg) # Build the aggregation expression expr_ast = _build_metric_aggregation(decomposed, cte_alias, gg, comp_mappings) @@ -536,6 +595,7 @@ def build_dimension_projection( def process_base_metrics( + ctx: BuildContext, grain_groups: list[GrainGroupSQL], cte_aliases: list[str], decomposed_metrics: dict[str, DecomposedMetricInfo], @@ -546,9 +606,13 @@ def process_base_metrics( For each metric in each grain group: - Non-decomposable metrics (like MAX_BY): use original expression with CTE refs - - Decomposable metrics: build aggregation expression from components + - Decomposable metrics in a grain group that did not pre-aggregate: likewise, + use the original expression (the components were never accumulated, so + their merge functions can't be applied) + - Decomposable metrics otherwise: build aggregation expression from components Args: + ctx: Build context, for the metrics' original queries grain_groups: List of grain group SQLs cte_aliases: CTE aliases corresponding to each grain group decomposed_metrics: Decomposed metric info by metric name @@ -576,10 +640,15 @@ def process_base_metrics( # No decomposition info at all - skip this metric continue - if not decomposed.components: - # Non-decomposable metric (like MAX_BY) - use original expression - # with column references rewritten to point to grain group CTE - expr_ast: ast.Expression = deepcopy(decomposed.combiner_ast) + if not decomposed.components or not gg.components_accumulated: + # Two cases, one treatment: a non-decomposable metric (like MAX_BY), + # and a decomposable metric whose grain group projected raw rows + # rather than pre-aggregating (because a non-decomposable metric on + # the same parent dragged the group down to NONE aggregability). + # Either way the CTE holds raw columns, so apply the metric's own + # expression to them. The components' merge functions would be + # wrong here: AVG(x) would become SUM(x) / SUM(x). + expr_ast: ast.Expression = original_metric_expression(ctx, decomposed) # Rewrite column references in the expression to use the CTE alias # _table must be an ast.Table (not ast.Name) for proper stringification @@ -588,6 +657,18 @@ def process_base_metrics( if col.name and not col._table: # type: ignore # pragma: no branch col._table = cte_table # type: ignore + # Components still map to their raw columns, so that derived metrics + # combining them can be resolved. + if decomposed.components: + for comp_name, ( + comp_cte_alias, + comp_col, + ) in build_component_mappings(decomposed, alias, gg).items(): + component_refs[comp_name] = ColumnRef( + cte_alias=comp_cte_alias, + column_name=comp_col, + ) + metric_exprs[metric_name] = MetricExprInfo( expr_ast=expr_ast, short_name=short_name, @@ -1783,6 +1864,7 @@ def generate_metrics_sql( # Process base metrics from base grain groups only # Window grain groups are handled separately after the base_metrics CTE base_metrics_result = process_base_metrics( + ctx, base_grain_groups, cte_aliases, decomposed_metrics, diff --git a/datajunction-server/datajunction_server/construction/build_v3/types.py b/datajunction-server/datajunction_server/construction/build_v3/types.py index faac2e34c..ea26c2cbc 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/types.py +++ b/datajunction-server/datajunction_server/construction/build_v3/types.py @@ -346,6 +346,15 @@ class GrainGroupSQL: # instead of individual grain group CTEs. is_cross_fact_window: bool = False + # True when the grain group CTE built every component with its accumulate + # function (e.g. SUM(x), COUNT(x), hll_sketch_agg(x)). False when the CTE + # projects raw rows instead, which happens for NONE aggregability -- a + # non-decomposable metric (a percentile, MAX_BY, ...) sharing the parent + # drags the whole group down to raw grain. Applying a component's merge + # function in the final SELECT is only valid when this is True; otherwise + # the metric's own expression has to be applied to the raw columns. + components_accumulated: bool = True + # Pre-aggregation: True when collect_and_build_ctes() added a wrapper CTE that # applies COUNT(DISTINCT grain_key) per requested dimension combination. # When True, _build_metric_aggregation() should emit SUM(pre_agg_col) instead of diff --git a/datajunction-server/tests/api/namespaces_test.py b/datajunction-server/tests/api/namespaces_test.py index 36e8d8dde..e5a6b0a3f 100644 --- a/datajunction-server/tests/api/namespaces_test.py +++ b/datajunction-server/tests/api/namespaces_test.py @@ -204,7 +204,7 @@ async def test_list_all_namespaces( }, { "namespace": "v3", - "num_nodes": 47, + "num_nodes": 50, "github_repo_path": None, "git_branch": None, }, diff --git a/datajunction-server/tests/api/nodes_test.py b/datajunction-server/tests/api/nodes_test.py index 94c974d1a..54f588bf7 100644 --- a/datajunction-server/tests/api/nodes_test.py +++ b/datajunction-server/tests/api/nodes_test.py @@ -323,12 +323,15 @@ async def test_get_nodes_with_details(client_with_examples: AsyncClient): "v3.customer", "v3.customer_count", "v3.date", + "v3.line_item_count", "v3.location", "v3.max_unit_price", "v3.min_unit_price", "v3.mom_revenue_change", "v3.order_count", "v3.order_details", + "v3.order_line_rows", + "v3.p90_unit_price", "v3.page_view_count", "v3.page_views_enriched", "v3.pages_per_session", diff --git a/datajunction-server/tests/construction/build_v3/metrics_sql_test.py b/datajunction-server/tests/construction/build_v3/metrics_sql_test.py index 697457e2d..6ab46fd4d 100644 --- a/datajunction-server/tests/construction/build_v3/metrics_sql_test.py +++ b/datajunction-server/tests/construction/build_v3/metrics_sql_test.py @@ -6586,3 +6586,411 @@ async def test_metric_on_dimension_node_groups_and_filters_by_its_columns( GROUP BY product_0.category """, ) + + +class TestGrainGroupWithoutPreAggregation: + """ + Metrics requested alongside a non-decomposable metric on the same parent. + + The non-decomposable metric (here the ``v3.p90_unit_price`` percentile) drags + the whole grain group down to raw grain, so its CTE never applies any + component's accumulate function. Each decomposable metric must therefore + fall back to its own expression: applying the components' merge functions to + raw columns is wrong for every shape where accumulate != merge. + + Each metric is asserted twice -- alone (pre-aggregated, the control) and + beside the percentile -- since the bug is invisible for SUM. + """ + + @pytest.mark.asyncio + async def test_avg_alone_pre_aggregates(self, client_with_build_v3): + """AVG alone: SUM and COUNT components, re-aggregated by their merges.""" + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.avg_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT o.status, oi.unit_price + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT + t1.status, + COUNT(t1.unit_price) unit_price_count_55cff00f, + SUM(t1.unit_price) unit_price_sum_55cff00f + FROM v3_order_details t1 + GROUP BY t1.status + ) + SELECT + order_details_0.status AS status, + SUM(order_details_0.unit_price_sum_55cff00f) + / NULLIF(SUM(order_details_0.unit_price_count_55cff00f), 0) + AS avg_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_avg_with_percentile(self, client_with_build_v3): + """ + AVG beside a percentile: emit AVG(unit_price) over the raw rows. + + Merging the components instead would give SUM(unit_price) / + SUM(unit_price) -- numerator and denominator read the same raw column, + so the metric would silently return 1.0. + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.avg_unit_price", "v3.p90_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT o.order_id, oi.line_number, o.status, oi.unit_price + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT + t1.status, + t1.line_number, + t1.order_id, + t1.unit_price unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + AVG(order_details_0.unit_price) AS avg_unit_price, + PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_count_alone_pre_aggregates(self, client_with_build_v3): + """COUNT alone: accumulate with COUNT in the CTE, merge with SUM.""" + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.line_item_count"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT oi.line_number, o.status + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT t1.status, COUNT(t1.line_number) line_number_count_f137b827 + FROM v3_order_details t1 + GROUP BY t1.status + ) + SELECT + order_details_0.status AS status, + SUM(order_details_0.line_number_count_f137b827) AS line_item_count + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_count_with_percentile(self, client_with_build_v3): + """ + COUNT beside a percentile: emit COUNT(line_number) over the raw rows. + + Merging instead would give SUM(line_number), which adds up line numbers + rather than counting rows. + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.line_item_count", "v3.p90_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT o.order_id, oi.line_number, o.status, oi.unit_price + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT t1.status, t1.line_number, t1.order_id, t1.unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + COUNT(order_details_0.line_number) AS line_item_count, + PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_count_star_alone_pre_aggregates(self, client_with_build_v3): + """COUNT(*) alone: COUNT(*) in the CTE, merged with SUM.""" + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.order_line_rows"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT o.status + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT t1.status, COUNT(*) count_5041f1a8 + FROM v3_order_details t1 + GROUP BY t1.status + ) + SELECT + order_details_0.status AS status, + SUM(order_details_0.count_5041f1a8) AS order_line_rows + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_count_star_with_percentile(self, client_with_build_v3): + """ + COUNT(*) beside a percentile: emit COUNT(*) over the raw rows. + + Merging instead would give SUM(order_details_0.*), which isn't valid + SQL, and the CTE would re-project every parent column via the "*" + component -- making the dimension references ambiguous too. + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.order_line_rows", "v3.p90_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT o.order_id, oi.line_number, o.status, oi.unit_price + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT t1.status, t1.line_number, t1.order_id, t1.unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + COUNT(*) AS order_line_rows, + PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_approx_count_distinct_alone_pre_aggregates( + self, + client_with_build_v3, + ): + """APPROX_COUNT_DISTINCT alone: HLL sketches built, then union-merged.""" + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.customer_count"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT o.customer_id, o.status + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT + t1.status, + hll_sketch_agg(t1.customer_id) customer_id_hll_23002251 + FROM v3_order_details t1 + GROUP BY t1.status + ) + SELECT + order_details_0.status AS status, + hll_sketch_estimate( + hll_union_agg(order_details_0.customer_id_hll_23002251) + ) AS customer_count + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_approx_count_distinct_with_percentile(self, client_with_build_v3): + """ + APPROX_COUNT_DISTINCT beside a percentile: emit the original + APPROX_COUNT_DISTINCT over the raw rows. + + Merging instead would union sketches that were never built -- + hll_union_agg applied to a raw customer_id column. + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.customer_count", "v3.p90_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT + o.order_id, + oi.line_number, + o.customer_id, + o.status, + oi.unit_price + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT + t1.status, + t1.line_number, + t1.order_id, + t1.customer_id customer_id, + t1.unit_price unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + APPROX_COUNT_DISTINCT(order_details_0.customer_id) AS customer_count, + PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_sum_with_percentile_is_unchanged(self, client_with_build_v3): + """ + SUM beside a percentile: still SUM(line_total) over the raw rows. + + SUM accumulates and merges with the same function, so this shape was + never broken -- which is exactly why one test wouldn't have caught it. + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.total_revenue", "v3.p90_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT + o.order_id, + oi.line_number, + o.status, + oi.unit_price, + 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 + ), + order_details_0 AS ( + SELECT + t1.status, + t1.line_number, + t1.order_id, + t1.line_total line_total, + t1.unit_price unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + SUM(order_details_0.line_total) AS total_revenue, + PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_percentile_alone(self, client_with_build_v3): + """The percentile on its own: raw rows, expression applied downstream.""" + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.p90_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT o.order_id, oi.line_number, o.status, oi.unit_price + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT t1.status, t1.order_id, t1.line_number, t1.unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) diff --git a/datajunction-server/tests/examples.py b/datajunction-server/tests/examples.py index f6dc73e51..0c4aeeb48 100644 --- a/datajunction-server/tests/examples.py +++ b/datajunction-server/tests/examples.py @@ -3329,10 +3329,38 @@ "mode": "published", }, ), - # Note: NONE aggregability metrics (e.g., MEDIAN) cannot be added currently because - # the MEDIAN function class in DJ doesn't have is_aggregation = True, which causes - # metric validation to fail. This should be fixed in functions.py. - # TODO: Add NONE aggregability metric once MEDIAN is properly registered as aggregate. + ( + "/nodes/metric/", + { + "name": "v3.line_item_count", + "description": "Count of order line items (COUNT accumulate, SUM merge)", + "query": "SELECT COUNT(line_number) FROM v3.order_details", + "mode": "published", + }, + ), + ( + "/nodes/metric/", + { + "name": "v3.order_line_rows", + "description": "Count of rows (COUNT(*), which reads no column)", + "query": "SELECT COUNT(*) FROM v3.order_details", + "mode": "published", + }, + ), + # A percentile has no decomposition at all, so it forces the whole grain + # group off the pre-aggregation path. + ( + "/nodes/metric/", + { + "name": "v3.p90_unit_price", + "description": "90th percentile unit price (non-decomposable percentile)", + "query": "SELECT PERCENTILE(unit_price, 0.9) FROM v3.order_details", + "mode": "published", + }, + ), + # Note: MEDIAN is still unusable here because the MEDIAN function class in DJ + # doesn't have is_aggregation = True, which causes metric validation to fail. + # This should be fixed in functions.py. # ========================================================================= # Base Metrics - On page_views_enriched # ========================================================================= From 481779eaa7f0743ac0e514dcdefbf3efb124defc Mon Sep 17 00:00:00 2001 From: Yian Shang Date: Tue, 1 Sep 2026 23:39:48 -0700 Subject: [PATCH 2/4] Make APPROX_PERCENTILE's accuracy argument actually optional approx_percentile(col, percentage) dispatched to the right infer_type overload and then died calling it: the accuracy parameter was annotated `ct.NumberType | None` but had no default, so a two-argument call raised `TypeError: infer_type() missing 1 required positional argument`. Creating a metric like `SELECT APPROX_PERCENTILE(price, 0.9) FROM ...` surfaced that as "Unknown TypeError on column ...". Give accuracy the `= None` default the other optional parameters in this module use. Also widen the scalar overload's percentage from FloatType to NumberType, matching PERCENTILE, so a DOUBLE percentage is accepted wherever a FLOAT one is. --- .../datajunction_server/sql/functions.py | 6 ++--- .../tests/sql/functions_test.py | 27 +++++++++++++++++++ 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/datajunction-server/datajunction_server/sql/functions.py b/datajunction-server/datajunction_server/sql/functions.py index b194d46e4..d1e3119b1 100644 --- a/datajunction-server/datajunction_server/sql/functions.py +++ b/datajunction-server/datajunction_server/sql/functions.py @@ -552,7 +552,7 @@ class ApproxPercentile(Function): def infer_type( col: ct.NumberType, percentage: ct.ListType, - accuracy: ct.NumberType | None, + accuracy: ct.NumberType | None = None, ) -> ct.DoubleType: return ct.ListType(element_type=col.type) # type: ignore @@ -560,8 +560,8 @@ def infer_type( @ApproxPercentile.register def infer_type( col: ct.NumberType, - percentage: ct.FloatType, - accuracy: ct.NumberType | None, + percentage: ct.NumberType, + accuracy: ct.NumberType | None = None, ) -> ct.NumberType: return col.type # type: ignore diff --git a/datajunction-server/tests/sql/functions_test.py b/datajunction-server/tests/sql/functions_test.py index 10a1290df..3fda4b889 100644 --- a/datajunction-server/tests/sql/functions_test.py +++ b/datajunction-server/tests/sql/functions_test.py @@ -288,6 +288,33 @@ async def test_approx_percentile(session: AsyncSession): assert not exc.errors assert query_with_list.select.projection[0].type == ct.FloatType() # type: ignore + # The accuracy argument is optional, in both the scalar and the list form + query_without_accuracy = parse("SELECT approx_percentile(10.0, 0.5)") + exc = DJException() + ctx = ast.CompileContext(session=session, exception=exc) + await query_without_accuracy.compile(ctx) + assert not exc.errors + assert query_without_accuracy.select.projection[0].type == ct.FloatType() # type: ignore + + query_without_accuracy = parse("SELECT approx_percentile(10.0, array(0.5, 0.9))") + exc = DJException() + ctx = ast.CompileContext(session=session, exception=exc) + await query_without_accuracy.compile(ctx) + assert not exc.errors + assert query_without_accuracy.select.projection[0].type == ct.ListType( # type: ignore + element_type=ct.FloatType(), + ) + + # A DOUBLE percentage is as good as a FLOAT one + query_double_percentage = parse( + "SELECT approx_percentile(10.0, CAST(0.5 AS DOUBLE))", + ) + exc = DJException() + ctx = ast.CompileContext(session=session, exception=exc) + await query_double_percentage.compile(ctx) + assert not exc.errors + assert query_double_percentage.select.projection[0].type == ct.FloatType() # type: ignore + @pytest.mark.asyncio async def test_array(session: AsyncSession): From 141e377e7f2d543f0d1462e67025d302bbf40fc2 Mon Sep 17 00:00:00 2001 From: Yian Shang Date: Tue, 1 Sep 2026 23:40:00 -0700 Subject: [PATCH 3/4] Cover derived metrics over a grain group that didn't pre-aggregate Derived metrics are built by inlining their base metrics' expressions (build_intermediate_metric_expr), so the fix for #2489 already carries them: v3.revenue_per_customer beside a percentile now divides by APPROX_COUNT_DISTINCT(customer_id) instead of unioning HLL sketches that were never built. Add the tests that pin this, and say so where the inlining happens -- it is no longer only about COUNT DISTINCT from _agg CTEs, it is what keeps the merge functions out of a raw-grain group. Tests cover a derived metric whose components are self-merging (v3.avg_order_value: SUM over COUNT DISTINCT, correct before and after -- its expression is byte-identical, only the grain group CTE lost the duplicate order_id projection) and one whose components are not (v3.revenue_per_customer), each alone and beside a percentile. The percentile fixture is now APPROX_PERCENTILE, the function that motivated the issue, and a PERCENTILE metric stays as a second non-decomposable aggregation so the behavior is visibly not tied to one function. Also assert an APPROX_PERCENTILE metric with an explicit accuracy argument can be created and builds SQL. --- .../construction/build_v3/metrics.py | 5 + .../tests/api/namespaces_test.py | 2 +- datajunction-server/tests/api/nodes_test.py | 1 + .../construction/build_v3/metrics_sql_test.py | 326 +++++++++++++++++- datajunction-server/tests/examples.py | 16 +- 5 files changed, 339 insertions(+), 11 deletions(-) diff --git a/datajunction-server/datajunction_server/construction/build_v3/metrics.py b/datajunction-server/datajunction_server/construction/build_v3/metrics.py index e924fa888..0e9ba46fe 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/metrics.py +++ b/datajunction-server/datajunction_server/construction/build_v3/metrics.py @@ -1217,6 +1217,11 @@ def process_derived_metrics( # COUNT DISTINCT base metrics from pre-aggregated (_agg) CTEs, where # the combiner_ast would produce COUNT(DISTINCT ...) but the pre-built # expression already uses MAX(...) as the correct passthrough aggregation. + # It is also what keeps a derived metric correct when its grain group + # didn't pre-aggregate (see GrainGroupSQL.components_accumulated): the + # inlined base expressions are the metrics' own expressions over raw + # columns, whereas combiner_ast would merge components that were never + # accumulated. # Fall back to build_derived_metric_expr when some base metrics are missing # from base_metric_exprs (e.g., NONE-aggregability metrics not in any grain group). expr_ast = ( # type: ignore[assignment] diff --git a/datajunction-server/tests/api/namespaces_test.py b/datajunction-server/tests/api/namespaces_test.py index e5a6b0a3f..95ec83792 100644 --- a/datajunction-server/tests/api/namespaces_test.py +++ b/datajunction-server/tests/api/namespaces_test.py @@ -204,7 +204,7 @@ async def test_list_all_namespaces( }, { "namespace": "v3", - "num_nodes": 50, + "num_nodes": 51, "github_repo_path": None, "git_branch": None, }, diff --git a/datajunction-server/tests/api/nodes_test.py b/datajunction-server/tests/api/nodes_test.py index 54f588bf7..1b096cbe0 100644 --- a/datajunction-server/tests/api/nodes_test.py +++ b/datajunction-server/tests/api/nodes_test.py @@ -326,6 +326,7 @@ async def test_get_nodes_with_details(client_with_examples: AsyncClient): "v3.line_item_count", "v3.location", "v3.max_unit_price", + "v3.median_unit_price", "v3.min_unit_price", "v3.mom_revenue_change", "v3.order_count", diff --git a/datajunction-server/tests/construction/build_v3/metrics_sql_test.py b/datajunction-server/tests/construction/build_v3/metrics_sql_test.py index 6ab46fd4d..ebbec58b5 100644 --- a/datajunction-server/tests/construction/build_v3/metrics_sql_test.py +++ b/datajunction-server/tests/construction/build_v3/metrics_sql_test.py @@ -6600,6 +6600,9 @@ class TestGrainGroupWithoutPreAggregation: Each metric is asserted twice -- alone (pre-aggregated, the control) and beside the percentile -- since the bug is invisible for SUM. + + Derived metrics are covered too: they inline their base metrics' + expressions, so they inherit whatever the base metrics do. """ @pytest.mark.asyncio @@ -6677,7 +6680,7 @@ async def test_avg_with_percentile(self, client_with_build_v3): SELECT order_details_0.status AS status, AVG(order_details_0.unit_price) AS avg_unit_price, - PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + APPROX_PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price FROM order_details_0 GROUP BY order_details_0.status """, @@ -6748,7 +6751,7 @@ async def test_count_with_percentile(self, client_with_build_v3): SELECT order_details_0.status AS status, COUNT(order_details_0.line_number) AS line_item_count, - PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + APPROX_PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price FROM order_details_0 GROUP BY order_details_0.status """, @@ -6820,7 +6823,7 @@ async def test_count_star_with_percentile(self, client_with_build_v3): SELECT order_details_0.status AS status, COUNT(*) AS order_line_rows, - PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + APPROX_PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price FROM order_details_0 GROUP BY order_details_0.status """, @@ -6909,7 +6912,7 @@ async def test_approx_count_distinct_with_percentile(self, client_with_build_v3) SELECT order_details_0.status AS status, APPROX_COUNT_DISTINCT(order_details_0.customer_id) AS customer_count, - PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + APPROX_PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price FROM order_details_0 GROUP BY order_details_0.status """, @@ -6957,7 +6960,7 @@ async def test_sum_with_percentile_is_unchanged(self, client_with_build_v3): SELECT order_details_0.status AS status, SUM(order_details_0.line_total) AS total_revenue, - PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + APPROX_PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price FROM order_details_0 GROUP BY order_details_0.status """, @@ -6989,7 +6992,318 @@ async def test_percentile_alone(self, client_with_build_v3): ) SELECT order_details_0.status AS status, - PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + APPROX_PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_avg_with_a_different_percentile_function( + self, + client_with_build_v3, + ): + """ + AVG beside PERCENTILE rather than APPROX_PERCENTILE. + + Nothing here is percentile-specific, let alone specific to one percentile + function: any aggregation with no registered decomposition takes the + grain group off the pre-aggregation path. + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.avg_unit_price", "v3.median_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT o.order_id, oi.line_number, o.status, oi.unit_price + FROM default.v3.orders o + JOIN default.v3.order_items oi ON o.order_id = oi.order_id + ), + order_details_0 AS ( + SELECT + t1.status, + t1.line_number, + t1.order_id, + t1.unit_price unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + AVG(order_details_0.unit_price) AS avg_unit_price, + PERCENTILE(order_details_0.unit_price, 0.5) AS median_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_derived_of_self_merging_bases_alone(self, client_with_build_v3): + """ + A derived metric over self-merging bases, pre-aggregated (control). + + ``v3.avg_order_value`` is total_revenue / order_count: a SUM component + (merged with SUM) over a COUNT DISTINCT (which has no merge at all). + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.avg_order_value"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT o.order_id, 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 + ), + order_details_0 AS ( + SELECT + t1.status, + t1.order_id, + SUM(t1.line_total) line_total_sum_e1f61696 + FROM v3_order_details t1 + GROUP BY t1.status, t1.order_id + ) + SELECT + order_details_0.status AS status, + SUM(order_details_0.line_total_sum_e1f61696) + / NULLIF(COUNT(DISTINCT order_details_0.order_id), 0) + AS avg_order_value + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_derived_of_self_merging_bases_with_percentile( + self, + client_with_build_v3, + ): + """ + The same derived metric beside a percentile. + + Derived metrics inline their base metrics' expressions, so this follows + whatever the base metrics do. SUM and COUNT DISTINCT survive being read + off raw rows, so the metric expression here is exactly what it was before + the fix -- only the grain group CTE changed, which no longer projects + order_id a second time under its own name. + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.avg_order_value", "v3.p90_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT + o.order_id, + oi.line_number, + o.status, + oi.unit_price, + 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 + ), + order_details_0 AS ( + SELECT + t1.status, + t1.line_number, + t1.order_id, + t1.line_total line_total, + t1.unit_price unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + SUM(order_details_0.line_total) + / NULLIF(COUNT(DISTINCT order_details_0.order_id), 0) + AS avg_order_value, + APPROX_PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_derived_of_sketch_base_alone(self, client_with_build_v3): + """ + A derived metric over a base metric that is not self-merging (control). + + ``v3.revenue_per_customer`` is total_revenue / customer_count, and + customer_count is an HLL sketch: accumulate hll_sketch_agg, merge + hll_union_agg. + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.revenue_per_customer"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT + o.customer_id, + 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 + ), + order_details_0 AS ( + SELECT + t1.status, + hll_sketch_agg(t1.customer_id) customer_id_hll_23002251, + SUM(t1.line_total) line_total_sum_e1f61696 + FROM v3_order_details t1 + GROUP BY t1.status + ) + SELECT + order_details_0.status AS status, + SUM(order_details_0.line_total_sum_e1f61696) + / NULLIF( + hll_sketch_estimate( + hll_union_agg(order_details_0.customer_id_hll_23002251) + ), + 0 + ) AS revenue_per_customer + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_derived_of_sketch_base_with_percentile(self, client_with_build_v3): + """ + The same derived metric beside a percentile. + + The denominator has to become APPROX_COUNT_DISTINCT over the raw column. + Before the fix it inlined the base metric's merge form and unioned + sketches that were never built. + """ + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.revenue_per_customer", "v3.p90_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT + o.order_id, + oi.line_number, + o.customer_id, + o.status, + oi.unit_price, + 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 + ), + order_details_0 AS ( + SELECT + t1.status, + t1.line_number, + t1.order_id, + t1.customer_id customer_id, + t1.line_total line_total, + t1.unit_price unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + SUM(order_details_0.line_total) + / NULLIF(APPROX_COUNT_DISTINCT(order_details_0.customer_id), 0) + AS revenue_per_customer, + APPROX_PERCENTILE(order_details_0.unit_price, 0.9) AS p90_unit_price + FROM order_details_0 + GROUP BY order_details_0.status + """, + ) + + @pytest.mark.asyncio + async def test_approx_percentile_metric_with_accuracy(self, client_with_build_v3): + """ + An APPROX_PERCENTILE metric can be created, including with the optional + accuracy argument, and builds SQL as a non-decomposable metric. + """ + response = await client_with_build_v3.post( + "/nodes/metric/", + json={ + "name": "v3.p99_unit_price", + "description": "99th percentile unit price, with explicit accuracy", + "query": ( + "SELECT APPROX_PERCENTILE(unit_price, 0.99, 1000) " + "FROM v3.order_details" + ), + "mode": "published", + }, + ) + assert response.status_code == 201, response.json() + assert response.json()["status"] == "valid" + + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.total_revenue", "v3.p99_unit_price"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + + assert_sql_equal( + response.json()["sql"], + """ + WITH v3_order_details AS ( + SELECT + o.order_id, + oi.line_number, + o.status, + oi.unit_price, + 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 + ), + order_details_0 AS ( + SELECT + t1.status, + t1.line_number, + t1.order_id, + t1.line_total line_total, + t1.unit_price unit_price + FROM v3_order_details t1 + ) + SELECT + order_details_0.status AS status, + SUM(order_details_0.line_total) AS total_revenue, + APPROX_PERCENTILE(order_details_0.unit_price, 0.99, 1000) + AS p99_unit_price FROM order_details_0 GROUP BY order_details_0.status """, diff --git a/datajunction-server/tests/examples.py b/datajunction-server/tests/examples.py index 0c4aeeb48..bbe81cd88 100644 --- a/datajunction-server/tests/examples.py +++ b/datajunction-server/tests/examples.py @@ -3354,13 +3354,21 @@ { "name": "v3.p90_unit_price", "description": "90th percentile unit price (non-decomposable percentile)", - "query": "SELECT PERCENTILE(unit_price, 0.9) FROM v3.order_details", + "query": "SELECT APPROX_PERCENTILE(unit_price, 0.9) FROM v3.order_details", + "mode": "published", + }, + ), + # A second, differently-named non-decomposable aggregation, to show the + # behavior belongs to "has no decomposition" rather than to one function. + ( + "/nodes/metric/", + { + "name": "v3.median_unit_price", + "description": "Median unit price (non-decomposable percentile)", + "query": "SELECT PERCENTILE(unit_price, 0.5) FROM v3.order_details", "mode": "published", }, ), - # Note: MEDIAN is still unusable here because the MEDIAN function class in DJ - # doesn't have is_aggregation = True, which causes metric validation to fail. - # This should be fixed in functions.py. # ========================================================================= # Base Metrics - On page_views_enriched # ========================================================================= From 3b294fb93d4b0aeb44ece7a7974fc111ecca09a8 Mon Sep 17 00:00:00 2001 From: Yian Shang Date: Wed, 2 Sep 2026 01:32:24 -0700 Subject: [PATCH 4/4] Stop generated SQL depending on the order rows came back in find_upstream_node_names ends its recursive CTE with SELECT DISTINCT and no ORDER BY, so parent_map's per-child list order is whatever the plan produced. For a derived metric that order decides which base metric resolves first, and with it the order their components are projected in the grain group CTE -- so the same request produced different SQL in CI than locally. Reversing every parent list reproduces CI's output exactly; with this change the SQL is byte-identical either way. Order the CTE, and sort the base metrics by name in get_base_metrics_for_derived so component order is a property of the metrics rather than of the graph walk that reached them. Sorting grain_group.components by name would also work, but it reorders thirteen existing expectations and replaces requested-metric order with alphabetical-by-hash in the measures response; this way nothing else moves. find_join_paths_batch had the same exposure with more at stake: row order set the frontier order and broke ties for a (source, dimension, role) key, where the first path found is the one kept. Two same-depth routes to one dimension sharing a role path would therefore join along whichever route the query happened to return first. Order those queries too, so the lowest dimension link id wins consistently. The new tests perturb parent_map rather than pinning a string: they build the SQL twice, once with every parent list reversed, and assert the two are identical. --- .../construction/build_v3/decomposition.py | 6 +- .../construction/build_v3/loaders.py | 8 ++ .../construction/build_v3/metrics_sql_test.py | 85 +++++++++++++++++++ 3 files changed, 98 insertions(+), 1 deletion(-) diff --git a/datajunction-server/datajunction_server/construction/build_v3/decomposition.py b/datajunction-server/datajunction_server/construction/build_v3/decomposition.py index 8a3c1b56f..53db1d9f8 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/decomposition.py +++ b/datajunction-server/datajunction_server/construction/build_v3/decomposition.py @@ -250,7 +250,10 @@ def get_base_metrics_for_derived(ctx: BuildContext, metric_node: Node) -> list[N For a derived metric, get all the base metrics it depends on. Returns list of base metric nodes (metrics that SELECT FROM a fact/transform - or a dimension source, not other metrics). + or a dimension source, not other metrics), ordered by name. The order decides + the order the base metrics' components are projected in the grain group CTE, + so it has to be a property of the metrics rather than of the order the graph + walk happened to reach them. """ base_metrics = [] visited = set() @@ -278,6 +281,7 @@ def collect_bases(node: Node): base_metrics.append(node) collect_bases(metric_node) + base_metrics.sort(key=lambda node: node.name) return base_metrics diff --git a/datajunction-server/datajunction_server/construction/build_v3/loaders.py b/datajunction-server/datajunction_server/construction/build_v3/loaders.py index 934bf069b..2f6d54e7c 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/loaders.py +++ b/datajunction-server/datajunction_server/construction/build_v3/loaders.py @@ -176,7 +176,11 @@ async def find_upstream_node_names( AND parent_n.current_version = parent_nr.version WHERE parent_n.deactivated_at IS NULL ) + -- Ordered because parent_map's list order is load-bearing: it decides + -- the order a derived metric's base metrics resolve in, and with them + -- the order their components are projected in the generated SQL. SELECT DISTINCT node_name, child_name FROM upstream + ORDER BY node_name, child_name """).bindparams(bindparam("starting_names", expanding=True)) result = await session.execute( @@ -240,6 +244,7 @@ async def find_join_paths_batch( SELECT nr.id as rev_id, nr.node_id FROM noderevision nr WHERE nr.id IN :source_revision_ids + ORDER BY nr.id """).bindparams(bindparam("source_revision_ids", expanding=True)) init_result = await session.execute( @@ -290,6 +295,9 @@ async def find_join_paths_batch( JOIN node n ON dl.dimension_id = n.id WHERE nr.node_id IN :node_ids AND nr.version = (SELECT current_version FROM node WHERE id = nr.node_id) + -- Ordered so that when two links reach the same dimension with the + -- same role, the first one found -- and kept -- is always the same. + ORDER BY nr.node_id, dl.id """).bindparams(bindparam("node_ids", expanding=True)) try: diff --git a/datajunction-server/tests/construction/build_v3/metrics_sql_test.py b/datajunction-server/tests/construction/build_v3/metrics_sql_test.py index ebbec58b5..b1810e9b3 100644 --- a/datajunction-server/tests/construction/build_v3/metrics_sql_test.py +++ b/datajunction-server/tests/construction/build_v3/metrics_sql_test.py @@ -1,5 +1,7 @@ import pytest +from datajunction_server.construction.build_v3 import loaders + from . import assert_sql_equal @@ -7308,3 +7310,86 @@ async def test_approx_percentile_metric_with_accuracy(self, client_with_build_v3 GROUP BY order_details_0.status """, ) + + +class TestGeneratedSQLIsOrderStable: + """ + Generated SQL must not depend on the order rows came back in. + + ``find_upstream_node_names`` builds parent_map from a recursive CTE, and for + a derived metric that map decides the order its base metrics are resolved -- + and therefore the order their components are projected in the grain group + CTE. Reversing every parent list simulates a different row order (a + different query plan, a different engine) and must not change the SQL. + """ + + @staticmethod + async def _build_sql(client, monkeypatch, metrics, dimensions, reverse_parents): + original = loaders.find_upstream_node_names + + async def ordered_differently(session, starting_node_names): + all_names, parent_map = await original(session, starting_node_names) + return all_names, {k: list(reversed(v)) for k, v in parent_map.items()} + + if reverse_parents: + monkeypatch.setattr( + loaders, + "find_upstream_node_names", + ordered_differently, + ) + response = await client.get( + "/sql/metrics/v3/", + params={"metrics": metrics, "dimensions": dimensions}, + ) + assert response.status_code == 200, response.json() + return response.json()["sql"] + + @pytest.mark.asyncio + async def test_derived_metric_sql_is_parent_order_independent( + self, + client_with_build_v3, + monkeypatch, + ): + """A derived metric over two base metrics on the same fact.""" + metrics = ["v3.revenue_per_customer"] + dimensions = ["v3.order_details.status"] + expected = await self._build_sql( + client_with_build_v3, + monkeypatch, + metrics, + dimensions, + reverse_parents=False, + ) + reordered = await self._build_sql( + client_with_build_v3, + monkeypatch, + metrics, + dimensions, + reverse_parents=True, + ) + assert reordered == expected + + @pytest.mark.asyncio + async def test_cross_fact_derived_metric_sql_is_parent_order_independent( + self, + client_with_build_v3, + monkeypatch, + ): + """Two derived metrics whose base metrics span two facts.""" + metrics = ["v3.conversion_rate", "v3.revenue_per_visitor"] + dimensions = ["v3.product.category"] + expected = await self._build_sql( + client_with_build_v3, + monkeypatch, + metrics, + dimensions, + reverse_parents=False, + ) + reordered = await self._build_sql( + client_with_build_v3, + monkeypatch, + metrics, + dimensions, + reverse_parents=True, + ) + assert reordered == expected