feat: warn on fan-out risk in build_v3 SQL generation

datajunction-server/datajunction_server/api/cubes.py

@@ -786,6 +786,7 @@ async def materialize_cube(
         metric_combiners=combined_result.metric_combiners,
         workflow_urls=workflow_urls,
         message=message,
+        warnings=combined_result.warnings,
     )
 
 

datajunction-server/datajunction_server/api/graphql/schema.graphql

@@ -163,6 +163,7 @@ enum ErrorCode {
   INVALID_COLUMN
   QUERY_SERVICE_ERROR
   INVALID_ORDER_BY
+  FANOUT_RISK
   COMPOUND_BUILD_EXCEPTION
   MISSING_PARENT
   TYPE_INFERENCE

datajunction-server/datajunction_server/api/preaggregations.py

@@ -684,6 +684,7 @@ async def plan_preaggregations(
 
     return PlanPreAggregationsResponse(
         preaggs=[await _preagg_to_info(p, session) for p in loaded_preaggs],
+        warnings=measures_result.warnings,
     )
 
 

datajunction-server/datajunction_server/api/sql.py

@@ -288,6 +288,8 @@ async def get_measures_sql_v3(
         _tags,
     )
     get_metrics_provider().counter("dj.sql.requests", tags=_tags)
+    if result.warnings:
+        get_metrics_provider().counter("dj.sql.fanout_warnings", tags=_tags)
 
     _logger.info(
         "[SQL] endpoint=%s metrics=%s dimensions=%s filters=%s elapsed_ms=%.1f",
@@ -409,6 +411,7 @@ def _build_measures_response(result) -> MeasuresSQLResponse:
         metric_formulas=metric_formulas,
         dialect=str(result.dialect) if result.dialect else None,
         requested_dimensions=result.requested_dimensions,
+        warnings=result.warnings,
     )
 
 
@@ -476,14 +479,15 @@ async def get_combined_measures_sql_v3(
             combined_result,
             preagg_sources,
             _,
-        ) = await build_combiner_sql_from_preaggs(  # pragma: no cover
+        ) = await build_combiner_sql_from_preaggs(
             session=session,
             metrics=metrics,
             dimensions=dimensions,
             filters=filters,
             dialect=dialect,
         )
-        source_tables = [src.table_ref for src in preagg_sources]  # pragma: no cover
+        source_tables = [src.table_ref for src in preagg_sources]
+        combined_warnings = combined_result.warnings
     else:
         # Build the measures SQL to get grain groups (compute from scratch)
         result = await build_measures_sql(
@@ -511,6 +515,10 @@ async def get_combined_measures_sql_v3(
             # Extract table references from the query
             source_tables.append(gg.parent_name)
 
+        # Warnings live on the measures build's ctx sink (build_combiner_sql
+        # doesn't carry them); read them off the measures result.
+        combined_warnings = result.warnings
+
     elapsed_ms = (time.monotonic() - _t0) * 1000
     _tags = {"query_type": "measures_combined", "query_version": "v3"}
     get_metrics_provider().timer(
@@ -519,6 +527,8 @@ async def get_combined_measures_sql_v3(
         _tags,
     )
     get_metrics_provider().counter("dj.sql.requests", tags=_tags)
+    if combined_warnings:
+        get_metrics_provider().counter("dj.sql.fanout_warnings", tags=_tags)
 
     _logger.info(
         "[SQL] endpoint=%s metrics=%s dimensions=%s filters=%s elapsed_ms=%.1f",
@@ -553,6 +563,7 @@ async def get_combined_measures_sql_v3(
         dialect=str(dialect),
         use_preagg_tables=use_preagg_tables,
         source_tables=source_tables,
+        warnings=combined_warnings,
     )
 
 
@@ -664,6 +675,7 @@ async def get_metrics_sql_v3(
         dialect=result.dialect,
         cube_name=result.cube_name,
         scan_estimate=result.scan_estimate,
+        warnings=result.warnings,
     )
 
 

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

@@ -622,6 +622,10 @@ async def build_metrics_sql(
     # restriction in the outer query so seed rows don't leak into the result.
     apply_output_restriction(result.query, window_plan)
 
+    # Surface build warnings (e.g. fan-out risk) from the single ctx sink. Empty
+    # on the materialized-cube branch above (no source build ran).
+    result.warnings = ctx.warnings
+
     substitute_query_params(result.query, query_parameters or {})
 
     return result

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

@@ -36,6 +36,7 @@
 )
 from datajunction_server.construction.build_v3.types import GrainGroupSQL
 from datajunction_server.construction.build_v3.utils import build_join_from_clause
+from datajunction_server.errors import DJWarning
 from datajunction_server.models.column import SemanticType
 from datajunction_server.models.query import V3ColumnMetadata
 from datajunction_server.sql.parsing import ast
@@ -74,6 +75,9 @@ class CombinedGrainGroupResult:
     metric_combiners: dict[str, str] = field(default_factory=dict)
     # Dialect for rendering SQL (used for dialect-specific function names)
     dialect: Dialect = Dialect.SPARK
+    # Build warnings (e.g. fan-out risk), set from the ctx sink so the pre-agg/cube
+    # endpoints (which hold only this result, not the ctx) can surface them.
+    warnings: list[DJWarning] = field(default_factory=list)
 
     @property
     def sql(self) -> str:
@@ -706,6 +710,10 @@ async def build_combiner_sql_from_preaggs(
             temporal_partition_info.column_name,
         )
 
+    # Copy warnings from the ctx sink onto the combined result for the pre-agg/cube
+    # endpoints. Set last so they survive the reorder above.
+    combined_result.warnings = result.warnings
+
     return combined_result, preagg_sources, temporal_partition_info
 
 
@@ -805,6 +813,7 @@ def _get_projection_name(proj: ast.Node) -> str | None:
             component_aliases=result.component_aliases,
             metric_combiners=result.metric_combiners,
             dialect=result.dialect,
+            warnings=result.warnings,
         )
 
     return result

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

@@ -18,19 +18,24 @@
     DJError,
     DJException,
     DJInvalidInputException,
+    DJWarning,
     ErrorCode,
 )
 from datajunction_server.construction.build_v3.materialization import (
     get_table_reference_parts_with_materialization,
 )
 from datajunction_server.construction.build_v3.types import (
     BuildContext,
+    DecomposedMetricInfo,
     DimensionRef,
+    GrainGroup,
     JoinPath,
     ResolvedDimension,
 )
 from datajunction_server.database.dimensionlink import DimensionLink
 from datajunction_server.database.node import Node, NodeType
+from datajunction_server.models.dimensionlink import JoinCardinality
+from datajunction_server.sql.decompose import is_duplication_invariant
 from datajunction_server.sql.parsing import ast
 from datajunction_server.sql.parsing.backends.antlr4 import parse
 from datajunction_server.utils import SEPARATOR
@@ -1023,3 +1028,116 @@ def build_join_clause(
     )
 
     return join
+
+
+# Fan-out cardinalities (default MANY_TO_ONE is safe). See check_fanout_safety.
+UNSAFE_JOIN_CARDINALITIES = frozenset(
+    {JoinCardinality.ONE_TO_MANY, JoinCardinality.MANY_TO_MANY},
+)
+
+
+def find_unsafe_cardinality_links(
+    resolved_dimensions: list[ResolvedDimension],
+) -> list[DimensionLink]:
+    """Return the emitted join-path links whose cardinality can fan out."""
+    unsafe: list[DimensionLink] = []
+    for resolved_dim in resolved_dimensions:
+        if resolved_dim.is_local or not resolved_dim.join_path:
+            continue
+        for link in resolved_dim.join_path.links:
+            if link.join_cardinality in UNSAFE_JOIN_CARDINALITIES:
+                unsafe.append(link)
+    return unsafe
+
+
+def _nondecomposable_inflates(info: DecomposedMetricInfo) -> bool:
+    """
+    Whether a non-decomposable metric is inflated by row duplication. Holistic
+    aggregations (MEDIAN, PERCENTILE, ARRAY_AGG) depend on the full multiset of
+    rows, so duplication distorts them; argmax-style MAX_BY/MIN_BY are immune.
+    """
+    for func in info.derived_ast.find_all(ast.Function):
+        func_class = func.function()
+        if func_class is None or not func_class.is_aggregation:
+            continue
+        if not is_duplication_invariant(func_class):
+            return True
+    return False
+
+
+def check_fanout_safety(
+    grain_group: GrainGroup,
+    unsafe_links: list[DimensionLink],
+) -> DJWarning | None:
+    """
+    Return a fan-out warning if this grain group aggregates a duplication-sensitive
+    metric across a fanning join.
+
+    A ONE_TO_MANY/MANY_TO_MANY link duplicates fact rows. A metric is inflated when
+    its result depends on how many times a row appears: additive re-aggregation (a
+    SUM merge, used by SUM/COUNT/AVG/...) and holistic aggregations applied over the
+    raw rows (MEDIAN, ARRAY_AGG, ...). MIN/MAX, COUNT(DISTINCT), and argmax-style
+    MAX_BY/MIN_BY are immune. We warn rather than error: the user may know the data
+    does not fan out.
+    """
+    if not unsafe_links:
+        return None
+
+    # Decomposable metrics with a SUM merge.
+    inflated_metrics: set[str] = {
+        metric_node.name
+        for metric_node, component in grain_group.components
+        if component.merge == "SUM"
+    }
+
+    # Non-decomposable holistic metrics (MEDIAN, ARRAY_AGG, ...).
+    inflated_metrics.update(
+        info.metric_node.name
+        for info in grain_group.non_decomposable_metrics
+        if _nondecomposable_inflates(info)
+    )
+
+    if not inflated_metrics:
+        return None
+
+    metric_names = sorted(inflated_metrics)
+    link_descriptions = sorted(
+        {
+            f"{link.node_revision.name} -> {link.dimension.name} "
+            f"({JoinCardinality(link.join_cardinality).value})"
+            for link in unsafe_links
+        },
+    )
+    metrics_str = ", ".join(metric_names)
+    links_str = "; ".join(link_descriptions)
+    message = (
+        f"Possible fan-out: metric(s) {metrics_str} aggregate across a fan-out "
+        f"link [{links_str}] when joining {grain_group.parent_node.name} to the "
+        f"requested dimensions. One fact row matches many dimension rows, so the "
+        f"result may be inflated by duplicated fact rows. If a filter constrains "
+        f"the relationship to be effectively one-to-one this result is correct; "
+        f"otherwise remove the offending dimension(s), or correct the link's "
+        f"cardinality if the relationship is not actually one-to-many or "
+        f"many-to-many."
+    )
+    logger.warning(
+        "[BuildV3] fan-out risk: metrics=%s links=%s parent=%s",
+        metrics_str,
+        links_str,
+        grain_group.parent_node.name,
+        extra={
+            "fanout_metrics": metric_names,
+            "fanout_links": link_descriptions,
+            "fanout_parent": grain_group.parent_node.name,
+            "error_code": ErrorCode.FANOUT_RISK.name,
+        },
+    )
+    return DJWarning(
+        code=ErrorCode.FANOUT_RISK,
+        message=message,
+        debug={
+            "metrics": metric_names,
+            "links": link_descriptions,
+            "parent": grain_group.parent_node.name,
+        },
+    )

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

@@ -66,6 +66,8 @@
     merge_grain_groups,
 )
 from datajunction_server.construction.build_v3.dimensions import (
+    check_fanout_safety,
+    find_unsafe_cardinality_links,
     parse_dimension_ref,
     resolve_dimensions,
     resolve_metric_expression_dimensions,
@@ -2315,9 +2317,15 @@ def process_metric_group(
     # Resolve dimensions (find join paths) - shared across grain groups
     resolved_dimensions = resolve_dimensions(ctx, parent_node)
 
+    # Shared across all grain groups here, so scan once.
+    unsafe_links = find_unsafe_cardinality_links(resolved_dimensions)
+
     # Build SQL for each grain group
     grain_group_sqls: list[GrainGroupSQL] = []
     for grain_group in grain_groups:
+        # Flag (don't block) fan-out risk for this grain group; see check_fanout_safety.
+        fanout_warning = check_fanout_safety(grain_group, unsafe_links)
+
         # Reset alias registry for each grain group to avoid conflicts
         ctx.alias_registry = AliasRegistry()
         ctx._table_alias_counter = 0
@@ -2328,6 +2336,7 @@ def process_metric_group(
             resolved_dimensions,
             components_per_metric,
         )
+        ctx.add_warning(fanout_warning)
         grain_group_sqls.append(grain_group_sql)
     return grain_group_sqls
 
@@ -2553,6 +2562,12 @@ def find_parent_for_window_metric(
             component_aggregabilities=component_aggregabilities,
         )
 
+        # Per-iteration resolved_dimensions here, so the scan can't be hoisted.
+        fanout_warning = check_fanout_safety(
+            grain_group,
+            find_unsafe_cardinality_links(resolved_dimensions),
+        )
+
         # Build GrainGroupSQL
         components_per_metric: dict[str, int] = {}
         for metric_name in base_metrics_needed:
@@ -2569,6 +2584,7 @@ def find_parent_for_window_metric(
             resolved_dimensions,
             components_per_metric,
         )
+        ctx.add_warning(fanout_warning)
 
         # Restore original dimensions
         ctx.dimensions = original_dimensions