Fix accumulate / merge

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
 
 

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:

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,

datajunction-server/datajunction_server/construction/build_v3/metrics.py

@@ -109,33 +109,53 @@ 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
         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)}
+        {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,

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

datajunction-server/datajunction_server/sql/functions.py

@@ -552,16 +552,16 @@ 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
 
 
 @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
 

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,
         },

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",