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/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..0e9ba46fe 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, @@ -1136,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] @@ -1783,6 +1869,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/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/api/namespaces_test.py b/datajunction-server/tests/api/namespaces_test.py index 36e8d8dde..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": 47, + "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 94c974d1a..1b096cbe0 100644 --- a/datajunction-server/tests/api/nodes_test.py +++ b/datajunction-server/tests/api/nodes_test.py @@ -323,12 +323,16 @@ 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.median_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..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 @@ -6586,3 +6588,808 @@ 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. + + Derived metrics are covered too: they inline their base metrics' + expressions, so they inherit whatever the base metrics do. + """ + + @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, + 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_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, + 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_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, + 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_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, + 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_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, + 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_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, + 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 + """, + ) + + +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 diff --git a/datajunction-server/tests/examples.py b/datajunction-server/tests/examples.py index f6dc73e51..bbe81cd88 100644 --- a/datajunction-server/tests/examples.py +++ b/datajunction-server/tests/examples.py @@ -3329,10 +3329,46 @@ "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 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", + }, + ), # ========================================================================= # Base Metrics - On page_views_enriched # ========================================================================= 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):