class QueryResolver:
"""Resolves a QueryObject + SemanticModel into a ResolvedQuery."""
def resolve(
self,
query: QueryObject,
model: SemanticModel,
qualify_table: Callable[[DataObject], str] | None = None,
) -> ResolvedQuery:
ctx = _ResolutionContext(
model=model,
result=ResolvedQuery(
limit=query.limit,
offset=query.offset,
use_path_names=list(query.use_path_names),
allow_fan_out=query.allow_fan_out,
is_raw=query.select.is_raw,
distinct=query.select.distinct,
grouping=query.grouping,
fill=query.fill,
),
qualify_table=qualify_table,
)
# Build global column lookup: col_name → (object_name, source_column)
for obj_name, obj in model.data_objects.items():
for col_name, col_obj in obj.columns.items():
ctx.global_columns[col_name] = (obj_name, col_obj.code)
if query.select.is_raw:
# Raw mode: project physical columns, no aggregation.
for ref in query.select.fields:
self._resolve_raw_field(ctx, ref)
else:
# Aggregate mode (default).
# 1. Resolve dimensions (string or coalesce group).
# Coalesce groups expand into their constituent dimensions, each
# tagged with the same coalesce_alias so the CFL outer wrapper can
# emit COALESCE(d1, d2, ...) AS <alias>.
for dim_entry in query.select.dimensions:
if isinstance(dim_entry, CoalesceDimension):
self._resolve_coalesce_dimension(ctx, dim_entry, ctx.result.coalesce_aliases)
else:
self._append_resolved_dimension(ctx, dim_entry)
# 2. Resolve measures and track their source objects
selected_measures: set[str] = set()
for measure_name in query.select.measures:
# A measure or metric named twice is one column named twice.
if measure_name in selected_measures:
ctx.errors.append(
SemanticError(
code="DUPLICATE_MEASURE",
message=f"'{measure_name}' is selected twice; list each measure once",
path="select.measures",
)
)
continue
selected_measures.add(measure_name)
resolved_meas = self._resolve_measure(ctx, measure_name)
if resolved_meas:
ctx.result.measures.append(resolved_meas)
source_objs = self._get_measure_source_objects(ctx, measure_name)
ctx.result.measure_source_objects.update(source_objs)
ctx.result.required_objects.update(source_objs)
ctx.result.required_objects.update(
self._get_measure_join_objects(ctx, measure_name)
)
# 2.5. Auto-include measures referenced by HAVING but not by SELECT.
# Without this, codegen emits a HAVING clause that references an
# alias for a column the SELECT doesn't project — every database
# rejects the SQL with a "must appear in GROUP BY" binder error.
# Routing this through the regular measure-resolution path also
# updates ``measure_source_objects`` so the multi-fact CFL trigger
# below sees the HAVING-only measure's source.
existing_measure_names = {m.name for m in ctx.result.measures}
for ref in self._collect_having_measure_refs(query, model):
if ref in existing_measure_names:
continue
resolved_meas = self._resolve_measure(ctx, ref)
if resolved_meas is None:
continue
ctx.result.measures.append(resolved_meas)
ctx.result.having_only_measures.add(ref)
existing_measure_names.add(ref)
source_objs = self._get_measure_source_objects(ctx, ref)
ctx.result.measure_source_objects.update(source_objs)
ctx.result.required_objects.update(source_objs)
ctx.result.required_objects.update(self._get_measure_join_objects(ctx, ref))
# 2.6. A window over a reaggregate metric, and a formula reading one,
# is computed after the reaggregate pass from finished columns; the
# ones not selected are included here and dropped from the output
# by that pass (``late_window``).
self._include_late_window_inputs(ctx)
# 3. Determine base object (the one with most joins / most measures).
# WHERE filters are resolved much later, so the objects they reference
# are collected up front — the base has to be able to reach them, or
# the filter is silently dropped as unreachable further down.
where_filter_objects = self._collect_where_filter_objects(query, model)
ctx.result.base_object = self._select_base_object(ctx, where_filter_objects)
if ctx.result.base_object:
ctx.result.required_objects.add(ctx.result.base_object)
# An anchored measure pins the base, which bypasses the re-anchoring that
# normally makes a filter's data object reachable. A predicate on a fact
# the anchor only reaches by conforming chooses that fact's rows, so it
# is applied inside the fact's conformed subquery, before the rows are
# aggregated to the shared key (``_push_filters_into_conformed_facts``).
# Recorded here, before the filters are resolved, so resolving one does
# not try to join such an object from the base.
for name in ctx.result.anchored_measures:
anchored = model.effective_measures.get(name)
if anchored is not None:
ctx.conformed_reach[name] = conformed_reach(
model, anchored, ctx.result.use_path_names
)
self._warn_repeated_conformed_terms(ctx)
# Detect multi-fact: CFL is needed only when measure source objects
# span multiple independent fact tables.
if len(ctx.result.measure_source_objects) > 1:
graph = JoinGraph.of(model, use_path_names=query.use_path_names or None)
reachable = graph.descendants(ctx.result.base_object)
unreachable = ctx.result.measure_source_objects - reachable - {ctx.result.base_object}
if unreachable:
ctx.result.requires_cfl = True
# Dimension-only queries: when dimensions span independent branches,
# join through intermediate bridge/fact tables. Where several roots
# join them (orders and returns both reaching customers and products),
# each holds its own combinations, and picking one by name answered
# "ordered pairs" or "returned pairs" depending on what the facts were
# called. CFL takes a leg from each root and the union of what they
# hold. With one root, add the intermediate tables from the join steps
# to required_objects so the star schema planner includes them.
if not ctx.result.measure_source_objects and ctx.result.dimensions:
dim_objects = {d.object_name for d in ctx.result.dimensions}
if not dim_objects <= {ctx.result.base_object}:
graph = JoinGraph.of(model, use_path_names=query.use_path_names or None)
# Each leg has to reach a role dimension's waypoint, the objects
# a computed dimension reads and those the WHERE names, or it
# would answer without them.
vias = {d.via for d in ctx.result.dimensions if d.via}
computed = {
obj
for d in ctx.result.dimensions
for obj in model.column_reference_objects(d.object_name, d.column_name)
}
leg_objects = dim_objects | vias | computed | where_filter_objects
roots = graph.minimal_roots(leg_objects)
if len(roots) > 1:
ctx.result.requires_cfl = True
ctx.result.dimension_roots = roots
# Every leg's tables are read, and the result cache is
# invalidated by the tables in ``required_objects``.
for root in roots:
ctx.result.required_objects.add(root)
for step in graph.find_join_path(
{root},
leg_objects,
via_constraints=ctx.result.via_constraints or None,
):
ctx.result.required_objects.add(step.from_object)
ctx.result.required_objects.add(step.to_object)
else:
steps = graph.find_join_path(
{ctx.result.base_object},
dim_objects,
via_constraints=ctx.result.via_constraints or None,
)
for step in steps:
ctx.result.required_objects.add(step.from_object)
ctx.result.required_objects.add(step.to_object)
# Raw mode: detect multi-fact (fields span objects unreachable from
# the base via directed joins). The pipeline rejects this case for
# now — raw CFL is a planned follow-up.
if ctx.result.is_raw and ctx.result.base_object:
field_objects = {f.object_name for f in ctx.result.fields}
if len(field_objects) > 1:
graph = JoinGraph.of(model, use_path_names=query.use_path_names or None)
reachable = graph.descendants(ctx.result.base_object)
unreachable = field_objects - reachable - {ctx.result.base_object}
if unreachable:
ctx.result.requires_cfl = True
# Validate dimensionsExclude constraints
if query.dimensions_exclude:
if query.select.measures:
ctx.errors.append(
SemanticError(
code="DIMENSIONS_EXCLUDE_WITH_MEASURES",
message="dimensionsExclude cannot be combined with measures",
path="select",
)
)
elif len(ctx.result.dimensions) < 2:
ctx.errors.append(
SemanticError(
code="DIMENSIONS_EXCLUDE_INSUFFICIENT",
message="dimensionsExclude requires at least 2 dimensions",
path="select.dimensions",
)
)
else:
ctx.result.dimensions_exclude = True
# Every object this *query* names, including the ones only a predicate
# does. A WHERE filter is resolved much later, so a guard reading
# ``required_objects`` alone sees none of them - and a filter is exactly
# how a nested object reaches a query that projects nothing from it.
#
# A static model filter is deliberately not counted. It is a property of
# the model rather than of the query, and one naming an object this plan
# cannot reach is documented as skipped rather than fatal
# (``test_unreachable_filter_silently_ignored``). Counting them made a
# single nested static filter refuse every multi-fact query in the
# model, including the ones that never go near it.
self._reject_unsupported_nested_shapes(
ctx, ctx.result.required_objects | where_filter_objects
)
# 4. Validate usePathNames before building join graph
self._validate_use_path_names(ctx, query.use_path_names)
# 5. Resolve join paths
ctx.graph = JoinGraph.of(model, use_path_names=query.use_path_names or None)
if ctx.result.base_object and len(ctx.result.required_objects) > 1:
ambiguous: dict[str, list[list[str]]] = {}
ctx.result.join_steps = ctx.graph.find_join_path(
{ctx.result.base_object},
ctx.result.required_objects,
via_constraints=ctx.result.via_constraints or None,
ambiguous=ambiguous,
)
# A dimension the query reaches by two equally close routes is two
# different roles of one data object, and they select different
# rows. Refused rather than picked — the same stance the filter
# path takes, since a projected dimension is no more guessable.
for object_name, routes in sorted(ambiguous.items()):
names = (
", ".join(
f"'{dim.name}'"
for dim in ctx.result.dimensions
if dim.object_name == object_name
)
or f"'{object_name}'"
)
ctx.errors.append(
SemanticError(
code="AMBIGUOUS_JOIN_PATH",
message=(
f"{names} is on '{object_name}', which this query reaches "
f"equally well by more than one route "
f"({', '.join(f'via {path[-2]!r}' for path in routes)}). "
f"Those are different roles of the same data object and "
f"they select different rows."
),
path="select.dimensions",
hint=(
"Say which one is meant: declare a data object per role "
"over the same table and select from that, or give the "
"dimension a 'via:' waypoint naming the object the join "
"must traverse."
),
)
)
# Build set of all objects present in the query's join graph
if ctx.result.base_object:
ctx.joined_objects.add(ctx.result.base_object)
for step in ctx.result.join_steps:
ctx.joined_objects.add(step.to_object)
# Detect required objects that the star-schema planner cannot reach.
# Many-to-one joins are forward-only (reverse traversal would inflate
# the base table), so a required object that's only reachable via a
# reverse m-to-1 hop is unreachable. Raise a clear error rather than
# silently producing wrong SQL. CFL legs are validated separately.
if ctx.result.base_object and not ctx.result.requires_cfl:
unreachable = ctx.result.required_objects - ctx.joined_objects
for unreachable_name in sorted(unreachable):
ctx.errors.append(
SemanticError(
code="UNREACHABLE_REQUIRED_OBJECT",
message=(
f"Data object '{unreachable_name}' is required by the query but "
f"cannot be reached from base '{ctx.result.base_object}' via "
f"directed joins. Many-to-one joins are forward-only; reverse "
f"traversal would inflate row counts. Add an explicit join from "
f"'{ctx.result.base_object}' (or an intermediate object) to "
f"'{unreachable_name}', or split the query so each fact is "
f"queried independently."
),
path="select",
)
)
# 5b. Inject static model filters — always applied as WHERE conditions
static_exprs: list[Expr] = []
for mf in model.filters:
static_filter = self._resolve_static_filter(ctx, mf)
if static_filter:
ctx.result.where_filters.append(static_filter)
static_exprs.append(static_filter.expression)
# 6. Classify filters — a query-time duplicate of a static filter is
# not added twice, but the one kept now carries the query's demand: a
# static filter may be skipped where irrelevant, a query filter may not.
for qfi in query.where:
resolved_filter = self._resolve_filter_item(ctx, qfi, is_having=False)
if not resolved_filter:
continue
if resolved_filter.expression not in static_exprs:
ctx.result.where_filters.append(resolved_filter)
continue
ctx.result.where_filters = [
replace(wf, static=False)
if wf.static and wf.expression == resolved_filter.expression
else wf
for wf in ctx.result.where_filters
]
for qfi in query.having:
resolved_filter = self._resolve_filter_item(ctx, qfi, is_having=True)
if resolved_filter:
ctx.result.having_filters.append(resolved_filter)
self._push_filters_into_conformed_facts(ctx)
self._plan_rows_without_children(ctx)
# 7. Resolve order by — a dimension or measure in SELECT, or a measure
# or metric the query does not select.
select_count = len(ctx.result.dimensions) + len(ctx.result.measures)
selected = {d.name for d in ctx.result.dimensions} | {m.name for m in ctx.result.measures}
for ob in query.order_by:
if self._order_only(ctx, ob.field, selected):
continue
expr = self._resolve_order_by_field(ctx, ob.field, select_count)
if expr:
ctx.result.order_by_exprs.append((expr, ob.direction == "desc", ob.nulls))
# 8. ROLLUP / CUBE: backfill NULLS FIRST on any explicit ORDER BY entry
# that didn't specify a NULLs position. Subtotal and grand-total rows
# carry NULLs in the rolled-up group-by columns, and BI tools expect
# those totals at the top of the result — not interleaved with details.
if ctx.result.grouping is not None and ctx.result.order_by_exprs:
ctx.result.order_by_exprs = [
(expr, desc, NullsPosition.FIRST if nulls is None else nulls)
for expr, desc, nulls in ctx.result.order_by_exprs
]
# 9. Auto-order — when no explicit ORDER BY, append ORDER BY over all
# SELECT dimensions (or raw fields) under two conditions:
# (a) LIMIT is set: cache hashes on compiled SQL; without ORDER BY
# ``LIMIT N`` returns any N rows, freezing one arbitrary slice.
# (b) ROLLUP / CUBE: subtotal layout is otherwise unpredictable.
# ROLLUP / CUBE defaults to NULLS FIRST (totals at the top).
# Aggregate-only queries (no dims, no fields) are already single-row
# deterministic — skip.
needs_auto_order = not ctx.result.order_by_exprs and (
ctx.result.limit is not None or ctx.result.grouping is not None
)
if needs_auto_order:
nulls_default = NullsPosition.FIRST if ctx.result.grouping is not None else None
if ctx.result.is_raw and ctx.result.fields:
for f in ctx.result.fields:
ctx.result.order_by_exprs.append(
(ColumnRef(name=f.alias), False, nulls_default)
)
elif ctx.result.dimensions:
for dim in ctx.result.dimensions:
ctx.result.order_by_exprs.append(
(ColumnRef(name=dim.name), False, nulls_default)
)
if query.fill:
self._check_fill(ctx, query)
if ctx.errors:
raise ResolutionError(ctx.errors)
selected = {d.name for d in ctx.result.dimensions}
# A cumulative metric only sorted by reads asOf too (``order_only``).
sorted_cumulative = any(
(metric := model.metrics.get(name)) is not None
and metric.type == MetricType.CUMULATIVE
and metric.time_dimension not in selected
for name in ctx.result.order_only_measures
)
if (
query.as_of is not None
and not sorted_cumulative
and not any(
m.is_cumulative and m.cumulative_time_dimension not in selected
for m in ctx.result.measures
)
):
ctx.result.warnings.append(
warning(
code=WarningCode.CUMULATIVE_CONSTRAINT_VIOLATED,
message=(
f"asOf {query.as_of.isoformat()} has no effect: it applies to a "
f"cumulative metric whose timeDimension the query does not select, "
f"and this query has none."
),
hint=(
"Drop asOf, or drop the time dimension to evaluate the cumulative "
"metrics as of that date."
),
context={"asOf": query.as_of.isoformat()},
)
)
return ctx.result
# -- raw mode fields -----------------------------------------------------
def _resolve_raw_field(self, ctx: _ResolutionContext, ref: str) -> None:
"""Resolve a ``DataObject.Column`` reference for raw-mode projection.
Errors are accumulated in the resolution context (raised at the end).
"""
raw_resolution.resolve_raw_field(self, ctx, ref)
# -- dimensions ----------------------------------------------------------
def _append_resolved_dimension(
self,
ctx: _ResolutionContext,
dim_str: str,
coalesce_alias: str | None = None,
) -> ResolvedDimension | None:
"""Resolve a single dimension string and append it to the result."""
try:
dim_ref = DimensionRef.parse(dim_str)
except ValueError:
# Pydantic's "'day_of_week' is not a valid TimeGrain" escaped as an
# untyped error, naming neither the entry nor the grains there are.
grain = dim_str.rsplit(":", 1)[1]
ctx.errors.append(
SemanticError(
code="INVALID_TIME_GRAIN",
message=(
f"Unknown time grain '{grain}' in '{dim_str}'; use one of "
f"{', '.join(g.value for g in TimeGrain)}"
),
path="select.dimensions",
)
)
return None
resolved_dim = self._resolve_dimension(ctx, dim_ref)
if resolved_dim is None:
return None
dim_def = ctx.model.dimensions.get(dim_ref.name)
if dim_def and dim_def.via:
resolved_dim.via = dim_def.via
ctx.result.required_objects.add(dim_def.via)
ctx.result.via_constraints[resolved_dim.object_name] = dim_def.via
if coalesce_alias is not None:
resolved_dim.coalesce_alias = coalesce_alias
else:
# The same dimension at the same grain twice is one column named
# twice: a client reading the result by name sees only one of them.
earlier = next(
(
d
for d in ctx.result.dimensions
if d.coalesce_alias is None
and (d.dimension, d.grain) == (resolved_dim.dimension, resolved_dim.grain)
),
None,
)
if earlier is not None:
ctx.errors.append(
SemanticError(
code="DUPLICATE_DIMENSION",
message=(
f"'{dim_str}' selects the same column as "
f"'{dimension_reference(earlier)}'; list each dimension once"
),
path="select.dimensions",
)
)
return None
# The same dimension at a second grain: a column of its own, or
# every wrapper joining and grouping by name reads the first twice.
resolved_dim.name = column_name(
resolved_dim.dimension,
resolved_dim.grain,
[(d.dimension, d.grain) for d in ctx.result.dimensions if d.coalesce_alias is None],
)
ctx.result.dimensions.append(resolved_dim)
ctx.result.required_objects.add(resolved_dim.object_name)
# A computed column may read a column of another data object, which the
# plan then has to join — the expression is inlined into the SELECT list
# and would otherwise name an alias nothing in the FROM chain binds.
ctx.result.required_objects.update(
ctx.model.column_reference_objects(resolved_dim.object_name, resolved_dim.column_name)
)
return resolved_dim
def _resolve_coalesce_dimension(
self,
ctx: _ResolutionContext,
coalesce: CoalesceDimension,
seen_aliases: set[str],
) -> None:
"""Expand a coalesce group into its constituent resolved dimensions.
Validates: at least 2 members, alias is unique within the query and
does not collide with an existing dimension/measure name, all members
resolve to the same abstract column type.
"""
alias = coalesce.alias
if not alias:
ctx.errors.append(
SemanticError(
code="COALESCE_MISSING_ALIAS",
message="Coalesce dimension requires a non-empty 'as' alias",
path="select.dimensions",
)
)
return
if alias in seen_aliases:
ctx.errors.append(
SemanticError(
code="DUPLICATE_COALESCE_ALIAS",
message=f"Duplicate coalesce alias '{alias}' in this query",
path="select.dimensions",
)
)
return
if alias in ctx.model.dimensions or alias in ctx.model.effective_measures:
ctx.errors.append(
SemanticError(
code="COALESCE_ALIAS_COLLISION",
message=(
f"Coalesce alias '{alias}' collides with an existing "
f"model dimension or measure name"
),
path="select.dimensions",
)
)
return
if len(coalesce.coalesce) < 2:
ctx.errors.append(
SemanticError(
code="COALESCE_TOO_FEW_MEMBERS",
message=(
f"Coalesce '{alias}' requires at least 2 dimensions "
f"(got {len(coalesce.coalesce)})"
),
path="select.dimensions",
)
)
return
seen_aliases.add(alias)
# Resolve each member with the alias tag; verify type compatibility.
member_types: set[str] = set()
for member in coalesce.coalesce:
resolved = self._append_resolved_dimension(ctx, member, coalesce_alias=alias)
if resolved:
dim_def = ctx.model.dimensions.get(member)
if dim_def:
member_types.add(dim_def.result_type.value)
if len(member_types) > 1:
ctx.errors.append(
SemanticError(
code="COALESCE_TYPE_MISMATCH",
message=(
f"Coalesce '{alias}' members have incompatible result types: "
f"{sorted(member_types)}"
),
path="select.dimensions",
)
)
def _resolve_dimension(
self, ctx: _ResolutionContext, ref: DimensionRef
) -> ResolvedDimension | None:
"""Resolve a dimension reference to its physical column."""
dim = ctx.model.dimensions.get(ref.name)
if dim is None:
kind = artefact_kind(ctx.model, ref.name)
ctx.errors.append(
SemanticError(
code="UNKNOWN_DIMENSION",
message=f"Unknown dimension '{ref.name}'",
path="select.dimensions",
hint=(
f"'{ref.name}' is a {kind}: list it under select.measures."
if kind in ("measure", "metric")
else None
),
suggestions=(
[] if kind else suggest_names(ref.name, dimension_names(ctx.model))
),
context={"dimension": ref.name},
)
)
return None
obj_name = dim.view
col_name = dim.column
obj = ctx.model.data_objects.get(obj_name)
if obj is None:
ctx.errors.append(
SemanticError(
code="UNKNOWN_DATA_OBJECT",
message=f"Dimension '{ref.name}' references unknown data object '{obj_name}'",
)
)
return None
vf = obj.columns.get(col_name)
source_col = vf.code if vf else col_name
grain = ref.grain or dim.time_grain
if grain is not None and not self._grain_fits_the_column(ctx, ref, dim, vf, grain):
return None
if grain is not None and not self._grain_survives_the_cast(ctx, ref, dim, grain):
return None
if grain is not None and not self._grain_fits_the_values(ctx, ref, dim, vf, grain):
return None
return ResolvedDimension(
name=ref.name,
object_name=obj_name,
column_name=col_name,
source_column=source_col,
grain=grain,
)
@staticmethod
def _grain_fits_the_column(
ctx: _ResolutionContext,
ref: DimensionRef,
dim: Dimension,
column: DataObjectColumn | None,
grain: TimeGrain,
) -> bool:
"""Refuse a grain over a column that carries no date to truncate.
A grain compiles to ``date_trunc(grain, column)``, which the engine
refuses over text: measured on DuckDB, ``"Name:hour"`` over a string
column compiled and then died in the binder with "No function matches
the given name and argument types 'date_trunc(STRING_LITERAL,
VARCHAR)'". The model validator refuses a declared ``timeGrain`` over
such a column for exactly that reason, and a query can name a grain the
model never declared, so the same rule is read here - an error naming
the dimension and the column beats one naming generated SQL.
A column the model does not define is left alone: its type is unknown
here, and the reference itself is reported elsewhere.
"""
if column is None or column.abstract_type in DATE_BEARING_TYPES:
return True
ctx.errors.append(
SemanticError(
code="TIME_GRAIN_ON_NON_TEMPORAL",
message=(
f"Dimension '{ref.name}' is asked for at grain '{grain.value}' "
f"but underlying column '{dim.view}.{dim.column}' has "
f"abstractType '{column.abstract_type.value}'. A time grain "
f"requires the column to be date, timestamp, or timestamp_tz. "
f"Ask for the dimension without a grain, fix the column's "
f"abstractType, or define a computed column with to_date()."
),
path="select.dimensions",
)
)
return False
@staticmethod
def _grain_fits_the_values(
ctx: _ResolutionContext,
ref: DimensionRef,
dim: Dimension,
column: DataObjectColumn | None,
grain: TimeGrain,
) -> bool:
"""Refuse a grain finer than the one the column's values already hold.
``date_trunc('day', <month start>)`` is the month start: over a column
of month starts, ``Order Month:day`` answered one row per month under
a label that says day, without an error. The model validator holds the
grains a model names to the same rule (``grain_derives``); a query
names its own.
"""
held = column.held_grain if column is not None else None
if held is None or grain_derives(grain, held):
return True
ctx.errors.append(
SemanticError(
code="GRAIN_FINER_THAN_DECLARED",
message=(
f"Dimension '{ref.name}' is asked for at grain '{grain.value}' "
f"but column '{dim.view}.{dim.column}' holds '{held.value}' "
f"values: {held_grain_cost(grain, held)}"
),
path="select.dimensions",
)
)
return False
@staticmethod
def _grain_survives_the_cast(
ctx: _ResolutionContext, ref: DimensionRef, dim: Dimension, grain: TimeGrain
) -> bool:
"""Refuse a grain the dimension's declared type cannot hold.
``make_dimension_expr`` casts a grained dimension to its declared
``resultType``, in the GROUP BY as well as the projection, so a
declaration narrower than the grain merges buckets and changes the
measures rather than relabelling the column. The model validator refuses
that combination at load, but a query writes its own: ``Occurred:hour``
names a grain the dimension never declared, so a dimension declaring
``date`` -- perfectly valid, with no ``timeGrain`` of its own -- answered
two rows where three were asked for, the two hours of one day summed
into one. The model is not at fault there and cannot be checked for it;
the query is, and this is where it is read.
``time_tz`` is refused here too, and it is the one case that merges
nothing: OBML has no cast target for it, so the value is left alone and
only the label is wrong. Refused all the same, because a grain always
carries a date and that declaration cannot describe one.
"""
declared = dim.result_type
if result_type_holds_grain(grain, declared):
return True
keeps = "timestamp" if grain in SUB_DAY_GRAINS else "date or timestamp"
asked = "asked for at" if ref.grain is not None else "grouped by"
if declared in CASTABLE_TEMPORAL_TYPES:
cost = (
"The cast is applied in the GROUP BY as well, so buckets would "
"merge and the measures would change without an error."
)
else:
cost = (
f"A grain always carries a date, and OBML has no cast target for "
f"'{declared.value}', so the dimension would answer a date-bearing "
f"value under a label for a time."
)
ctx.errors.append(
SemanticError(
code="RESULT_TYPE_LOSES_GRAIN",
message=(
f"Dimension '{ref.name}' is {asked} grain '{grain.value}' but "
f"declares resultType '{declared.value}', which cannot hold it. "
f"{cost} Declare {keeps}, or ask for the grain the type implies."
),
path="select.dimensions",
)
)
return False
# -- measures & metrics --------------------------------------------------
def _resolve_measure(self, ctx: _ResolutionContext, name: str) -> ResolvedMeasure | None:
"""Resolve a measure name to its aggregate expression."""
measure = ctx.model.effective_measures.get(name)
if measure is None:
metric = ctx.model.metrics.get(name)
if metric:
return self._resolve_metric(ctx, name, metric)
ctx.errors.append(
SemanticError(
code="UNKNOWN_MEASURE",
message=f"Unknown measure '{name}'",
path="select.measures",
hint=(
f"'{name}' is a dimension: list it under select.dimensions."
if name in ctx.model.dimensions
else None
),
suggestions=(
[]
if name in ctx.model.dimensions
else suggest_names(name, measure_names(ctx.model))
),
context={"measure": name},
)
)
return None
expr = self._build_measure_expr(ctx, measure)
grain_override = measure.grain
effective_grain: list[str] | None = None
if grain_override is not None:
query_dim_names = [d.name for d in ctx.result.dimensions]
effective_grain = _resolve_effective_grain(grain_override, ctx.result.dimensions)
if effective_grain is not None and not set(effective_grain) <= set(query_dim_names):
bad = sorted(set(effective_grain) - set(query_dim_names))
ctx.errors.append(
SemanticError(
code="GRAIN_NOT_SUBSET",
message=(
f"Measure '{name}' grain {bad} is not a subset of "
f"query dimensions {query_dim_names}. "
f"This would cause row multiplication."
),
path="select.measures",
hint=f"Add {', '.join(repr(d) for d in bad)} to select.dimensions.",
context={"measure": name, "missingDimensions": bad},
)
)
return ResolvedMeasure(
name=name,
aggregation=measure.aggregation,
expression=expr,
is_expression=measure.expression is not None,
total=measure.total,
default_value=measure.default_value,
grain_override=grain_override,
effective_grain=effective_grain,
filter_context=measure.filter_context,
)
def _build_measure_expr(self, ctx: _ResolutionContext, measure: Measure) -> Expr:
"""Build the aggregate expression for a measure."""
# Engine-delegated aggregation (Databricks Metric View). Emit
# ``MEASURE("<label>")`` literally — there's no source column
# to read; the engine resolves the aggregation by name. Dialect
# support is enforced downstream by ``_check_aggregation_supported``.
if measure.aggregation == AggregationType.MEASURE:
return FunctionCall(
name="MEASURE",
args=[ColumnRef(name=measure.name, table=None)],
)
if measure.expression:
return self._expand_expression(ctx, measure)
# Build column references for all columns. Routes through
# ``make_column_expr`` so a measure column that points at a
# computed (``expression:``) column inlines the template body
# — without this, ``count_distinct`` over an ``expression:``
# column would emit ``COUNT(DISTINCT "obj"."")`` (zero-length
# identifier, DB error).
# Whether a cast is coming, and whether it is a numeric one. A boolean
# source only has to become a number when it is about to be read as
# one; ``None`` here means the measure passes its value through.
numeric_output = _reads_a_number(measure, ctx.model)
args: list[Expr] = []
if measure.columns:
for ref in measure.columns:
obj_name = ref.view or ""
col_name = ref.column or ""
# A column-less ref (``dataObject`` set, ``column`` empty) anchors the
# measure on the object without naming a column — used by the
# synthesized row-count measure to emit ``COUNT(*)`` while still
# contributing the anchor to source-object resolution.
if not col_name:
continue
obj = ctx.model.data_objects.get(obj_name)
if obj and col_name in obj.columns:
col_expr = make_column_expr(ctx.model, obj_name, col_name)
if numeric_output:
col_expr = _flag_as_number(col_expr)
args.append(col_expr)
else:
args.append(ColumnRef(name=col_name, table=obj_name))
if not args:
args = [Literal.number(1)]
agg = measure.aggregation.upper()
distinct = measure.distinct
if agg == "COUNT_DISTINCT":
agg = "COUNT"
distinct = True
# LISTAGG: attach separator and optional ordering; FIRST / LAST: the key
separator: str | None = None
if agg == "LISTAGG":
separator = measure.delimiter if measure.delimiter is not None else ","
order_by = self._within_group_order(ctx, measure, agg)
result = FunctionCall(
name=agg,
args=args,
distinct=distinct,
order_by=order_by,
separator=separator,
)
return self._apply_measure_default(
measure, _with_percentile(measure, self._apply_measure_filters(ctx, measure, result))
)
def _expand_expression(self, ctx: _ResolutionContext, measure: Measure) -> Expr:
"""Expand a measure expression with ``{[DataObject].[Column]}`` refs into AST."""
formula = measure.expression or ""
agg = measure.aggregation.upper()
tokens = tokenize_measure_expression(formula, ctx.model)
# The tokenizer resolves {[Object].[Column]} straight to a physical
# ref, so the query zone has to be applied here as it is for a column
# a dimension names: otherwise one column means two instants depending
# on how the query reached it.
inner = apply_query_timezone(parse_expression(tokens), ctx.model)
# The same rule the ``columns:`` form gets: two spellings of one
# measure, so a boolean reaches a numeric output as a number either
# way. Scoping it to the other branch left this one failing.
if _reads_a_number(measure, ctx.model):
inner = _flag_as_number(inner)
distinct = measure.distinct
if agg == "COUNT_DISTINCT":
agg = "COUNT"
distinct = True
result = FunctionCall(
name=agg,
args=[inner],
distinct=distinct,
order_by=(
self._within_group_order(ctx, measure, agg) if agg in _ORDERED_VALUE_CALLS else []
),
)
return self._apply_measure_default(
measure, _with_percentile(measure, self._apply_measure_filters(ctx, measure, result))
)
def _include_late_window_inputs(self, ctx: _ResolutionContext) -> None:
from orionbelt.compiler.late_window import is_late, late_inputs
components = ctx.result.metric_components
existing = {m.name for m in ctx.result.measures}
for measure in list(ctx.result.measures):
if not is_late(measure, components):
continue
for name in late_inputs(measure, components):
if name in existing:
continue
included = components.get(name) or self._resolve_measure(ctx, name)
if included is None:
continue
ctx.result.measures.append(included)
ctx.result.late_window_inputs.add(name)
existing.add(name)
if name in ctx.model.effective_measures:
source_objs = self._get_measure_source_objects(ctx, name)
ctx.result.measure_source_objects.update(source_objs)
ctx.result.required_objects.update(source_objs)
ctx.result.required_objects.update(self._get_measure_join_objects(ctx, name))
@staticmethod
def _within_group_order(
ctx: _ResolutionContext, measure: Measure, agg: str
) -> list[OrderByItem]:
"""The ``withinGroup`` ordering of a LISTAGG, or the key of a FIRST / LAST.
LISTAGG orders as declared; LAST is the value at the greatest key, so it
orders descending, FIRST ascending.
"""
wg = measure.within_group
if wg is None or (agg != "LISTAGG" and agg not in _ORDERED_VALUE_CALLS):
return []
wg_obj_name = wg.column.view or ""
wg_col_name = wg.column.column or ""
wg_obj = ctx.model.data_objects.get(wg_obj_name)
if wg_obj and wg_col_name in wg_obj.columns:
# A FIRST / LAST key in the column's own time, not the query's: it
# only orders, and instants order truly where wall-clock times
# repeat (the hour a zone falls back). A LISTAGG's ordering stays in
# the query's time, as its value is: a self-ordered one is
# recognised by the two being the same expression.
wg_expr: Expr = make_column_expr(
ctx.model,
wg_obj_name,
wg_col_name,
in_query_timezone=agg not in _ORDERED_VALUE_CALLS,
)
else:
wg_expr = ColumnRef(name=wg_col_name, table=wg_obj_name)
desc = agg == "LAST" if agg in _ORDERED_VALUE_CALLS else wg.order.upper() == "DESC"
return [OrderByItem(expr=wg_expr, desc=desc)]
@staticmethod
def _apply_measure_default(measure: Measure, expr: Expr) -> Expr:
"""Wrap an aggregate in its declared empty-set value.
Outside the aggregate rather than inside: ``COALESCE(SUM(x), 0)``
answers 0 when the aggregate saw nothing, where ``SUM(COALESCE(x, 0))``
would answer 0 for a row whose value is missing — a different claim.
Emitted for every dialect, which is the point: an aggregate over an
empty row set is NULL in standard SQL and 0 on ClickHouse, so a model
that says what it wants no longer depends on which engine runs it.
"""
if measure.default_value is None:
return expr
return FunctionCall(name="COALESCE", args=[expr, Literal(value=measure.default_value)])
@staticmethod
def _apply_measure_filters(
ctx: _ResolutionContext, measure: Measure, func: FunctionCall
) -> FunctionCall:
"""Wrap aggregate args with CASE WHEN if the measure has filters.
A filter on a fact the measure's anchor conforms is not wrapped: it is
applied inside that fact's conformed subquery (``conformed_filters``).
"""
if not measure.filters:
return func
pushed, kept = split_conformed_filters(ctx.model, measure, ctx.result.use_path_names)
for fact, items in pushed.items():
pushed_condition = build_measure_filter_condition(items, ctx.model, ctx.errors)
bucket = ctx.result.conformed_measure_filters.setdefault((measure.name, fact), [])
if pushed_condition is not None and pushed_condition not in bucket:
bucket.append(pushed_condition)
if not kept:
return func
condition = build_measure_filter_condition(kept, ctx.model, ctx.errors)
if condition is None:
return func
wrapped_args: list[Expr] = [CaseExpr(when_clauses=[(condition, arg)]) for arg in func.args]
return replace(func, args=wrapped_args)
def _resolve_metric(
self, ctx: _ResolutionContext, name: str, metric: Metric
) -> ResolvedMeasure | None:
"""Resolve a metric to its combined expression."""
return metric_resolution.resolve_metric(self, ctx, name, metric)
def _validate_partition_dimensions(
self,
ctx: _ResolutionContext,
metric_name: str,
partition_by: list[str],
path_template: str,
) -> bool:
return metric_resolution.validate_partition_dimensions(
self, ctx, metric_name, partition_by, path_template
)
def _resolve_window_metric(
self, ctx: _ResolutionContext, name: str, metric: Metric
) -> ResolvedMeasure | None:
"""Resolve a window metric (rank/lag/lead/ntile/first_value/last_value)."""
return metric_resolution.resolve_window_metric(self, ctx, name, metric)
def _resolve_derived_metric(
self, ctx: _ResolutionContext, name: str, metric: Metric
) -> ResolvedMeasure | None:
"""Resolve a derived metric to its combined expression."""
return metric_resolution.resolve_derived_metric(self, ctx, name, metric)
def _resolve_cumulative_metric(
self, ctx: _ResolutionContext, name: str, metric: Metric
) -> ResolvedMeasure | None:
"""Resolve a cumulative metric referencing an existing measure."""
return metric_resolution.resolve_cumulative_metric(self, ctx, name, metric)
def _resolve_pop_metric(
self, ctx: _ResolutionContext, name: str, metric: Metric
) -> ResolvedMeasure | None:
"""Resolve a period-over-period metric."""
return metric_resolution.resolve_pop_metric(self, ctx, name, metric)
def _order_only(self, ctx: _ResolutionContext, name: str, selected: set[str]) -> bool:
"""Whether the ORDER BY field *name* is a measure or metric not selected.
"The top three customers by revenue", showing the names: the rows are
the query's own, so the measure is not added to its plan, where its
fact would choose the base and drop the customers without revenue. The
``order_only`` pass computes it in a query of its own over the same
dimensions and filters, joins it onto the finished rows and sorts them.
A dimension is not taken: the query's grain holds no one value of it.
"""
model = ctx.model
if ctx.result.is_raw or name in selected or name in ctx.result.coalesce_aliases:
return False
if name not in model.effective_measures and name not in model.metrics:
return False
if ctx.result.grouping is not None:
ctx.errors.append(
SemanticError(
code="ORDER_BY_COMBINATION_NOT_SUPPORTED",
message=(
f"ORDER BY '{name}', which the query does not select, is not "
f"supported beside {ctx.result.grouping.value.upper()}: its subtotal "
"rows have no value of it to sort by"
),
path="orderBy",
hint=f"Select '{name}' to sort by it.",
)
)
return True
ctx.result.order_only_measures.add(name)
# Read by the sort query, so the result cache depends on its tables.
ctx.result.subquery_objects.update(self._get_measure_source_objects(ctx, name))
ctx.result.subquery_objects.update(self._get_measure_join_objects(ctx, name))
return True
def _collect_having_measure_refs(self, query: QueryObject, model: SemanticModel) -> list[str]:
"""Collect measure/metric names referenced in any HAVING filter.
Walks ``query.having`` recursively (each entry is a
``QueryFilter`` or a ``QueryFilterGroup``) and returns the
ordered, de-duplicated list of ``field`` values that name a
known measure or metric in the model. Order is preserved for
deterministic resolution; duplicates are dropped on first sight.
"""
seen: set[str] = set()
out: list[str] = []
measure_names = model.effective_measures
def _visit(item: QueryFilterItem) -> None:
if isinstance(item, QueryFilterGroup):
for child in item.filters:
_visit(child)
return
field = item.field
if field in seen:
return
if field in measure_names or field in model.metrics:
seen.add(field)
out.append(field)
for entry in query.having:
_visit(entry)
return out
def _get_measure_join_objects(self, ctx: _ResolutionContext, name: str) -> set[str]:
"""Objects a measure needs *joined* without being sourced from them.
A ``withinGroup`` column becomes the aggregate's ``ORDER BY``, so it has
to resolve — but it contributes no value to the measure. Kept out of
``measure_source_objects`` for that reason: that set drives CFL
detection and the explain output's fact-table list, and a LISTAGG's sort
column is neither a fact nor a source.
Without this the object is never joined and the compiler emits SQL that
binds to nothing:
SELECT LISTAGG("Products"."id", ',' ORDER BY "Sales"."quantity")
FROM "products" AS "Products" -- Sales never joined
which every engine rejects at execution time.
A column the measure reads may itself be computed from a column of
another object, which lands here for the same reason and with the same
care about ``measure_source_objects``: the object supplies part of an
expression evaluated per fact row, not a second fact to union.
"""
result: set[str] = set()
if name in ctx.model.effective_measures:
return ctx.model.measure_join_objects(name)
metric = ctx.model.metrics.get(name)
if metric is not None:
if metric.measure:
result.update(self._get_measure_join_objects(ctx, metric.measure))
if metric.expression:
for ref_name in re.findall(r"\{\[([^\]]+)\]\}", metric.expression):
result.update(self._get_measure_join_objects(ctx, ref_name))
return result
def _push_filters_into_conformed_facts(self, ctx: _ResolutionContext) -> None:
"""Apply a WHERE on a conformed fact inside its conformed subqueries.
A predicate reading only what one conformed fact reaches
(``conformed_reach``) chooses that fact's rows: it is moved from the
WHERE to ``conformed_filters`` for every anchored measure conforming
the fact, and applied before the rows are aggregated to the shared key.
In the WHERE it would name a table the plan reads only as a per-key
total.
Every measure the query computes has to apply it: an anchored measure
conforming the fact, or, in a union of facts, one whose own fact
reaches it (the filter then stays in the WHERE for that leg). A
measure that can do neither is computed without it, and the query
names it (``FILTER_NOT_APPLIED``; a static filter
``STATIC_FILTER_NOT_APPLIED``, or refused under ``enforce: always``). A
predicate that
also reads what the anchor reaches - an OR across both sides - has no
single place to be applied and is refused (``FILTER_ON_CONFORMED_OBJECT``).
"""
reach = ctx.conformed_reach
pushable = {name for facts in reach.values() for objs in facts.values() for name in objs}
if not pushable:
return
computed = self._computed_measure_names(ctx)
graph = JoinGraph.of(ctx.model, use_path_names=ctx.result.use_path_names or None)
kept: list[ResolvedFilter] = []
for wf in ctx.result.where_filters:
tables = filter_tables(wf.expression)
if not tables & pushable:
kept.append(wf)
continue
facts: dict[str, list[str]] = {}
missing: list[str] = []
for name in computed:
# Every conformed fact reaching it: a dimension both facts join
# restricts both, whatever they are called.
fitting = [f for f, objs in reach.get(name, {}).items() if tables <= objs]
if fitting:
facts[name] = fitting
continue
measure = ctx.model.effective_measures.get(name)
sources = measure.source_objects if measure is not None else set()
reaches = ctx.result.requires_cfl and any(
tables <= graph.descendants_without_unnest(src) | {src} for src in sources
)
if not reaches:
missing.append(name)
read = ", ".join(sorted(tables & pushable))
if not facts:
if not wf.static or wf.enforced:
self._refuse_filter_on_conformed_object(ctx, tables & pushable)
else:
filter_resolution.report_unapplied_static_filter(
[], ctx.result.warnings, None, read
)
continue
if missing:
names = ", ".join(f"'{name}'" for name in missing)
if wf.enforced:
ctx.errors.append(
static_filter_unreachable_error(
f"static filter on '{read}'", read, f"the fact behind {names}"
)
)
continue
# Kept for a union leg that reads it natively, the filter
# reaches the planner, whose legs name what they skip.
native = len(facts) + len(missing) < len(computed)
if not native and wf.static:
filter_resolution.report_unapplied_static_filter(
[], ctx.result.warnings, None, read, names
)
elif not native:
filter_resolution.report_unapplied_query_filter(
ctx.result.warnings, read, names
)
for name, fitting in facts.items():
for fact in fitting:
bucket = ctx.result.conformed_filters.setdefault((name, fact), [])
if wf not in bucket:
bucket.append(wf)
for fitting in facts.values():
for fact in fitting:
ctx.result.subquery_objects.update(
conformed_join_objects(ctx.model, fact, tables, ctx.result.use_path_names)
)
if len(facts) + len(missing) < len(computed):
# Measures on a union leg that reaches the filter natively.
kept.append(replace(wf, conformed_for=frozenset(facts)))
ctx.result.where_filters = kept
@staticmethod
def _computed_measure_names(ctx: _ResolutionContext) -> list[str]:
"""The model measures the query aggregates, a metric's placeholder
(reaggregate, cumulative, window, nested) read as the measures under it."""
names = {
m.name
for m in (*ctx.result.measures, *ctx.result.metric_components.values())
if not m.component_measures
}
return sorted(computed_measures(ctx.model, names))
@staticmethod
def _refuse_filter_on_conformed_object(ctx: _ResolutionContext, objects: set[str]) -> None:
listed = ", ".join(f"'{name}'" for name in sorted(objects))
ctx.errors.append(
SemanticError(
code="FILTER_ON_CONFORMED_OBJECT",
message=(
f"A filter reads {listed}, which an anchored measure reaches only "
f"by aggregating it to a shared key, together with data objects "
f"outside it. Such a filter is applied before the aggregation, "
f"inside the conformed subquery, so it can read only what that "
f"subquery reads. This covers the query's own filters and the "
f"model's static ones alike."
),
path="where",
hint=(f"Split the condition: one on {listed} alone, and one on the anchor's side."),
)
)
@staticmethod
def _check_fill(ctx: _ResolutionContext, query: QueryObject) -> None:
"""Refuse ``fill`` where it has no single axis, or a feature reads the
rows before the empty periods are added.
A window or period-over-period metric and a total are computed over
the rows the query returns, so an empty period added afterwards would
show none of their values; ROLLUP / CUBE subtotals and a HAVING would
have to be evaluated again over it. A cumulative metric is evaluated
for an empty period by the fill itself, from the periods its window
reads, unless it is not computed per period of the filled dimension:
one over another time dimension is evaluated as of a single period,
and one over a reaggregate metric by a later pass.
"""
result = ctx.result
grains = [(d.name, d.grain) for d in result.dimensions if d.grain is not None]
declared = [ctx.model.dimensions.get(d.dimension) for d in result.dimensions if d.grain]
coalesced = any(d.grain and d.coalesce_alias for d in result.dimensions)
if (
coalesced
or len(grains) != 1
or any(
dim is None or dim.result_type not in (DataType.DATE, DataType.TIMESTAMP)
for dim in declared
)
):
ctx.errors.append(
SemanticError(
code="FILL_REQUIRES_ONE_TIME_DIMENSION",
message=(
f"fill returns every period of the query's time dimension, so the "
f"query has to select exactly one dimension at a time grain, with "
f"resultType date or timestamp and outside a coalesce group; it "
f"selects {len(grains)} at a grain."
),
path="fill",
hint="Select one time dimension at a grain, such as 'Order Date:month'.",
)
)
return
name, grain = grains[0]
if grain in SUB_DAY_GRAINS:
ctx.errors.append(
SemanticError(
code="FILL_GRAIN_NOT_SUPPORTED",
message=(
f"fill generates periods of a day or longer; '{name}' is at "
f"{grain.value} grain."
),
path="fill",
)
)
combined = [
label
for label, present in (
("ROLLUP / CUBE", result.grouping is not None),
("having", bool(query.having)),
("a total", result.has_totals),
(
"a cumulative metric over a reaggregate metric",
any(m.is_cumulative and m.cumulative_over_reaggregate for m in result.measures),
),
(
f"a cumulative metric over a time dimension other than '{name}'",
any(
m.is_cumulative and m.cumulative_time_dimension != name
for m in result.measures
),
),
("a period-over-period metric", result.has_pop),
("a window metric", result.has_window),
)
if present
]
if combined:
ctx.errors.append(
SemanticError(
code="FILL_COMBINATION_NOT_SUPPORTED",
message=(
f"fill adds the empty periods after the query's rows are computed, "
f"which {', '.join(combined)} would have to read: it is not supported "
f"together with them."
),
path="fill",
)
)
@staticmethod
def _warn_repeated_conformed_terms(ctx: _ResolutionContext) -> None:
"""Warn where a summed term reads only a fact conformed to a peer anchor.
Anchored on a fact, a conformed value is its key's total, read once per
anchor row. In a product with an anchor column that is right:
``SUM(r * S)`` is ``R * S`` per key. A term without one is added once
per anchor row: ``SUM(r - S)`` is ``R - n * S``, and a key without
anchor rows loses it. Anchored on the shared key itself there is one
row per key, so nothing repeats.
"""
overrides = path_overrides(ctx.result.use_path_names)
resolved = {
m.name: m for m in (*ctx.result.measures, *ctx.result.metric_components.values())
}
for name, anchor in ctx.result.anchored_measures.items():
measure = ctx.model.effective_measures.get(name)
resolved_measure = resolved.get(name)
if measure is None or resolved_measure is None or measure.aggregation != "sum":
continue
repeated = [
fact
for fact in ctx.conformed_reach.get(name, {})
if all(j.join_to != anchor for j in ctx.model.effective_joins(fact, overrides))
]
conformed = {obj for fact in repeated for obj in ctx.conformed_reach[name][fact]}
# The SUM may be wrapped: ``defaultValue`` renders COALESCE(SUM(...), 0).
summed: list[Expr] = []
def collect_sums(node: Expr, summed: list[Expr] = summed) -> Expr | None:
if isinstance(node, FunctionCall) and node.name == "SUM":
summed.extend(node.args)
return node
return None
map_nodes(resolved_measure.expression, collect_sums)
if not conformed or not any(
_has_term_without_anchor(arg, conformed)[0] for arg in summed
):
continue
facts = ", ".join(f"'{fact}'" for fact in repeated)
ctx.result.warnings.append(
warning(
code=WarningCode.FAN_TRAP_RISK,
message=(
f"Measure '{name}' sums a term that reads only {facts}, conformed "
f"to the key it shares with its anchor '{anchor}'. That term is "
f"the key's total and is added once per '{anchor}' row: counted "
f"several times for a key with several, and not at all for a key "
f"with none."
),
hint=(
f"Remove anchor: to conform both facts to the data object they "
f"share, or compute the term in a measure of its own on {facts} "
f"and combine the two with a metric."
),
context={"measure": name, "anchor": anchor, "conformed": repeated},
)
)
def _record_anchor(
self,
ctx: _ResolutionContext,
name: str,
measure: Measure,
result: set[str],
) -> None:
"""Settle the grain a cross-fact measure is evaluated at, or refuse.
A declared ``anchor:`` settles it. Otherwise the facts must share
exactly one directly-joined object: several are several different
answers, and the measure has to say which it means.
"""
conformed = measure.source_objects
anchor = effective_anchor(ctx.model, measure, ctx.result.use_path_names)
if anchor is None:
candidates = conform_key_candidates(ctx.model, measure, ctx.result.use_path_names)
if len(candidates) != 1:
ctx.errors.append(
SemanticError(
code="ANCHOR_REQUIRED_AMBIGUOUS_KEY",
message=(
f"Measure '{name}' reads {', '.join(sorted(conformed))}, which no "
f"join path reaches together, so each has to be aggregated to a key "
f"they share. "
+ (
f"They share {', '.join(candidates)}, and conforming at each "
f"gives a different answer."
if candidates
else "They share no directly joined data object."
)
),
path=f"measures.{name}",
hint=(
"Set anchor: on the measure to the data object whose grain the "
"expression should be evaluated at - one of the facts it reads, or "
"a data object all of them join to."
if candidates
else "Join the data objects to a common one, or read them in "
"separate measures and combine those with a metric."
),
)
)
return
anchor = candidates[0]
ctx.result.warnings.append(
warning(
code=WarningCode.CONFORMED_GRAIN_ASSUMED,
message=(
f"Measure '{name}' reads {', '.join(sorted(conformed))}, which no join "
f"path reaches together. Each was aggregated to '{anchor}', the only "
f"data object they both join to, and the expression evaluated once per "
f"'{anchor}' row."
),
hint=(
"Set anchor: to evaluate per row of one of the facts instead, which "
"changes AVG / MIN / MAX, and a SUM with a term reading one fact alone."
),
context={"measure": name, "conformedAt": anchor},
)
)
ctx.result.anchored_measures[name] = anchor
# Only the objects actually conformed leave the join requirements. An
# object the anchor *can* reach is read directly, by an ordinary join,
# so dropping it here left the expression naming a table the plan no
# longer joined - visible only when nothing else in the query happened
# to require it.
result -= anchored_conformed_objects(ctx.model, measure, ctx.result.use_path_names)
# So are the objects its pushed filters read: they are joined inside
# the conformed subquery.
# The conformed facts, and what the pushed filters join, are read in
# subqueries: still tables the result depends on.
conformed = anchored_conformed_objects(ctx.model, measure, ctx.result.use_path_names)
ctx.result.subquery_objects.update(conformed)
pushed, _ = split_conformed_filters(ctx.model, measure, ctx.result.use_path_names)
reach = conformed_reach(ctx.model, measure, ctx.result.use_path_names)
for fact, items in pushed.items():
result -= reach[fact]
read: set[str] = set()
for item in items:
collect_measure_filter_objects(item, read)
ctx.result.subquery_objects.update(
conformed_join_objects(ctx.model, fact, read, ctx.result.use_path_names)
)
result.add(anchor)
def _get_measure_source_objects(self, ctx: _ResolutionContext, name: str) -> set[str]:
"""Extract all source data objects for a measure or metric."""
result: set[str] = set()
measure = ctx.model.effective_measures.get(name)
if measure:
for cref in measure.columns:
if cref.view:
result.add(cref.view)
if measure.expression:
for obj_name, _col_name in find_qualified_refs(measure.expression):
result.add(obj_name)
for fi in measure.filters:
collect_measure_filter_objects(fi, result)
# An anchored measure's independent facts are conformed into
# subqueries by the planner, not joined. Reporting them here would
# add them to the join requirements and, worse, make the multi-fact
# check below flip the query into a CFL plan - whose UNION ALL puts
# the two facts' columns on different rows, which is the exact
# arrangement the anchor exists to avoid.
# Gated on "needs a grain" rather than "has one": a measure that
# needs one and cannot be given one has to be refused, and treating
# that as "needs nothing" is what let it through to a CFL leg that
# projected no such column.
if needs_conforming(ctx.model, measure, ctx.result.use_path_names):
self._record_anchor(ctx, name, measure, result)
return result
metric = ctx.model.metrics.get(name)
if metric:
if (
metric.type in (MetricType.CUMULATIVE, MetricType.WINDOW, MetricType.REAGGREGATE)
and metric.measure
):
# Measure-based metric: source objects come from the referenced measure
result.update(self._get_measure_source_objects(ctx, metric.measure))
elif metric.expression:
# Derived or PoP metric: parse expression for measure references
measure_refs = re.findall(r"\{\[([^\]]+)\]\}", metric.expression)
for ref_name in measure_refs:
result.update(self._get_measure_source_objects(ctx, ref_name))
return result
# -- base object selection -----------------------------------------------
@staticmethod
def _collect_where_filter_objects(query: QueryObject, model: SemanticModel) -> set[str]:
"""Data objects the query's WHERE clause references.
Walks ``query.where`` recursively; each ``field`` is either a dimension
name or a qualified ``DataObject.Column``. Measure names are skipped —
those are HAVING predicates, evaluated after aggregation rather than
joined into the FROM.
``EXISTS`` / ``NONEXISTS`` targets are skipped too: they compile to a
correlated subquery, not a join, so they place no reachability demand on
the base object.
A predicate on a computed column demands whatever objects its expression
reads as well — the predicate is only as reachable as the columns it
compares.
"""
found: set[str] = set()
measure_names = model.effective_measures
def visit(item: QueryFilterItem) -> None:
if isinstance(item, QueryFilterGroup):
for child in item.filters:
visit(child)
return
field = item.field
if not field or field in measure_names or field in model.metrics:
return
dimension = model.dimensions.get(field)
if dimension is not None:
if dimension.view:
found.add(dimension.view)
found.update(model.column_reference_objects(dimension.view, dimension.column))
return
if "." in field:
object_name, _, column_name = field.partition(".")
object_name, column_name = object_name.strip(), column_name.strip()
if object_name in model.data_objects:
found.add(object_name)
found.update(model.column_reference_objects(object_name, column_name))
for entry in query.where:
visit(entry)
return found
@staticmethod
def _reanchor_if_unreachable(
ctx: _ResolutionContext, best: str, filter_objects: set[str]
) -> str:
"""Re-anchor the base when the chosen measure source cannot reach the query.
Anchoring on a measure's own source object is right whenever that object
is the fact table. It is wrong when the measure lives on a *dimension*
table: joins are declared many-to-one and traversed forward-only, so a
base of ``Customers`` can reach nothing, and a query grouping
``Avg Customer Age`` by ``Category`` fails with
``UNREACHABLE_REQUIRED_OBJECT`` even though ``Sales`` joins to both.
Such a query is not multi-fact — it is single-fact viewed from the wrong
end. Re-anchoring on the common root that reaches every required object
makes it plan as an ordinary star; the measure then sits on the replicated
side of a forward join, where ``compiler.grain_dedup`` aggregates it over
deduplicated rows.
Deliberately narrow, so this can only turn an error into a result and
never re-plan a query that already works:
* With **one** measure source object, with several the chosen base
reaches (Customers and the Regions it joins: one fact seen from
its dimension end, as much as a single source is), or with several
when the query's WHERE names an object the chosen base cannot reach.
Otherwise multi-fact queries keep their existing base so CFL
detection (which runs on it straight after) is untouched. With several sources and an
unreachable filter object the existing base had no right answer: the
filter was dropped, and a count of customers ``WHERE status = 'ok'``
counted every customer. Rooted where the filter's rows are, it counts
the customers that have such an order, as the single-source case
already does.
* Only when that base genuinely cannot reach the rest — the case that
errors today.
* Only when a common root actually exists; otherwise the original base
is returned and the existing error still fires.
*filter_objects* are the data objects the query's WHERE clause names.
They are not in ``required_objects`` yet — filters resolve much later —
but the base still has to reach them: a WHERE on an unreachable object
is silently dropped downstream, which would answer a different question
than the one asked. Static model filters are deliberately excluded; those
are declared "apply where relevant", so they must not drag the base
towards a table the query never mentioned.
"""
graph = JoinGraph.of(ctx.model, use_path_names=ctx.result.use_path_names or None)
sources = ctx.result.measure_source_objects
reached = graph.descendants(best)
# Measure sources the base reaches without multiplying its rows are one
# fact, as a single source is, unless the star on the root they are
# re-anchored on would lose their rows without children.
one_fact = sources - {best} <= graph.descendants_without_fanout(
best
) and QueryResolver._keeps_childless_parents(ctx, None)
if not one_fact and filter_objects - {best} <= reached:
return best
wanted = ctx.result.required_objects | sources | filter_objects
remaining = wanted - {best}
if not remaining:
return best
if remaining <= reached:
return best
root = graph.find_common_root(wanted | {best})
if not root:
return best
# ``find_common_root`` falls back to a Steiner centre when no single
# ancestor covers everything, so its answer is a best effort rather than
# a guarantee. Re-anchor only on a root that genuinely reaches the whole
# query; otherwise keep the original base and let the existing
# unreachable error or filter skip stand.
if wanted <= graph.descendants(root) | {root}:
return root
return best
@staticmethod
def _keeps_childless_parents(ctx: _ResolutionContext, base: str | None) -> bool:
"""Whether a star on *base* keeps the other measure sources' rows
without children: under ROLLUP / CUBE, for one such source only (see
:meth:`_childless_obstacle`). *base* ``None`` is an object that is no
measure source."""
others = ctx.result.measure_source_objects - {base}
return ctx.result.grouping is None or len(others) <= 1
@staticmethod
def _parent_of(ctx: _ResolutionContext, name: str, reached: set[str]) -> str | None:
"""The object a measure's values come from, when that is one object the
base reaches: a measure read through the base's rows. Not a measure
that reads the base itself (``Orders.Amount * Customers.Credit`` is
evaluated per order row), and not the objects only its filter reads."""
measure = ctx.model.effective_measures.get(name)
if measure is None:
return None
sources = measure.source_objects
if len(sources) != 1:
return None
source = next(iter(sources))
return source if source in reached else None
@staticmethod
def _reads_beyond(ctx: _ResolutionContext, name: str) -> bool:
"""Whether a measure's filter reads an object besides its source."""
measure = ctx.model.effective_measures.get(name)
if measure is None or not measure.filters:
return False
objects: set[str] = set()
for item in measure.filters:
collect_measure_filter_objects(item, objects)
return bool(objects - measure.source_objects)
def _plan_parents_without_children(
self, ctx: _ResolutionContext, required: frozenset[str] = frozenset()
) -> None:
"""Add the parents without child rows when no WHERE asks for them.
A query based on the orders reads the customers through their orders,
so a customer measure beside an order measure lost every customer
without one: ``Customers Count`` was 3 alone and 2 beside ``Revenue``.
A measure's value must not depend on what else is selected, so those
customers are added as a leg of their own, as for a WHERE that admits
them; grouped by an order column, they are its NULL group.
A measure filter on the orders is evaluated for them with every order
column NULL: ``order_id notset`` holds, ``order_id set`` does not. A
shape the leg cannot express (measures on several such objects,
``ROLLUP`` / ``CUBE``, period-over-period, a nested object) keeps the
plan as it was, unless such a filter can hold for them
(:mod:`compiler.null_rejection`): that measure would then be answered
without the only rows it accepts, and is refused
(``FILTER_ADMITS_ROWS_WITHOUT``) instead.
"""
base = ctx.result.base_object
graph = ctx.graph
if graph is None or (ctx.result.allow_fan_out and not required):
return
reached = graph.descendants(base)
names = [m.name for m in ctx.result.measures] + list(ctx.result.metric_components)
admitting: dict[str, str] = {}
for name in names:
measure = ctx.model.effective_measures.get(name)
if measure is None or not measure.filters or measure.allow_fan_out:
continue
parent = self._parent_of(ctx, name, reached)
if parent is None:
continue
filter_objects: set[str] = set()
for item in measure.filters:
collect_measure_filter_objects(item, filter_objects)
children = filter_objects - graph.descendants(parent) - {parent}
if not children:
continue
gate = build_measure_filter_condition(measure.filters, ctx.model, [])
if gate is not None and can_hold_without(gate, children):
admitting[name] = parent
parents = sorted(
{parent for name in names if (parent := self._parent_of(ctx, name, reached))} | required
)
joined = {base} | {step.to_object for step in ctx.result.join_steps}
legs: list[ChildlessLeg] = []
for parent in parents:
needed = sorted(name for name, owner in admitting.items() if owner == parent)
# A WHERE on what the query joins outside the parent (the sale
# year) holds for none of these rows, which read it NULL: nothing
# to add.
outside = joined - joined_beneath(ctx.result.join_steps, parent)
if any(not can_hold_without(wf.expression, outside) for wf in ctx.result.where_filters):
continue
reason = self._childless_obstacle(
ctx, parent, parents, names, reached, bool(needed) or parent in required
)
if reason is None:
legs.append(
ChildlessLeg(
parent=parent,
child=base,
children=frozenset(
{base} | (reached - graph.descendants(parent) - {parent})
),
)
)
elif parent in required:
ctx.errors.append(
SemanticError(
code="FILTER_ADMITS_ROWS_WITHOUT",
message=(
f"The WHERE can hold for a row of '{parent}' with no '{base}' "
f"rows (an OR with another condition, a NOT, or an IS NULL), and "
f"such rows cannot be added to this query: {reason}. Answered "
f"from the '{base}' rows alone, they would be left out without "
f"a word."
),
path="where",
hint=(
"Write the condition as 'exists' / 'nonexists', which tests for "
"a matching row and keeps the ones without any."
),
)
)
elif needed:
measures = ", ".join(f"'{name}'" for name in needed)
ctx.errors.append(
SemanticError(
code="FILTER_ADMITS_ROWS_WITHOUT",
message=(
f"The filter of {measures} can hold for a row of '{parent}' with "
f"no '{base}' rows (an IS NULL, a NOT, or an OR with another "
f"condition), and such rows cannot be added to this query: "
f"{reason}. Answered from the '{base}' rows alone, they would be "
f"left out without a word."
),
)
)
# Otherwise the plan stays as it was for this parent.
ctx.result.childless_legs = legs
def _childless_obstacle(
self,
ctx: _ResolutionContext,
parent: str,
parents: list[str],
names: list[str],
reached: set[str],
needed: bool,
) -> str | None:
"""Why *parent*'s rows without children cannot be added, if they cannot."""
base = ctx.result.base_object
graph = ctx.graph
assert graph is not None
if len(parents) > 1 and ctx.result.grouping is not None:
return "measures on several such objects under ROLLUP / CUBE"
if any(
((obj := ctx.model.data_objects.get(name)) is not None and obj.is_nested)
or ctx.model.unnest_root(name) != name
for name in (parent, base)
) or any(
# Reached through an array: no table to correlate a NOT EXISTS on.
ctx.model.unnest_root(name) != name and parent in graph.descendants(name)
for name in reached
):
return "a nested data object is involved"
if not needed and any(
(measure := ctx.model.effective_measures.get(name)) is not None
and measure.allow_fan_out
and parent in measure.source_objects
for name in names
):
# Its rows are read as the joins replicate them, on purpose; the
# leg is joined to deduplicated rows only.
return "a measure on it sets allowFanOut"
if (
not needed
and ctx.result.grouping is not None
and all(
self._reads_beyond(ctx, name)
for name in names
if self._parent_of(ctx, name, reached) == parent
)
):
# Under ROLLUP / CUBE the groups are read from the parent's
# unfiltered measures; a measure filtered on another object reads
# fewer rows and cannot list them.
return "every measure on it is filtered on another object"
return None
def _plan_rows_without_children(self, ctx: _ResolutionContext) -> None:
"""Add the parents without child rows a WHERE accepts, or refuse.
Measures on the *one* side of a join (customers) with a WHERE on the
*many* side (orders) are answered from the order rows, the customer
measures deduplicated. A customer counts only through an order the
predicate accepts, so one without orders is left out: right for
``status = 'ok'``, which no such customer satisfies, and wrong for
``tier = 'gold' OR status = 'ok'`` or ``order_id IS NULL``, which one
does.
The answer meant is ``customers LEFT JOIN orders WHERE <predicate>``,
each customer once. Its customers are the ones with an order the
predicate accepts - the plan as it stands - and the ones without orders
for which the predicate holds with every order column NULL. When that
second set can be non-empty (:mod:`compiler.null_rejection`), the plan
gets a leg for it (:attr:`ResolvedQuery.childless_legs`, one per such
object, built by
``grain_dedup``): the two sets do not overlap, and the new leg never
joins the orders, so neither multiplies the other.
A shape the leg cannot express is refused instead
(``FILTER_ADMITS_ROWS_WITHOUT``): a period-over-period metric, a nested
object on either side, or measures on several such objects under
``ROLLUP`` / ``CUBE``.
"""
if ctx.result.requires_cfl or ctx.graph is None:
return
# The objects the measures' values come from: one only a measure filter
# reads is no source of a row, and is handled with that filter.
sources: set[str] = set()
for name in [m.name for m in ctx.result.measures] + list(ctx.result.metric_components):
measure = ctx.model.effective_measures.get(name)
if measure is not None:
sources |= measure.source_objects
if not sources:
return
read: set[str] = set()
for wf in ctx.result.where_filters:
collect_referenced_tables(wf.expression, read)
unreached = {source: read - ctx.graph.descendants(source) - {source} for source in sources}
children = set().union(*unreached.values())
combined: Expr | None = None
for wf in ctx.result.where_filters:
combined = (
wf.expression
if combined is None
else BinaryOp(left=combined, op="AND", right=wf.expression)
)
if not children or combined is None:
# No WHERE asks for parents without children; a measure, or its
# filter, still may.
self._plan_parents_without_children(ctx)
return
base = ctx.result.base_object
graph = ctx.graph
where = combined
# One leg per object the measures read that cannot reach what the WHERE
# reads, for which the WHERE can hold with those columns NULL: each on
# its own, since ``Category = games OR Tier = gold`` holds for a
# customer without orders and for a product never ordered, but not
# with both sides NULL at once.
parents = sorted(
source
for source, missing in unreached.items()
if missing and can_hold_without(where, missing)
)
if not parents:
self._plan_parents_without_children(ctx)
return
reason: str | None = None
for parent in parents:
if len(parents) > 1 and ctx.result.grouping is not None:
reason = "measures on several such objects under ROLLUP / CUBE"
elif parent == base or parent not in graph.descendants(base):
reason = f"'{parent}' is not reached from '{base}'"
elif any(
(obj := ctx.model.data_objects.get(name)) is not None and obj.is_nested
for name in (parent, base)
):
reason = "a nested data object is involved"
if reason is not None:
break
if reason is None:
# These parents need their rows; every other one the measures read
# through the base keeps them too, as without a WHERE.
self._plan_parents_without_children(ctx, required=frozenset(parents))
return
objects = ", ".join(f"'{name}'" for name in sorted(children))
ctx.errors.append(
SemanticError(
code="FILTER_ADMITS_ROWS_WITHOUT",
message=(
f"The WHERE can hold for a row with no {objects} rows (an OR with "
f"another condition, a NOT, or an IS NULL), and such rows cannot be "
f"added to this query: {reason}. Answered from the {objects} rows "
f"alone, they would be left out without a word."
),
path="where",
hint=(
f"Write the condition on {objects} as 'exists' / 'nonexists', which "
f"tests for a matching row and keeps the ones without any."
),
)
)
def _reject_unsupported_nested_shapes(
self, ctx: _ResolutionContext, touched_objects: set[str]
) -> None:
"""Refuse the two nested shapes the planner cannot render yet.
**A nested object in a union leg.** Each CFL leg picks its own root and
builds its own FROM, and neither knows how to carry an unnest, so the
leg would select from a table the object does not have. Two nested
objects in one query are meant to *become* that union; this is what has
to land first.
**A nested object whose parent is itself nested.** An unnest names its
parent's array column, and where the parent is an element rather than a
row that reference is not a column at all: Snowflake's ``FLATTEN`` row
exposes the element under ``value``, so the array is ``p.value:"Parts"``
and ``p."Parts"`` does not compile, and MySQL's ``JSON_TABLE`` projects
only the scalar columns it was told to extract, so the array is not
there to read. Reaching it needs a per-dialect parent-element access
that is designed and measured on its own; until then the shape is
refused rather than emitted as SQL two engines reject and five have
never been asked.
Both are refused rather than compiled, because the alternative is SQL
naming a table that does not exist or - worse, and this is what the
filter case actually did - a plausible number from unfiltered rows.
"""
multi_fact = ctx.result.requires_cfl or ctx.result.dimensions_exclude
for name in sorted(touched_objects):
obj = ctx.model.data_objects.get(name)
source = obj.nested_in if obj is not None else None
if source is None:
continue
parent = ctx.model.data_objects.get(source.data_object)
if parent is not None and parent.is_nested:
ctx.errors.append(
SemanticError(
code="NESTED_WITHIN_NESTED_UNSUPPORTED",
message=(
f"Data object '{name}' is nested in '{source.data_object}', which "
f"is itself nested. An unnest names its parent's array column, and "
f"where the parent is an array element rather than a row, that "
f"reference is spelled differently on every engine - so this "
f"compiles to SQL some of them reject."
),
path=f"dataObjects.{name}.nestedIn",
hint=(
"Nest the object directly in the table's own object, or read the "
"inner array through a flattening view by declaring 'code' "
"alongside 'nestedIn'."
),
)
)
continue
if multi_fact:
ctx.errors.append(
SemanticError(
code="NESTED_OBJECT_IN_MULTI_FACT",
message=(
f"Data object '{name}' takes its rows by unnesting "
f"'{source.data_object}.{source.column}', and this "
f"query is planned as a union of independent facts, where each "
f"leg selects from a table of its own. A nested object has none."
),
path="select",
hint=(
"Query the nested object with measures from its own parent only, "
"or read it through a flattening view by declaring 'code' "
"alongside 'nestedIn'."
),
)
)
def _select_base_object(
self, ctx: _ResolutionContext, filter_objects: set[str] | None = None
) -> str:
"""Select the base (fact) object, which is never a nested one.
A ``nestedIn`` object has no table: its rows are an array column on its
parent, reached by an unnest that names the parent. Every route into
this function can nominate one anyway - it is a measure source like any
other, and the "prefer the object with the most joins" fallback does not
look at where rows come from - so the answer is mapped up to the nearest
ancestor a FROM clause can name, which reaches everything the nested
object does and more.
"""
return ctx.model.unnest_root(self._choose_base_object(ctx, filter_objects))
def _choose_base_object(
self, ctx: _ResolutionContext, filter_objects: set[str] | None = None
) -> str:
"""Prefer measure source objects with the most joins, then the widest reach."""
# An anchored measure has already been told which grain to run at, and
# the planner joins its conformed subqueries against that object. Moving
# the base elsewhere leaves those joins referencing a table no longer in
# the FROM. Where the anchor cannot reach the query's other objects the
# query is genuinely unanswerable at that grain, and the reachability
# check below says so - which is more use than a binder error.
anchors = set(ctx.result.anchored_measures.values())
if len(anchors) == 1:
return next(iter(anchors))
if ctx.result.measure_source_objects:
# Among the measure sources, prefer one that can actually reach the
# others. "Most joins" alone picks the busiest fact, which is not
# the same thing: a measure reading Sales and Returns, where the
# declared join runs Returns -> Sales, has to be based on Returns,
# because Sales reaches Returns only by traversing that join
# backwards. Basing it on Sales left the join path empty and the SQL
# said AVG("Sales"."salesamount" * "Returns"."returnquantity") over
# a FROM with no Returns in it.
#
# Where nothing reaches everything the facts are genuinely
# independent, and the fallback below leaves CFL and the conformed
# anchor path to deal with them.
#
# Only a measure that *by itself* reads several objects constrains
# the base, because only it needs them on one row. Constraining on
# the union of every measure's objects instead re-bases ordinary
# multi-fact queries: two independent measures would base at
# whichever fact reaches the other, and the reached fact's rows
# would then repeat once per row of the base. That is the fanout
# those queries stay on CFL to avoid.
spanning: set[str] = set()
for resolved_measure in ctx.result.measures:
model_measure = ctx.model.effective_measures.get(resolved_measure.name)
if model_measure is None:
continue
# A measure filter is evaluated on the same row as the value it
# gates (``CASE WHEN <filter> THEN value END``), so its objects
# have to be on that row too. Left out, a filter on Orders
# gating a Customers value based the query on Customers and
# named "orders" in a FROM that never joins it.
objects = set(model_measure.source_objects)
for item in model_measure.filters:
collect_measure_filter_objects(item, objects)
if len(objects) > 1:
spanning |= objects
sources = ctx.result.measure_source_objects
candidates = sorted(sources)
graph = JoinGraph.of(ctx.model, use_path_names=ctx.result.use_path_names or None)
if spanning:
candidates = [
name for name in candidates if spanning <= (graph.descendants(name) | {name})
]
if not candidates:
# No source reaches the rest, but an object beside them may:
# a Customers value gated by a Products filter is on one row
# only through the Orders that join both.
root = graph.find_common_root(spanning)
if root and spanning <= graph.descendants(root) | {root}:
return root
candidates = sorted(sources)
# Most joins first; between equals, the one reaching more of the
# query. Customers and Orders with one join each, under a dimension
# on Orders: Customers, first by name, cannot reach Orders, and the
# one join tree was planned as independent facts. Reached without
# multiplying the candidate's rows, and only where its star keeps
# the other sources' rows without children. A candidate that
# reaches some measure source only by multiplying its rows has no
# reach at all: a star on it fans that measure out, where CFL
# keeps it whole.
wanted = ctx.result.required_objects | sources | (filter_objects or set())
best = ""
best_rank = (-1, -1)
for obj_name in candidates:
obj = ctx.model.data_objects.get(obj_name)
safe = graph.descendants_without_fanout(obj_name) | {obj_name}
reach = (
len(wanted & safe)
if sources <= safe and self._keeps_childless_parents(ctx, obj_name)
else 0
)
rank = (len(obj.joins) if obj else 0, reach)
if rank > best_rank:
best = obj_name
best_rank = rank
if best:
return self._reanchor_if_unreachable(ctx, best, filter_objects or set())
# Dimension-only: use JoinGraph to find the deepest ancestor
# (possibly an intermediate fact/bridge table) that can reach
# all required dimension objects via directed join paths.
# (See ``_reanchor_if_unreachable`` — the same idea, applied when a
# measure pinned the base to an object that cannot reach the rest.)
if len(ctx.result.required_objects) > 1:
graph = JoinGraph.of(ctx.model, use_path_names=ctx.result.use_path_names or None)
root = graph.find_common_root(ctx.result.required_objects)
if root:
return root
for obj_name in sorted(ctx.result.required_objects):
obj = ctx.model.data_objects.get(obj_name)
if obj and obj.joins:
return obj_name
if ctx.result.required_objects:
return next(iter(sorted(ctx.result.required_objects)))
if ctx.model.data_objects:
return next(iter(ctx.model.data_objects))
return ""
# -- usePathNames validation ---------------------------------------------
def _validate_use_path_names(
self, ctx: _ResolutionContext, use_path_names: list[UsePathName]
) -> None:
"""Validate usePathNames references."""
for upn in use_path_names:
if upn.source not in ctx.model.data_objects:
ctx.errors.append(
SemanticError(
code="UNKNOWN_DATA_OBJECT",
message=f"usePathNames references unknown data object '{upn.source}'",
path="usePathNames",
)
)
continue
if upn.target not in ctx.model.data_objects:
ctx.errors.append(
SemanticError(
code="UNKNOWN_DATA_OBJECT",
message=f"usePathNames references unknown data object '{upn.target}'",
path="usePathNames",
)
)
continue
source_obj = ctx.model.data_objects[upn.source]
found = any(
j.join_to == upn.target and j.secondary and j.path_name == upn.path_name
for j in source_obj.joins
)
if not found:
ctx.errors.append(
SemanticError(
code="UNKNOWN_PATH_NAME",
message=(
f"No secondary join with pathName '{upn.path_name}' "
f"from '{upn.source}' to '{upn.target}'"
),
path="usePathNames",
)
)
# -- static model filters ------------------------------------------------
def _resolve_static_filter(
self, ctx: _ResolutionContext, mf: ModelFilter
) -> ResolvedFilter | None:
"""Resolve a static model filter to a physical WHERE expression.
Silently skips filters on data objects that are unreachable from the
query's join graph — they are simply irrelevant to the current query.
"""
return filter_resolution.resolve_static_filter(self, ctx, mf)
# -- filters -------------------------------------------------------------
def _resolve_filter_object(
self,
ctx: _ResolutionContext,
obj_name: str,
filter_path: str,
_field_label: str,
*,
static: bool = False,
) -> bool:
"""Ensure *obj_name* is joined; auto-extend if reachable.
An unreachable object refuses a query filter and skips a static one;
see ``filter_resolution.resolve_filter_object``.
"""
return filter_resolution.resolve_filter_object(
self, ctx, obj_name, filter_path, _field_label, static=static
)
def _resolve_filter_item(
self, ctx: _ResolutionContext, item: QueryFilterItem, *, is_having: bool
) -> ResolvedFilter | None:
"""Resolve a filter item (leaf or group) to a physical expression."""
return filter_resolution.resolve_filter_item(self, ctx, item, is_having=is_having)
def _resolve_filter_group(
self, ctx: _ResolutionContext, group: QueryFilterGroup, *, is_having: bool
) -> ResolvedFilter | None:
"""Resolve a filter group recursively, combining with AND/OR."""
return filter_resolution.resolve_filter_group(self, ctx, group, is_having=is_having)
def _resolve_filter(
self, ctx: _ResolutionContext, qf: QueryFilter, *, is_having: bool
) -> ResolvedFilter | None:
"""Resolve a query filter to a physical expression.
Filter fields can reference:
1. A dimension name (e.g. ``"Order Priority"``)
2. A qualified column ``"DataObject.Column"`` (e.g. ``"Orders.Order Priority"``)
3. For HAVING filters, a measure name (e.g. ``"Revenue"``)
If the referenced data object is reachable but not yet joined, the
join path is auto-extended.
"""
return filter_resolution.resolve_filter(self, ctx, qf, is_having=is_having)
# -- order by ------------------------------------------------------------
def _resolve_order_by_field(
self, ctx: _ResolutionContext, field_name: str, select_count: int
) -> Expr | None:
"""Resolve an order-by field to its expression."""
return filter_resolution.resolve_order_by_field(self, ctx, field_name, select_count)