From 52d9b50f4b38dd5b60aa720e999adcfed3fb96cc Mon Sep 17 00:00:00 2001 From: Robin Davis Date: Mon, 20 Jul 2026 13:27:46 -0700 Subject: [PATCH 1/2] feat: warn on fan-out risk in build_v3 SQL generation Emit a non-blocking FANOUT_RISK warning when a metric query joins across ONE_TO_MANY/MANY_TO_MANY dimension links that could inflate an additive (SUM) merge. A dj.sql.fanout_warnings counter is emitted alongside the existing build telemetry. --- .../datajunction_server/api/cubes.py | 1 + .../api/graphql/schema.graphql | 1 + .../api/preaggregations.py | 1 + .../datajunction_server/api/sql.py | 16 +- .../construction/build_v3/builder.py | 4 + .../construction/build_v3/combiners.py | 9 + .../construction/build_v3/dimensions.py | 90 ++++ .../construction/build_v3/measures.py | 16 + .../construction/build_v3/types.py | 43 +- .../database/dimensionlink.py | 1 + .../datajunction_server/errors.py | 12 +- .../internal/deployment/orchestrator.py | 1 + .../datajunction_server/internal/sql.py | 2 + .../datajunction_server/mcp/tools.py | 7 + .../models/cube_materialization.py | 4 +- .../datajunction_server/models/deployment.py | 4 + .../datajunction_server/models/metric.py | 3 + .../models/preaggregation.py | 2 + .../datajunction_server/models/sql.py | 4 +- datajunction-server/tests/api/cubes_test.py | 102 +++- .../tests/api/deployments_test.py | 53 +- .../tests/api/namespaces_test.py | 4 + .../build_v3/fanout_guard_test.py | 474 ++++++++++++++++++ .../tests/dj_mcp/test_tools.py | 51 ++ .../tests/models/deployment_test.py | 32 ++ 25 files changed, 926 insertions(+), 11 deletions(-) create mode 100644 datajunction-server/tests/construction/build_v3/fanout_guard_test.py diff --git a/datajunction-server/datajunction_server/api/cubes.py b/datajunction-server/datajunction_server/api/cubes.py index ed2b07ef95..682e72d1af 100644 --- a/datajunction-server/datajunction_server/api/cubes.py +++ b/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, ) diff --git a/datajunction-server/datajunction_server/api/graphql/schema.graphql b/datajunction-server/datajunction_server/api/graphql/schema.graphql index 28ccab6913..d2e070d5f3 100644 --- a/datajunction-server/datajunction_server/api/graphql/schema.graphql +++ b/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 diff --git a/datajunction-server/datajunction_server/api/preaggregations.py b/datajunction-server/datajunction_server/api/preaggregations.py index 8d3717f0cb..14fae20a6f 100644 --- a/datajunction-server/datajunction_server/api/preaggregations.py +++ b/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, ) diff --git a/datajunction-server/datajunction_server/api/sql.py b/datajunction-server/datajunction_server/api/sql.py index 00d02e4de2..7aacdfd850 100644 --- a/datajunction-server/datajunction_server/api/sql.py +++ b/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, ) diff --git a/datajunction-server/datajunction_server/construction/build_v3/builder.py b/datajunction-server/datajunction_server/construction/build_v3/builder.py index 9562ac03b0..6397203c7e 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/builder.py +++ b/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 diff --git a/datajunction-server/datajunction_server/construction/build_v3/combiners.py b/datajunction-server/datajunction_server/construction/build_v3/combiners.py index 8aa092ea9d..d04794db5e 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/combiners.py +++ b/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 diff --git a/datajunction-server/datajunction_server/construction/build_v3/dimensions.py b/datajunction-server/datajunction_server/construction/build_v3/dimensions.py index 4e009aa4c4..6b1d5934cf 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/dimensions.py +++ b/datajunction-server/datajunction_server/construction/build_v3/dimensions.py @@ -18,6 +18,7 @@ DJError, DJException, DJInvalidInputException, + DJWarning, ErrorCode, ) from datajunction_server.construction.build_v3.materialization import ( @@ -26,11 +27,14 @@ from datajunction_server.construction.build_v3.types import ( BuildContext, DimensionRef, + GrainGroup, JoinPath, ResolvedDimension, ) from datajunction_server.database.dimensionlink import DimensionLink from datajunction_server.database.node import Node, NodeType +from datajunction_server.models.decompose import MetricComponent +from datajunction_server.models.dimensionlink import JoinCardinality from datajunction_server.sql.parsing import ast from datajunction_server.sql.parsing.backends.antlr4 import parse from datajunction_server.utils import SEPARATOR @@ -1023,3 +1027,89 @@ 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 check_fanout_safety( + grain_group: GrainGroup, + unsafe_links: list[DimensionLink], +) -> DJWarning | None: + """ + Return a fan-out warning if this grain group re-aggregates a component with an + additive merge function across a fanning join. + + A ONE_TO_MANY/MANY_TO_MANY link duplicates fact rows, inflating additive (SUM) + merges. We warn rather than error: the user may know the data does not fan out. + """ + if not unsafe_links: + return None + + # Components with fan-out risk: those with a SUM merge + inflated: list[tuple[Node, MetricComponent]] = [ + (metric_node, component) + for metric_node, component in grain_group.components + if component.merge == "SUM" + ] + if not inflated: + return None + + metric_names = sorted({metric_node.name for metric_node, _ in inflated}) + 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, + }, + ) diff --git a/datajunction-server/datajunction_server/construction/build_v3/measures.py b/datajunction-server/datajunction_server/construction/build_v3/measures.py index 3b7ff01a96..6b8e12b4eb 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/measures.py +++ b/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 diff --git a/datajunction-server/datajunction_server/construction/build_v3/types.py b/datajunction-server/datajunction_server/construction/build_v3/types.py index b24964fefa..2025b0dce9 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/types.py +++ b/datajunction-server/datajunction_server/construction/build_v3/types.py @@ -9,7 +9,7 @@ from datajunction_server.construction.build_v3.alias_registry import AliasRegistry from datajunction_server.database.dimensionlink import DimensionLink from datajunction_server.database.node import Node -from datajunction_server.errors import DJInvalidInputException +from datajunction_server.errors import DJInvalidInputException, DJWarning from datajunction_server.models.decompose import MetricComponent, Aggregability from datajunction_server.models.dialect import Dialect from datajunction_server.models.node_type import NodeType @@ -30,10 +30,11 @@ @dataclass class BuildContext: """ - Immutable context passed through the SQL generation pipeline. + Context threaded through the SQL generation pipeline. - Contains all the information needed to build SQL for a set of metrics - and dimensions. + Holds the build inputs (metrics, dimensions, filters, dialect) plus the + state accumulated during a build: loaded nodes, join paths, caches, and + warnings. One instance per top-level build. """ session: AsyncSession @@ -128,6 +129,25 @@ class BuildContext: # ResolvedDimension for these and skips the unreachable-dim error. pushdown_resolved_dims: set[str] = field(default_factory=set) + # Build warnings (e.g. fan-out risk): the single sink. Producers append via + # add_warning(); each endpoint reads ctx.warnings once at its boundary. + warnings: list[DJWarning] = field(default_factory=list) + + def add_warning(self, warning: DJWarning | None) -> None: + """ + Record a build warning, de-duplicated by (code, message). No-ops on None. + + The same risk (e.g. fan-out) can be detected once per grain group, so + callers append unconditionally and rely on this to keep the surfaced + list clean. + """ + if warning is None: + return + key = (warning.code, warning.message) + if any((w.code, w.message) == key for w in self.warnings): + return + self.warnings.append(warning) + def next_table_alias(self, base_name: str) -> str: """Generate a unique table alias.""" self._table_alias_counter += 1 @@ -351,6 +371,17 @@ class GeneratedMeasuresSQL: # These are LAG/LEAD window function metrics that need aggregation at a different grain window_metric_grains: dict[str, set[str]] = field(default_factory=dict) + @property + def warnings(self) -> list[DJWarning]: + """ + Warnings accumulated during this build. + + Derived from the build context (the single source of truth) rather than + stored, so every consumer of a measures build reads ``result.warnings`` + and no intermediate shape has to copy them around. + """ + return self.ctx.warnings + @dataclass class GeneratedSQL: @@ -374,6 +405,10 @@ class GeneratedSQL: # Scan estimate aggregated from all grain groups scan_estimate: Optional["ScanEstimate"] = None + # Build warnings (e.g. fan-out risk), populated from the build context's + # warnings sink (see builder.build_metrics_sql). + warnings: list[DJWarning] = field(default_factory=list) + @property def sql(self) -> str: """Render the query AST to SQL string for the target dialect.""" diff --git a/datajunction-server/datajunction_server/database/dimensionlink.py b/datajunction-server/datajunction_server/database/dimensionlink.py index 935eee6f19..64e7bf375b 100644 --- a/datajunction-server/datajunction_server/database/dimensionlink.py +++ b/datajunction-server/datajunction_server/database/dimensionlink.py @@ -100,6 +100,7 @@ def to_spec(self): dimension_node=self.dimension.name, join_on=self.join_sql, join_type=self.join_type if self.join_type else JoinType.LEFT, + join_cardinality=self.join_cardinality, spark_hints=self.spark_hints, ) diff --git a/datajunction-server/datajunction_server/errors.py b/datajunction-server/datajunction_server/errors.py index 0e3efa4b9d..4eece60a9d 100644 --- a/datajunction-server/datajunction_server/errors.py +++ b/datajunction-server/datajunction_server/errors.py @@ -36,6 +36,7 @@ class ErrorCode(IntEnum): INVALID_COLUMN = 206 QUERY_SERVICE_ERROR = 207 INVALID_ORDER_BY = 208 + FANOUT_RISK = 209 # SQL Build Error COMPOUND_BUILD_EXCEPTION = 300 @@ -130,7 +131,10 @@ class DJWarningType(TypedDict): Type for serialized warnings. """ - code: Optional[int] + # Serialized as the symbolic ErrorCode name (e.g. "FANOUT_RISK"), matching + # DJWarning.serialize_code and DJErrorType.code. Optional because a warning's + # code may be unset. + code: Optional[str] message: str debug: Optional[DebugType] @@ -144,6 +148,12 @@ class DJWarning(BaseModel): message: str debug: Optional[Dict[str, Any]] = None + @field_serializer("code") + def serialize_code(self, code: Optional[ErrorCode]) -> Optional[str]: + # Mirror DJError: serialize the symbolic name (e.g. "FANOUT_RISK") rather + # than the integer value, so UI/CLI consumers get a stable string. + return code.name if code is not None else None + DBAPIExceptions = Literal[ "Warning", diff --git a/datajunction-server/datajunction_server/internal/deployment/orchestrator.py b/datajunction-server/datajunction_server/internal/deployment/orchestrator.py index f0ac454e7c..cb67e2c6be 100644 --- a/datajunction-server/datajunction_server/internal/deployment/orchestrator.py +++ b/datajunction-server/datajunction_server/internal/deployment/orchestrator.py @@ -2044,6 +2044,7 @@ async def _create_or_update_dimension_link( link_input = JoinLinkInput( dimension_node=join_link.rendered_dimension_node, join_type=join_link.join_type, + join_cardinality=join_link.join_cardinality, join_on=join_link.rendered_join_on, role=join_link.role, default_value=join_link.default_value, diff --git a/datajunction-server/datajunction_server/internal/sql.py b/datajunction-server/datajunction_server/internal/sql.py index 896575c806..3c47653599 100644 --- a/datajunction-server/datajunction_server/internal/sql.py +++ b/datajunction-server/datajunction_server/internal/sql.py @@ -277,6 +277,8 @@ async def generate_metrics_sql( provider = get_metrics_provider() provider.timer("dj.sql.build_latency_ms", elapsed_ms, _tags) provider.counter("dj.sql.requests", tags=_tags) + if result.warnings: + provider.counter("dj.sql.fanout_warnings", tags=_tags) logger.info( "[SQL] endpoint=%s metrics=%s dimensions=%s filters=%s elapsed_ms=%.1f", endpoint, diff --git a/datajunction-server/datajunction_server/mcp/tools.py b/datajunction-server/datajunction_server/mcp/tools.py index 651fc7b8c5..2ff38e41c5 100644 --- a/datajunction-server/datajunction_server/mcp/tools.py +++ b/datajunction-server/datajunction_server/mcp/tools.py @@ -577,6 +577,13 @@ async def get_query_plan( "", ] + if result.warnings: + lines += ["⚠ Warnings", "-" * 60] + for warning in result.warnings: + code = warning.code.name if warning.code else "WARNING" + lines.append(f" [{code}] {warning.message}") + lines.append("") + lines += ["Metric Formulas", "-" * 60] for metric_name, decomposed in result.decomposed_metrics.items(): derived_tag = ( diff --git a/datajunction-server/datajunction_server/models/cube_materialization.py b/datajunction-server/datajunction_server/models/cube_materialization.py index 1db3f5c2e0..2bb88e632d 100644 --- a/datajunction-server/datajunction_server/models/cube_materialization.py +++ b/datajunction-server/datajunction_server/models/cube_materialization.py @@ -5,7 +5,7 @@ from pydantic import BaseModel, Field, field_validator, computed_field -from datajunction_server.errors import DJInvalidInputException +from datajunction_server.errors import DJInvalidInputException, DJWarning from datajunction_server.models.column import SemanticType from datajunction_server.models.decompose import ( Aggregability, @@ -548,6 +548,8 @@ class CubeMaterializeResponse(BaseModel): # Status message: str + warnings: List[DJWarning] = [] + class CubeMaterializationV2Input(BaseModel): """ diff --git a/datajunction-server/datajunction_server/models/deployment.py b/datajunction-server/datajunction_server/models/deployment.py index 8b3fed93f5..2064234317 100644 --- a/datajunction-server/datajunction_server/models/deployment.py +++ b/datajunction-server/datajunction_server/models/deployment.py @@ -16,6 +16,7 @@ ) from datajunction_server.models.base import labelize from datajunction_server.models.dimensionlink import ( + JoinCardinality, JoinType, LinkType, SparkJoinStrategy, @@ -222,6 +223,7 @@ class DimensionJoinLinkSpec(DimensionLinkSpec): node_column: str | None = None join_type: JoinType = JoinType.LEFT + join_cardinality: JoinCardinality = JoinCardinality.MANY_TO_ONE join_on: str | None = None default_value: str | None = None spark_hints: Optional[SparkJoinStrategy] = None @@ -249,6 +251,7 @@ def __hash__(self) -> int: self.role, self.rendered_dimension_node, self.join_type, + self.join_cardinality, self.rendered_join_on, self.node_column, self.default_value, @@ -262,6 +265,7 @@ def __eq__(self, other: Any) -> bool: super().__eq__(other) and self.rendered_dimension_node == other.rendered_dimension_node and self.join_type == other.join_type + and self.join_cardinality == other.join_cardinality and self.rendered_join_on == other.rendered_join_on and self.node_column == other.node_column and self.default_value == other.default_value diff --git a/datajunction-server/datajunction_server/models/metric.py b/datajunction-server/datajunction_server/models/metric.py index fdda373f72..1e12e43af6 100644 --- a/datajunction-server/datajunction_server/models/metric.py +++ b/datajunction-server/datajunction_server/models/metric.py @@ -8,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from datajunction_server.database.node import Node +from datajunction_server.errors import DJWarning from datajunction_server.models.cube_materialization import MetricComponent from datajunction_server.models.engine import Dialect from datajunction_server.models.node import ( @@ -153,3 +154,5 @@ class V3TranslatedSQL(BaseModel): # Scan estimate (aggregated from all grain groups) scan_estimate: Optional["ScanEstimate"] = None + + warnings: List[DJWarning] = [] diff --git a/datajunction-server/datajunction_server/models/preaggregation.py b/datajunction-server/datajunction_server/models/preaggregation.py index bc943a3e8b..86c439c9bb 100644 --- a/datajunction-server/datajunction_server/models/preaggregation.py +++ b/datajunction-server/datajunction_server/models/preaggregation.py @@ -8,6 +8,7 @@ from pydantic import BaseModel, Field, field_validator from datajunction_server.enum import StrEnum +from datajunction_server.errors import DJWarning from datajunction_server.models.materialization import MaterializationStrategy from datajunction_server.models.node import PartitionAvailability from datajunction_server.models.node_type import NodeNameVersion @@ -83,6 +84,7 @@ class PlanPreAggregationsResponse(BaseModel): """Response model for /preaggs/plan endpoint.""" preaggs: List["PreAggregationInfo"] + warnings: List[DJWarning] = [] class ExternalPreAggTable(BaseModel): diff --git a/datajunction-server/datajunction_server/models/sql.py b/datajunction-server/datajunction_server/models/sql.py index 90d7e60a2e..2fe7897a23 100644 --- a/datajunction-server/datajunction_server/models/sql.py +++ b/datajunction-server/datajunction_server/models/sql.py @@ -8,7 +8,7 @@ from pydantic import field_validator from pydantic.main import BaseModel -from datajunction_server.errors import DJQueryBuildError +from datajunction_server.errors import DJQueryBuildError, DJWarning from datajunction_server.models.cube_materialization import MetricComponent from datajunction_server.models.engine import Dialect from datajunction_server.models.node_type import NodeNameVersion @@ -144,6 +144,7 @@ class MeasuresSQLResponse(BaseModel): metric_formulas: List[MetricFormulaResponse] # How metrics combine components dialect: Optional[str] = None requested_dimensions: List[str] + warnings: List[DJWarning] = [] class CombinedMeasuresSQLResponse(BaseModel): @@ -163,3 +164,4 @@ class CombinedMeasuresSQLResponse(BaseModel): bool # If True, data is read from pre-agg tables; if False, from source tables ) source_tables: List[str] # Tables being read (pre-agg tables or source tables) + warnings: List[DJWarning] = [] diff --git a/datajunction-server/tests/api/cubes_test.py b/datajunction-server/tests/api/cubes_test.py index 4ce626cd69..c92fcb4e5b 100644 --- a/datajunction-server/tests/api/cubes_test.py +++ b/datajunction-server/tests/api/cubes_test.py @@ -9,7 +9,11 @@ import pytest_asyncio from httpx import AsyncClient -from datajunction_server.errors import DJQueryServiceClientException +from datajunction_server.errors import ( + DJQueryServiceClientException, + DJWarning, + ErrorCode, +) from datajunction_server.construction.build_v3.combiners import ( PreAggSourceInfo, TemporalPartitionInfo, @@ -4298,6 +4302,7 @@ def _create_mock_combined_result( mock_result.measure_components = measure_components or [] mock_result.component_aliases = component_aliases or {} mock_result.metric_combiners = metric_combiners or {} + mock_result.warnings = [] return mock_result @@ -4398,6 +4403,101 @@ async def test_materialize_cube_full_strategy_success( assert "workflow_urls" in data assert len(data["workflow_urls"]) == 1 + @pytest.mark.asyncio + async def test_materialize_cube_surfaces_fanout_warning( + self, + client_with_repairs_cube: AsyncClient, + mocker, + ): + """ + A fan-out risk detected while building the combined SQL from pre-agg + tables must be surfaced on the materialize response — this is when the + inflation gets baked into the cube, so it is the right moment to flag it. + """ + cube_name = "default.test_materialize_fanout_cube" + await make_a_test_cube( + client_with_repairs_cube, + cube_name, + with_materialization=False, + ) + + mock_columns = [ + V3ColumnMetadata( + name="state", + type="string", + semantic_name="default.hard_hat.state", + semantic_type="dimension", + ), + V3ColumnMetadata( + name="date_id", + type="int", + semantic_name="default.hard_hat.hire_date", + semantic_type="dimension", + ), + V3ColumnMetadata( + name="total_repair_cost", + type="double", + semantic_name="default.total_repair_cost", + semantic_type="measure", + ), + ] + mock_combined_result = _create_mock_combined_result( + mocker, + columns=mock_columns, + shared_dimensions=["default.hard_hat.state", "default.hard_hat.hire_date"], + sql_string="SELECT state, date_id, SUM(cost) AS total_repair_cost FROM preagg GROUP BY state, date_id", + ) + # The combined build detected a fan-out risk. + mock_combined_result.warnings = [ + DJWarning( + code=ErrorCode.FANOUT_RISK, + message="Possible fan-out: metric(s) default.total_repair_cost ...", + debug={"metrics": ["default.total_repair_cost"]}, + ), + ] + + mock_temporal_info = TemporalPartitionInfo( + column_name="date_id", + format="yyyyMMdd", + granularity="day", + ) + mocker.patch( + "datajunction_server.api.cubes.build_combiner_sql_from_preaggs", + return_value=( + mock_combined_result, + [ + PreAggSourceInfo( + table_ref="catalog.schema.preagg_table1", + parent_name="default.repair_orders", + strategy=None, + ), + ], + mock_temporal_info, + ), + ) + mocker.patch( + "datajunction_server.api.cubes._reorder_partition_column_last", + side_effect=lambda result, _col: result, + ) + qs_client = client_with_repairs_cube.app.dependency_overrides[ + get_query_service_client + ]() + mocker.patch.object( + qs_client, + "materialize_cube_v2", + return_value=mocker.MagicMock(urls=["http://workflow/cube-workflow"]), + ) + + response = await client_with_repairs_cube.post( + f"/cubes/{cube_name}/materialize", + json={"strategy": "full", "schedule": "0 0 * * *"}, + ) + + assert response.status_code == 200, response.json() + warnings = response.json()["warnings"] + assert len(warnings) == 1 + assert warnings[0]["code"] == "FANOUT_RISK" + @pytest.mark.asyncio async def test_materialize_cube_full_uses_cube_partition_with_role( self, diff --git a/datajunction-server/tests/api/deployments_test.py b/datajunction-server/tests/api/deployments_test.py index 34fb45e8c8..33945f6151 100644 --- a/datajunction-server/tests/api/deployments_test.py +++ b/datajunction-server/tests/api/deployments_test.py @@ -27,7 +27,7 @@ InProcessExecutor, _normalize_repo_path, ) -from datajunction_server.models.dimensionlink import JoinType +from datajunction_server.models.dimensionlink import JoinCardinality, JoinType from datajunction_server.database.node import Node, NodeRelationship from datajunction_server.database.tag import Tag from datajunction_server.models.node import ( @@ -1543,6 +1543,57 @@ async def test_deploy_with_dimension_link_update( "status": "success", } + @pytest.mark.asyncio + async def test_deploy_dimension_link_join_cardinality_round_trip( + self, + session, + client, + default_hard_hats, + default_us_states, + default_us_state, + ): + """ + A non-default join_cardinality set in a deployment spec must persist onto + the DimensionLink (orchestrator threads it into the JoinLinkInput) and + survive DimensionLink.to_spec(). + """ + namespace = "link_cardinality" + dim_spec = DimensionSpec( + name="default.hard_hat", + query="SELECT hard_hat_id, state FROM ${prefix}default.hard_hats", + primary_key=["hard_hat_id"], + owners=["dj"], + dimension_links=[ + # one_to_many: a worker can be assigned across several states, so + # one hard_hat row matches many us_state rows (a fan-out link). + DimensionJoinLinkSpec( + dimension_node="${prefix}default.us_state", + join_type="inner", + join_on="${prefix}default.hard_hat.state = ${prefix}default.us_state.state_short", + join_cardinality=JoinCardinality.ONE_TO_MANY, + ), + ], + ) + data = await deploy_and_wait( + client, + DeploymentSpec( + namespace=namespace, + nodes=[ + dim_spec, + default_hard_hats, + default_us_states, + default_us_state, + ], + ), + ) + assert data["status"] == "success" + + hard_hat = await Node.get_by_name(session, "link_cardinality.default.hard_hat") + (link,) = hard_hat.current.dimension_links + # Persisted onto the link, and preserved by to_spec(). + assert link.join_cardinality == JoinCardinality.ONE_TO_MANY + assert link.to_spec().join_cardinality == JoinCardinality.ONE_TO_MANY + @pytest.mark.asyncio async def test_deploy_with_reference_dimension_link( self, diff --git a/datajunction-server/tests/api/namespaces_test.py b/datajunction-server/tests/api/namespaces_test.py index cba5b6e16b..0027cfc694 100644 --- a/datajunction-server/tests/api/namespaces_test.py +++ b/datajunction-server/tests/api/namespaces_test.py @@ -1054,12 +1054,14 @@ async def test_export_namespaces_deployment(client_with_roads: AsyncClient): "join_on": "${prefix}repair_orders_fact.municipality_id = " "${prefix}municipality_dim.municipality_id", "join_type": "inner", + "join_cardinality": "many_to_one", "type": "join", }, { "dimension_node": "${prefix}hard_hat", "join_on": "${prefix}repair_orders_fact.hard_hat_id = ${prefix}hard_hat.hard_hat_id", "join_type": "inner", + "join_cardinality": "many_to_one", "type": "join", }, { @@ -1067,12 +1069,14 @@ async def test_export_namespaces_deployment(client_with_roads: AsyncClient): "join_on": "${prefix}repair_orders_fact.hard_hat_id = " "${prefix}hard_hat_to_delete.hard_hat_id", "join_type": "left", + "join_cardinality": "many_to_one", "type": "join", }, { "dimension_node": "${prefix}dispatcher", "join_on": "${prefix}repair_orders_fact.dispatcher_id = ${prefix}dispatcher.dispatcher_id", "join_type": "inner", + "join_cardinality": "many_to_one", "type": "join", }, ] diff --git a/datajunction-server/tests/construction/build_v3/fanout_guard_test.py b/datajunction-server/tests/construction/build_v3/fanout_guard_test.py new file mode 100644 index 0000000000..213e78353c --- /dev/null +++ b/datajunction-server/tests/construction/build_v3/fanout_guard_test.py @@ -0,0 +1,474 @@ +""" +Tests for the cardinality-aware fan-out guard (DJ issue #1531). + +When a metric's dimension join path crosses a link declared ONE_TO_MANY or +MANY_TO_MANY, one fact row matches many dimension rows, duplicating the fact's +measures. Additive aggregations (SUM/COUNT/AVG) over the duplicated rows +over-count. The V3 builder still generates SQL (DJ cannot statically prove a +filter makes the join safe, so a hard error would be a false positive) but +attaches a ``FANOUT_RISK`` warning to the response. + +These tests exercise that guard end-to-end through the /sql/measures/v3/ and +/sql/metrics/v3/ endpoints, plus the negative cases that must NOT trip it +(safe cardinalities and fan-out-immune aggregations). +""" + +import pytest + + +def fanout_warnings(payload: dict) -> list[dict]: + """Return the FANOUT_RISK warnings on a measures/metrics SQL response.""" + return [ + warning + for warning in payload.get("warnings") or [] + if warning.get("code") == "FANOUT_RISK" + ] + + +@pytest.fixture +async def setup_fanout_links(client_with_build_v3): + """ + Add a dimension that fans out from ``v3.order_details``. + + A single order line item maps to many promotions (one order has several + promo codes applied), so the link is declared ``one_to_many``. A second + ``many_to_many`` link is added to exercise that cardinality too. A ``MIN`` + metric is added to cover the duplication-immune aggregation case. + + Defined locally so only the fan-out tests pay the setup cost; the global + BUILD_V3 fixture is unaffected. + """ + response = await client_with_build_v3.post( + "/nodes/source/", + json={ + "name": "v3.src_order_promotions", + "description": "Promotions applied to orders (many per order)", + "columns": [ + {"name": "promo_code", "type": "string"}, + {"name": "order_id", "type": "int"}, + {"name": "discount", "type": "float"}, + {"name": "campaign", "type": "string"}, + ], + "mode": "published", + "catalog": "default", + "schema_": "v3", + "table": "order_promotions", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + + response = await client_with_build_v3.post( + "/nodes/dimension/", + json={ + "name": "v3.order_promotion", + "description": "Promotion dimension keyed on promo_code", + "query": ( + "SELECT promo_code, order_id, discount, campaign " + "FROM v3.src_order_promotions" + ), + "primary_key": ["promo_code"], + "mode": "published", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + + # order_details -> order_promotion: one order line maps to many promotions. + response = await client_with_build_v3.post( + "/nodes/v3.order_details/link", + json={ + "dimension_node": "v3.order_promotion", + "join_type": "left", + "join_on": "v3.order_details.order_id = v3.order_promotion.order_id", + "join_cardinality": "one_to_many", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + + # A second dimension linked many_to_many, to exercise that cardinality. + response = await client_with_build_v3.post( + "/nodes/source/", + json={ + "name": "v3.src_order_channels", + "description": "Sales channels associated with orders (many-to-many)", + "columns": [ + {"name": "channel_id", "type": "int"}, + {"name": "order_id", "type": "int"}, + {"name": "channel_name", "type": "string"}, + ], + "mode": "published", + "catalog": "default", + "schema_": "v3", + "table": "order_channels", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + + response = await client_with_build_v3.post( + "/nodes/dimension/", + json={ + "name": "v3.order_channel", + "description": "Channel dimension keyed on channel_id", + "query": ( + "SELECT channel_id, order_id, channel_name FROM v3.src_order_channels" + ), + "primary_key": ["channel_id"], + "mode": "published", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + + response = await client_with_build_v3.post( + "/nodes/v3.order_details/link", + json={ + "dimension_node": "v3.order_channel", + "join_type": "left", + "join_on": "v3.order_details.order_id = v3.order_channel.order_id", + "join_cardinality": "many_to_many", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + + # A MIN metric: its merge function is MIN (non-additive), so it's + # duplication-immune and must NOT trip the guard even across a fan-out link. + response = await client_with_build_v3.post( + "/nodes/metric/", + json={ + "name": "v3.min_unit_price", + "description": "Minimum unit price (duplication-immune aggregation)", + "query": "SELECT MIN(unit_price) FROM v3.order_details", + "mode": "published", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + + # A COUNT_IF metric: merge=SUM (additive), so it inflates under fan-out exactly + # like SUM/COUNT. Its phase-1 aggregation name is "COUNT_IF" (not SUM/COUNT/AVG), + # so keying on the merge function — not the aggregation name — is what catches it. + response = await client_with_build_v3.post( + "/nodes/metric/", + json={ + "name": "v3.completed_order_count", + "description": "Count of completed order lines (conditional count)", + "query": "SELECT COUNT_IF(status = 'completed') FROM v3.order_details", + "mode": "published", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + + +class TestFanoutGuardWarns: + """Queries that cross a fan-out link with an additive metric must warn (not block).""" + + @pytest.mark.asyncio + async def test_sum_over_one_to_many_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """SUM across a one_to_many link inflates → 200 with an actionable warning.""" + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + warnings = fanout_warnings(response.json()) + assert len(warnings) == 1, response.json() + message = warnings[0]["message"] + assert "v3.total_revenue" in message + assert "fan-out" in message + assert "one_to_many" in message + + @pytest.mark.asyncio + async def test_sum_over_many_to_many_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """SUM across a many_to_many link inflates → 200 with a warning.""" + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.order_channel.channel_name"], + }, + ) + assert response.status_code == 200, response.json() + warnings = fanout_warnings(response.json()) + assert warnings, response.json() + assert "many_to_many" in warnings[0]["message"] + + @pytest.mark.asyncio + async def test_avg_over_one_to_many_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """AVG decomposes into SUM/COUNT components, both inflate → 200 + warning.""" + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.avg_unit_price"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + assert fanout_warnings(response.json()), response.json() + + @pytest.mark.asyncio + async def test_count_if_over_one_to_many_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """COUNT_IF has merge=SUM (additive) → inflates → 200 + warning. + + COUNT_IF's phase-1 aggregation name is not in {SUM,COUNT,AVG}, so a name-based + guard would miss it; keying on the merge function is what catches it. + """ + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.completed_order_count"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + assert fanout_warnings(response.json()), response.json() + + @pytest.mark.asyncio + async def test_metrics_endpoint_also_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """The guard fires on /sql/metrics/v3/ as well, not just measures.""" + response = await client_with_build_v3.get( + "/sql/metrics/v3/", + params={ + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + assert fanout_warnings(response.json()), response.json() + + @pytest.mark.asyncio + async def test_window_metric_over_one_to_many_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """ + A period-over-period (window) metric warns when its base measure grain + crosses a fan-out link. + + Requesting the window's ORDER BY dimension, a second dimension from the + same date node (so it is filtered out of the window grain), plus the + fan-out dimension forces a separate window grain group at the fan-out + grain, so the guard must fire on that grain group too (not just the base + metric path). + """ + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.wow_revenue_change"], + "dimensions": [ + "v3.date.week[order]", + "v3.date.month[order]", + "v3.order_promotion.campaign", + ], + }, + ) + assert response.status_code == 200, response.json() + assert fanout_warnings(response.json()), response.json() + + +class TestFanoutGuardAllows: + """Queries that are safe (or fan-out-immune) must NOT warn.""" + + @pytest.mark.asyncio + async def test_count_distinct_over_one_to_many_ok( + self, + client_with_build_v3, + setup_fanout_links, + ): + """COUNT(DISTINCT) has no merge function → fan-out-immune → no warning.""" + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.order_count"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + assert not fanout_warnings(response.json()), response.json() + + @pytest.mark.asyncio + async def test_min_over_one_to_many_ok( + self, + client_with_build_v3, + setup_fanout_links, + ): + """MIN's merge function is MIN, not additive → duplication-immune → no warning.""" + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.min_unit_price"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + assert not fanout_warnings(response.json()), response.json() + + @pytest.mark.asyncio + async def test_sum_over_safe_many_to_one_ok( + self, + client_with_build_v3, + setup_fanout_links, + ): + """SUM across the default many_to_one customer link does not fan out → no warning.""" + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.customer.name"], + }, + ) + assert response.status_code == 200, response.json() + assert not fanout_warnings(response.json()), response.json() + + @pytest.mark.asyncio + async def test_sum_without_fanout_dimension_ok( + self, + client_with_build_v3, + setup_fanout_links, + ): + """ + The same SUM metric is fine as long as the fan-out dimension is not + requested — the unsafe link is never traversed. + """ + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + assert not fanout_warnings(response.json()), response.json() + + +class TestFanoutGuardCombinedEndpoint: + """The combined endpoint must surface the warning on BOTH the compute-from-source + and the pre-agg (Druid) paths — the pre-agg path builds measures from source + internally, so it has the warning available and must not drop it.""" + + @pytest.mark.asyncio + async def test_combined_source_path_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """SUM across a one_to_many link, combined-from-source → 200 + warning.""" + response = await client_with_build_v3.get( + "/sql/measures/v3/combined", + params={ + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + assert fanout_warnings(response.json()), response.json() + + @pytest.mark.asyncio + async def test_combined_preagg_path_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """ + The pre-agg (Druid) path warns at query time. + + ``use_preagg_tables=true`` reads from pre-agg tables, but the builder still + computes measures from source internally to derive the grain groups, so the + fan-out risk is known and must be surfaced. + """ + response = await client_with_build_v3.get( + "/sql/measures/v3/combined", + params={ + "use_preagg_tables": "true", + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + assert response.json()["use_preagg_tables"] is True + warnings = fanout_warnings(response.json()) + assert len(warnings) == 1, response.json() + assert "one_to_many" in warnings[0]["message"] + + @pytest.mark.asyncio + async def test_combined_preagg_path_without_fanout_ok( + self, + client_with_build_v3, + setup_fanout_links, + ): + """The pre-agg path does not warn when no fan-out dimension is requested.""" + response = await client_with_build_v3.get( + "/sql/measures/v3/combined", + params={ + "use_preagg_tables": "true", + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.order_details.status"], + }, + ) + assert response.status_code == 200, response.json() + assert not fanout_warnings(response.json()), response.json() + + +class TestFanoutGuardPreaggPlan: + """The /preaggs/plan endpoint must surface the warning too — it hands back + materialization SQL that Flow B (user-managed) callers run in their own + engine, so they must see the fan-out risk before materializing inflated + measures.""" + + @pytest.mark.asyncio + async def test_preaggs_plan_over_one_to_many_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """Planning a pre-agg for SUM across a one_to_many link → 201 + warning.""" + response = await client_with_build_v3.post( + "/preaggs/plan", + json={ + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.order_promotion.campaign"], + "strategy": "full", + }, + ) + assert response.status_code == 201, response.json() + warnings = fanout_warnings(response.json()) + assert len(warnings) == 1, response.json() + assert "one_to_many" in warnings[0]["message"] + + @pytest.mark.asyncio + async def test_preaggs_plan_without_fanout_ok( + self, + client_with_build_v3, + setup_fanout_links, + ): + """Planning a pre-agg without a fan-out dimension does not warn.""" + response = await client_with_build_v3.post( + "/preaggs/plan", + json={ + "metrics": ["v3.total_revenue"], + "dimensions": ["v3.product.category"], + "strategy": "full", + }, + ) + assert response.status_code == 201, response.json() + assert not fanout_warnings(response.json()), response.json() diff --git a/datajunction-server/tests/dj_mcp/test_tools.py b/datajunction-server/tests/dj_mcp/test_tools.py index 345b170e3f..bd7f5e700b 100644 --- a/datajunction-server/tests/dj_mcp/test_tools.py +++ b/datajunction-server/tests/dj_mcp/test_tools.py @@ -900,6 +900,57 @@ async def fake_build_measures(**kwargs): assert out.count("Components:") == 0 +@pytest.mark.asyncio +async def test_get_query_plan_renders_warnings( + bound_session_for_tools, + monkeypatch, +) -> None: + """Build warnings (e.g. fan-out risk) are rendered in the plan. + + Covers both the coded-warning branch (uses the ErrorCode name) and the + uncoded fallback (labelled "WARNING"). + """ + coded = MagicMock() + coded.code.name = "FANOUT_RISK" + coded.message = "Possible fan-out on v3.total_revenue" + + uncoded = MagicMock() + uncoded.code = None + uncoded.message = "Something worth noting" + + decomposed = MagicMock() + decomposed.is_derived_for_parents = MagicMock(return_value=False) + decomposed.metric_node.current.query = "SELECT 1" + decomposed.components = [] + + grain = MagicMock() + grain.metrics = ["m1"] + grain.grain = [] + grain.aggregability = MagicMock(value="FULL") + grain.parent_name = "n.fact" + grain.components = [] + grain.sql = "SELECT 1" + + result = MagicMock() + result.dialect = MagicMock(value="spark") + result.requested_dimensions = [] + result.grain_groups = [grain] + result.decomposed_metrics = {"m1": decomposed} + result.warnings = [coded, uncoded] + result.ctx.parent_map = {} + result.ctx.nodes = {} + + async def fake_build_measures(**kwargs): + return result + + monkeypatch.setattr(tools, "build_measures_sql", fake_build_measures) + + out = await tools.get_query_plan(metrics=["m1"]) + assert "⚠ Warnings" in out + assert "[FANOUT_RISK] Possible fan-out on v3.total_revenue" in out + assert "[WARNING] Something worth noting" in out + + # --------------------------------------------------------------------------- # tools.get_node_lineage — direction-specific paths # --------------------------------------------------------------------------- diff --git a/datajunction-server/tests/models/deployment_test.py b/datajunction-server/tests/models/deployment_test.py index f641d312ae..9dce26d16f 100644 --- a/datajunction-server/tests/models/deployment_test.py +++ b/datajunction-server/tests/models/deployment_test.py @@ -436,6 +436,38 @@ def test_dimension_join_link_spec_with_default_value(): assert hash(link_spec) != hash(different_default) +def test_dimension_join_link_spec_with_join_cardinality(): + """Test DimensionJoinLinkSpec join_cardinality default, equality, and hashing.""" + from datajunction_server.models.dimensionlink import JoinCardinality + + # Default cardinality is many_to_one (a fact row joins at most one dim row). + default_link = DimensionJoinLinkSpec( + dimension_node="some.dimension.users", + join_type="left", + join_on="events.user_id = some.dimension.users.id", + ) + assert default_link.join_cardinality == JoinCardinality.MANY_TO_ONE + + fanout_link = DimensionJoinLinkSpec( + dimension_node="some.dimension.users", + join_type="left", + join_on="events.user_id = some.dimension.users.id", + join_cardinality=JoinCardinality.ONE_TO_MANY, + ) + assert fanout_link.join_cardinality == JoinCardinality.ONE_TO_MANY + + # Equality and hashing must take join_cardinality into account. + same_default = DimensionJoinLinkSpec( + dimension_node="some.dimension.users", + join_type="left", + join_on="events.user_id = some.dimension.users.id", + ) + assert default_link == same_default + assert hash(default_link) == hash(same_default) + assert default_link != fanout_link + assert hash(default_link) != hash(fanout_link) + + def test_source_spec_with_dimension_link_default_value(): """Test SourceSpec with dimension_links including default_value.""" source_spec = SourceSpec( From 4b374b19aca23d520661a41cd1dcb980328337d0 Mon Sep 17 00:00:00 2001 From: Robin Davis Date: Thu, 23 Jul 2026 07:49:30 -0700 Subject: [PATCH 2/2] feat: extend fan-out guard to non-decomposable metrics The guard keyed only on an additive (SUM) merge, so it missed holistic aggregations that are inflated by row duplication: MEDIAN, PERCENTILE, ARRAY_AGG, etc. check_fanout_safety now also inspects grain_group.non_decomposable_metrics and warns on them --- .../construction/build_v3/dimensions.py | 52 ++++++++++---- .../datajunction_server/sql/decompose.py | 19 ++++++ .../build_v3/fanout_guard_test.py | 67 +++++++++++++++++++ .../tests/sql/decompose_test.py | 21 ++++++ 4 files changed, 147 insertions(+), 12 deletions(-) diff --git a/datajunction-server/datajunction_server/construction/build_v3/dimensions.py b/datajunction-server/datajunction_server/construction/build_v3/dimensions.py index 6b1d5934cf..f1d13260b0 100644 --- a/datajunction-server/datajunction_server/construction/build_v3/dimensions.py +++ b/datajunction-server/datajunction_server/construction/build_v3/dimensions.py @@ -26,6 +26,7 @@ ) from datajunction_server.construction.build_v3.types import ( BuildContext, + DecomposedMetricInfo, DimensionRef, GrainGroup, JoinPath, @@ -33,8 +34,8 @@ ) from datajunction_server.database.dimensionlink import DimensionLink from datajunction_server.database.node import Node, NodeType -from datajunction_server.models.decompose import MetricComponent 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 @@ -1049,30 +1050,57 @@ def find_unsafe_cardinality_links( 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 re-aggregates a component with an - additive merge function across a fanning join. - - A ONE_TO_MANY/MANY_TO_MANY link duplicates fact rows, inflating additive (SUM) - merges. We warn rather than error: the user may know the data does not fan out. + 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 - # Components with fan-out risk: those with a SUM merge - inflated: list[tuple[Node, MetricComponent]] = [ - (metric_node, component) + # 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" - ] - if not inflated: + } + + # 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({metric_node.name for metric_node, _ in inflated}) + metric_names = sorted(inflated_metrics) link_descriptions = sorted( { f"{link.node_revision.name} -> {link.dimension.name} " diff --git a/datajunction-server/datajunction_server/sql/decompose.py b/datajunction-server/datajunction_server/sql/decompose.py index cdbcbecf7b..a61027be9f 100644 --- a/datajunction-server/datajunction_server/sql/decompose.py +++ b/datajunction-server/datajunction_server/sql/decompose.py @@ -646,6 +646,25 @@ def get_decomposition(func_class: type) -> AggDecomposition | None: return decomp_class() +def is_duplication_invariant(func_class: type) -> bool: + """ + Whether row duplication (e.g. from a fan-out join) leaves this aggregation's + result unchanged. + + Derived from the merge function for decomposable aggregations: a SUM merge sums + duplicated rows and inflates, so anything without a SUM merge is invariant. + Argmax-style MAX_BY/MIN_BY are invariant too (they read a single extreme row) + but are non-decomposable, so they're named explicitly. Everything else + non-decomposable is assumed sensitive (fail closed). + """ + if func_class in (dj_functions.MaxBy, dj_functions.MinBy): + return True + decomposition = get_decomposition(func_class) + if decomposition is None: + return False + return all(component.merge != "SUM" for component in decomposition.components) + + # ============================================================================= # Decomposition Result # ============================================================================= diff --git a/datajunction-server/tests/construction/build_v3/fanout_guard_test.py b/datajunction-server/tests/construction/build_v3/fanout_guard_test.py index 213e78353c..fd158534a9 100644 --- a/datajunction-server/tests/construction/build_v3/fanout_guard_test.py +++ b/datajunction-server/tests/construction/build_v3/fanout_guard_test.py @@ -155,6 +155,33 @@ async def setup_fanout_links(client_with_build_v3): ) assert response.status_code in (200, 201, 409), response.json() + # A MEDIAN metric: non-decomposable (holistic), so it builds via raw + # passthrough and is aggregated over the duplicated rows. It depends on the + # multiset of rows, so a fan-out distorts it and the guard MUST fire. + response = await client_with_build_v3.post( + "/nodes/metric/", + json={ + "name": "v3.median_unit_price", + "description": "Median unit price (non-decomposable, duplication-sensitive)", + "query": "SELECT MEDIAN(unit_price) FROM v3.order_details", + "mode": "published", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + + # A MAX_BY metric: non-decomposable too, but argmax reads a single extreme + # row, so duplication can't change it. It must NOT trip the guard. + response = await client_with_build_v3.post( + "/nodes/metric/", + json={ + "name": "v3.unit_price_at_max_order", + "description": "Unit price of the highest order_id (duplication-immune argmax)", + "query": "SELECT MAX_BY(unit_price, order_id) FROM v3.order_details", + "mode": "published", + }, + ) + assert response.status_code in (200, 201, 409), response.json() + class TestFanoutGuardWarns: """Queries that cross a fan-out link with an additive metric must warn (not block).""" @@ -238,6 +265,29 @@ async def test_count_if_over_one_to_many_warns( assert response.status_code == 200, response.json() assert fanout_warnings(response.json()), response.json() + @pytest.mark.asyncio + async def test_median_over_one_to_many_warns( + self, + client_with_build_v3, + setup_fanout_links, + ): + """A non-decomposable holistic metric (MEDIAN) inflates → 200 + warning. + + MEDIAN has no components and no SUM merge, so the merge check alone can't + see it; it's caught via the grain group's non-decomposable metrics. + """ + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.median_unit_price"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + warnings = fanout_warnings(response.json()) + assert len(warnings) == 1, response.json() + assert "v3.median_unit_price" in warnings[0]["message"] + @pytest.mark.asyncio async def test_metrics_endpoint_also_warns( self, @@ -323,6 +373,23 @@ async def test_min_over_one_to_many_ok( assert response.status_code == 200, response.json() assert not fanout_warnings(response.json()), response.json() + @pytest.mark.asyncio + async def test_max_by_over_one_to_many_ok( + self, + client_with_build_v3, + setup_fanout_links, + ): + """MAX_BY is non-decomposable but argmax reads one extreme row → immune → no warning.""" + response = await client_with_build_v3.get( + "/sql/measures/v3/", + params={ + "metrics": ["v3.unit_price_at_max_order"], + "dimensions": ["v3.order_promotion.campaign"], + }, + ) + assert response.status_code == 200, response.json() + assert not fanout_warnings(response.json()), response.json() + @pytest.mark.asyncio async def test_sum_over_safe_many_to_one_ok( self, diff --git a/datajunction-server/tests/sql/decompose_test.py b/datajunction-server/tests/sql/decompose_test.py index 0a3117e573..cbffdac5f8 100644 --- a/datajunction-server/tests/sql/decompose_test.py +++ b/datajunction-server/tests/sql/decompose_test.py @@ -17,6 +17,7 @@ from datajunction_server.sql import functions as dj_functions from datajunction_server.sql.decompose import ( MetricComponentExtractor, + is_duplication_invariant, safe_denominator, wrap_divisions_in_nullif, ) @@ -2303,6 +2304,26 @@ def test_real_aggregations_marked_as_aggregation(self): ) +def test_is_duplication_invariant(): + """ + Duplication-invariance is derived from the merge function where decomposable + (SUM merge inflates, everything else is invariant), stated for MAX_BY/MIN_BY, + and assumed False for other non-decomposable (holistic) aggregations. + """ + # Derived from a non-SUM merge -> invariant. + assert is_duplication_invariant(dj_functions.Max) is True + assert is_duplication_invariant(dj_functions.Min) is True + assert is_duplication_invariant(dj_functions.ApproxCountDistinct) is True + # Derived from a SUM merge -> inflates. + assert is_duplication_invariant(dj_functions.Sum) is False + assert is_duplication_invariant(dj_functions.Count) is False + assert is_duplication_invariant(dj_functions.Avg) is False + # Non-decomposable: stated invariant for argmax/argmin, sensitive otherwise. + assert is_duplication_invariant(dj_functions.MaxBy) is True + assert is_duplication_invariant(dj_functions.MinBy) is True + assert is_duplication_invariant(dj_functions.Median) is False + + @pytest.mark.asyncio async def test_sum_abs_decomposes(session: AsyncSession, create_metric): """SUM(ABS(x)) decomposes into one SUM component over ABS(x)."""