Add aggregate flag to /sql/measures/v3/ for raw-row output

datajunction-server/datajunction_server/api/sql.py

@@ -189,6 +189,20 @@ async def get_measures_sql_v3(
         ),
     ),
     use_materialized: bool = Query(True),
+    aggregate: bool = Query(
+        True,
+        description=(
+            "Whether to apply server-side aggregation at the requested grain "
+            "(default: True). When False, the response SQL is a flat SELECT "
+            "with no GROUP BY and projects each component column under its "
+            "exact node column name (the v3-compatible output format, not "
+            "v2's `<node>_DOT_<column>` convention). Pre-aggregation table "
+            "matching is bypassed. Use this when the caller applies its own "
+            "aggregation downstream. The `metric_formulas[].combiner` field "
+            "is still populated in either mode so the caller can re-aggregate "
+            "the raw output. Semantic parity with v2's `preaggregate=False`."
+        ),
+    ),
     dialect: Dialect = Query(
         Dialect.SPARK,
         description="SQL dialect for the generated query.",
@@ -243,6 +257,10 @@ async def get_measures_sql_v3(
         use_materialized: If True (default), use materialized tables when available.
             Set to False when generating SQL for materialization refresh to avoid
             circular references.
+        aggregate: If True (default), apply server-side aggregation. When False,
+            emit a flat SELECT with no GROUP BY, project each component column
+            under its exact node column name, and bypass pre-aggregation table
+            matching.
         include_temporal_filters: If True, checks if metrics+dimensions resolve to
             a cube with temporal partitions, and applies partition filters if so.
         lookback_window: Lookback window for temporal filters when applicable.
@@ -277,6 +295,7 @@ async def get_measures_sql_v3(
         lookback_window=lookback_window,
         query_parameters=json.loads(query_params) or None,
         matched_cube=cube_node.current if cube_node else None,
+        aggregate=aggregate,
     )
 
     response = _build_measures_response(result)

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

@@ -241,6 +241,7 @@ async def setup_build_context(
     include_temporal_filters: bool = False,
     lookback_window: str | None = None,
     matched_cube: Optional["NodeRevision"] = None,
+    aggregate: bool = True,
 ) -> BuildContext:
     """
     Create and initialize a BuildContext with all setup done.
@@ -261,6 +262,12 @@ async def setup_build_context(
         use_materialized: Whether to use materialized tables
         include_temporal_filters: Whether to include temporal partition filters from cube
         lookback_window: Lookback window for temporal filters
+        aggregate: Whether to apply server-side aggregation (GROUP BY). When False,
+            emits a flat SELECT projecting each component column under its
+            exact node column name (v3-compatible output format, not the v2
+            ``<node>_DOT_<column>`` convention) and bypasses pre-aggregation
+            table matching. Used by callers that apply their own aggregation
+            downstream (semantic parity with v2's ``preaggregate=False`` mode).
 
     Returns:
         Fully initialized BuildContext
@@ -284,6 +291,7 @@ async def setup_build_context(
         use_materialized=use_materialized,
         temporal_partition_columns=temporal_partition_columns or {},
         lookback_window=lookback_window,
+        aggregate=aggregate,
     )
 
     # Add filter-driven dimensions upfront — this only parses filter strings and
@@ -346,6 +354,7 @@ async def build_measures_sql(
     lookback_window: str | None = None,
     query_parameters: dict[str, Any] | None = None,
     matched_cube: Optional["NodeRevision"] = None,
+    aggregate: bool = True,
 ) -> GeneratedMeasuresSQL:
     """
     Build measures SQL for a set of metrics, dimensions, and filters.
@@ -370,6 +379,15 @@ async def build_measures_sql(
             incremental materialization and partition pruning.
         lookback_window: Lookback window for temporal filters (e.g., "3 DAY").
             If not provided, filters to exactly the logical timestamp partition.
+        aggregate: If True (default), apply server-side aggregation at the
+            requested grain. When False, emit a flat SELECT with no GROUP BY,
+            project each component column under its exact node column name
+            (v3-compatible output format, not the v2 ``<node>_DOT_<column>``
+            convention), and bypass pre-aggregation table matching. Intended
+            for callers that apply their own aggregation downstream (semantic
+            parity with v2's ``preaggregate=False`` mode). The
+            ``metric_formulas[].combiner`` field is still populated so callers
+            know how to aggregate the raw output.
 
     Returns:
         GeneratedMeasuresSQL with one GrainGroupSQL per aggregation level,
@@ -386,6 +404,7 @@ async def build_measures_sql(
         include_temporal_filters=include_temporal_filters,
         lookback_window=lookback_window,
         matched_cube=matched_cube,
+        aggregate=aggregate,
     )
 
     # Build grain groups from context

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

@@ -1887,8 +1887,14 @@ def build_grain_group_sql(
     """
     parent_node = grain_group.parent_node
 
-    # Check for matching pre-aggregation
-    if ctx.use_materialized and ctx.available_preaggs:
+    # Check for matching pre-aggregation.
+    # Skip matching when ``ctx.aggregate`` is False: the caller is asking
+    # for raw row-level output, which by definition cannot be served from a
+    # pre-aggregated table. We do NOT touch ``ctx.use_materialized`` here
+    # because that flag is dual-purpose: it also controls whether parent
+    # transforms are read from their materialized table vs recomputed via
+    # CTE (see ``should_use_materialized_table`` in ``materialization.py``).
+    if ctx.aggregate and ctx.use_materialized and ctx.available_preaggs:
         requested_grain = [dim.original_ref for dim in resolved_dimensions]
         matching_preagg = find_matching_preagg(
             ctx,
@@ -1933,6 +1939,25 @@ def build_grain_group_sql(
         # Collect unique components for API response
         unique_components.append(component)
 
+        # When the caller has opted out of server-side aggregation (the v3
+        # equivalent of v2's ``preaggregate=False``), project the raw inner
+        # expression under its exact node column name (e.g. ``is_product_view``)
+        # instead of wrapping it in the component's aggregation (``SUM(...)``).
+        # The downstream caller applies its own aggregation using the value of
+        # ``metric_formulas[].combiner`` from the response, so emitting the
+        # aggregation server-side would double-count. Mirrors the existing
+        # NONE-aggregability path below: the v3 output stays v3-shaped (bare
+        # column names), not the v2 ``<node>_DOT_<column>`` convention.
+        if not ctx.aggregate and component.expression:
+            col_ast = make_column_ref(component.expression)
+            component_alias = component.expression
+            component_expressions.append((component_alias, col_ast))
+            component_metadata.append(
+                (component_alias, component, metric_node),
+            )
+            component_aliases[component.name] = component_alias
+            continue
+
         # For NONE aggregability, output raw columns (no aggregation possible)
         # Note: This path is only hit for BASE metrics with NONE aggregability
         # (e.g., metrics with RANK() directly). Derived metrics with window functions
@@ -2067,9 +2092,15 @@ def build_grain_group_sql(
         # refs (not pre-aggregations).  A GROUP BY here would be redundant
         # — every projected column is part of the full grain — and the
         # real aggregation happens downstream after the join.
-        skip_agg = bool(
-            grain_group.non_decomposable_metrics
-            and grain_group.aggregability == Aggregability.NONE,
+        # When ``ctx.aggregate`` is False, every component above also emitted
+        # raw column refs, so a GROUP BY would aggregate raw rows the caller
+        # explicitly asked to receive ungrouped.
+        skip_agg = (
+            bool(
+                grain_group.non_decomposable_metrics
+                and grain_group.aggregability == Aggregability.NONE,
+            )
+            or not ctx.aggregate
         )
         query_ast, scanned_sources = build_select_ast(
             ctx,

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

@@ -53,6 +53,15 @@ class BuildContext:
     # Set to False when building SQL for materialization to avoid circular references
     use_materialized: bool = True
 
+    # Whether to apply server-side aggregation (default: True)
+    # When False, emits a flat SELECT with no GROUP BY and projects each
+    # component column under its exact node column name (the v3-compatible
+    # output format, not the v2 `<node>_DOT_<column>` convention).
+    # Pre-aggregation table matching is also bypassed in this mode, since the
+    # callers asking for raw rows are typically applying their own aggregation
+    # downstream. This is the v3 equivalent of v2's `preaggregate=False` mode.
+    aggregate: bool = True
+
     # Temporal filter settings for incremental materialization
     # Dict mapping column refs to partition objects from cube (e.g., {"v3.date.date_id": partition_obj})
     # When provided, temporal filters are pushed down to parent nodes via dimension links