fix: preserve metric semantics across fact sources

docs/concepts/metrics/definition.md

@@ -59,7 +59,7 @@ Because the `prod.users.country` and `prod.searches.num_searches` models have sp
 
 Metrics can perform additional operations/calculations with other metrics.
 
-In this example, the third metric `clicks_per_search` is calculated by dividing the first metric `total_searches` by the second metric `total_clicks`:
+In this example, the third metric `clicks_per_search` is calculated by dividing `total_clicks` by `total_searches`:
 
 ```sql linenums="1"
 METRIC (
@@ -78,6 +78,8 @@ METRIC (
 );
 ```
 
+Metric references are case insensitive, including quoted references, and may refer to metrics defined later in the project. Shared dependencies are resolved once during expansion. An unknown dependency or a dependency cycle raises a configuration error with the dependency chain and the path of the definition containing the invalid reference.
+
 ## Properties
 
 The `METRIC` definition supports the following keys. The `name` and `expression` keys are required.

docs/concepts/metrics/overview.md

@@ -41,7 +41,7 @@ FROM __semantic.__table  -- special table for simple metric queries
 GROUP BY ds
 ```
 
-When that model query is run, SQLMesh uses its semantic understanding of the query and metrics definitions to generate the code that is actually executed by the SQL engine:
+SQLMesh expands the metric into SQL equivalent to the following query (generated aliases may differ):
 
 ``` sql linenums="1"
 SELECT
@@ -60,3 +60,15 @@ FROM (
 ```
 
 SQLMesh automatically generates the correct join to use values from both the `sushi.orders` and `sushi.customers` tables.
+
+## Filters and grouping across metrics
+
+Queries against `__semantic.__table` treat its columns as logical dimensions. A `WHERE` predicate is resolved independently for every contributing fact source and applied before aggregation, including when the filtered dimension is not in `GROUP BY`. Derived metrics are calculated from those filtered aggregates.
+
+Dimension resolution prefers a column on the fact itself. Otherwise, the column must resolve to one reachable model through the configured grains and references. An explicit dimension-table alias selects that model instead of a same-named fact column. Unknown, unreachable, or ambiguous dimensions are rejected rather than ignored.
+
+Fact aggregates are combined with a full join by default. Group keys present only in a later fact are retained, and matching `NULL` group keys are combined, including composite keys. Missing metric values remain `NULL`; they are not automatically converted to zero.
+
+On PostgreSQL, full joins use composite-key equality to retain `NULL` groups without the planner restriction on `IS NOT DISTINCT FROM` join conditions. Corresponding dimensions must have matching PostgreSQL types; use an explicit cast in the grouping expression when models expose different types. Metric arithmetic follows the definition's SQL dialect: use a numeric or floating-point cast for fractional ratios of integer counts, and `NULLIF(denominator, 0)` when a zero denominator should produce `NULL`.
+
+The prototype rejects subqueries in metric `WHERE` filters, grouping sets, and reference paths that cannot be compiled into safe matching-key joins. This includes multi-hop paths that change reference keys. These checks do not restrict ordinary SQL scopes without `METRIC` expressions. They are not a substitute for application authorization.

docs/development.md

@@ -68,6 +68,42 @@ Run more comprehensive tests that run on each commit:
 make slow-test
 ```
 
+### PostgreSQL Metrics acceptance
+
+The native Metrics suite loads a temporary SQLMesh project, applies its models to PostgreSQL, compiles queries with `Context.rewrite`, and checks results returned by PostgreSQL. It uses the existing `inttest_postgres` gateway and `postgres` / `docker` test markers. Only scheduling metadata uses an isolated in-memory DuckDB connection; model and metric SQL run on PostgreSQL. Temporary model schemas are cleaned up by the integration fixtures.
+
+With the project's PostgreSQL test service available (see `make engine-postgres-up`), run:
+
+```bash
+pytest tests/core/engine_adapter/integration/test_integration_metrics.py -q
+```
+
+On a Linux Docker host, a separate test container can instead be run without publishing a database port. After activating the development virtual environment:
+
+```bash
+(
+  set -eu
+  name="sqlmesh-metrics-pg-$$"
+  docker network create --internal "$name"
+  trap 'docker rm -f "$name" >/dev/null 2>&1 || true; docker network rm "$name" >/dev/null 2>&1 || true' EXIT
+  docker run -d --rm --name "$name" --network "$name" \
+    -e POSTGRES_HOST_AUTH_METHOD=trust postgres:16-alpine
+  ready=false
+  for attempt in $(seq 1 30); do
+    if docker exec "$name" pg_isready -h 127.0.0.1 -U postgres; then
+      ready=true
+      break
+    fi
+    sleep 1
+  done
+  "$ready"
+  export DOCKER_HOSTNAME="$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$name")"
+  pytest tests/core/engine_adapter/integration/test_integration_metrics.py -q
+)
+```
+
+The trust-authenticated container is for disposable tests only: its internal network has no published host port or persistent data volume. Do not point this suite at a business database. Personal SQLMesh gateway overrides must not redirect `inttest_postgres` away from the intended test instance.
+
 ## Documentation
 
 In order to run the documentation server, you will need to install the dependencies by running the following command.

sqlmesh/core/metric/definition.py

@@ -99,43 +99,71 @@ def to_metric(
         self, metas: t.Dict[str, MetricMeta], metrics: UniqueKeyDict[str, Metric]
     ) -> Metric:
         """Converts a metric meta into a fully expanded and standalone metric."""
-        metric_refs = {}
-        agg_or_ref = False
-
-        for node in self.expression.walk():
-            if isinstance(node, exp.Alias):
-                _raise_metric_config_error(
-                    f"Alias found for metric '{self.name}' which is not allowed", self._path
-                )
-            elif isinstance(node, exp.AggFunc):
-                agg_or_ref = True
-            elif isinstance(node, exp.Column) and not node.table:
-                agg_or_ref = True
-                ref = node.sql(dialect=self.dialect)
-
-                if ref not in metrics:
-                    metrics[ref] = metas[ref].to_metric(metas, metrics)
-
-                metric_refs[node] = metrics[ref]
-
-        if not agg_or_ref:
-            _raise_metric_config_error(
-                f"Metric '{self.name}' missing an aggregation or metric ref", self._path
-            )
-
-        if metric_refs:
-            expanded = self.expression.copy()
-            for column in expanded.find_all(exp.Column):
-                metric = metric_refs.get(column)
-
-                if metric:
-                    column.replace(metric.expanded.copy())
-        else:
-            expanded = exp.alias_(self.expression, self.name)
-
-        metric = Metric(**self.dict(), expanded=expanded)
-        metric._path = self._path
-        return metric
+        # Suspend each expression walk while resolving a dependency so cycles of
+        # any depth can be diagnosed without using the Python call stack.
+        stack: t.List[t.Tuple[MetricMeta, t.Iterator[exp.Expr], t.Dict[exp.Column, str], bool]] = [
+            (self, self.expression.walk(), {}, False)
+        ]
+        visiting = {self.name}
+
+        while True:
+            meta, nodes, metric_refs, agg_or_ref = stack.pop()
+
+            for node in nodes:
+                if isinstance(node, exp.Alias):
+                    _raise_metric_config_error(
+                        f"Alias found for metric '{meta.name}' which is not allowed", meta._path
+                    )
+                elif isinstance(node, exp.AggFunc):
+                    agg_or_ref = True
+                elif isinstance(node, exp.Column) and not node.table:
+                    agg_or_ref = True
+                    ref = node.name.lower()
+                    metric_refs[node] = ref
+
+                    if ref not in metrics:
+                        is_cycle = ref in visiting
+                        if is_cycle or ref not in metas:
+                            dependency_path = " -> ".join(
+                                [frame[0].name for frame in stack] + [meta.name, ref]
+                            )
+                            if is_cycle:
+                                message = f"Metric dependency cycle detected: {dependency_path}"
+                            else:
+                                message = (
+                                    f"Unknown metric '{ref}' referenced by metric '{meta.name}' "
+                                    f"(dependency path: {dependency_path})"
+                                )
+                            _raise_metric_config_error(message, meta._path)
+
+                        dependency = metas[ref]
+                        stack.append((meta, nodes, metric_refs, agg_or_ref))
+                        stack.append((dependency, dependency.expression.walk(), {}, False))
+                        visiting.add(ref)
+                        break
+            else:
+                if not agg_or_ref:
+                    _raise_metric_config_error(
+                        f"Metric '{meta.name}' missing an aggregation or metric ref", meta._path
+                    )
+
+                if metric_refs:
+                    expanded = meta.expression.copy()
+                    for column in expanded.find_all(exp.Column):
+                        reference = metric_refs.get(column)
+                        if reference is not None:
+                            column.replace(metrics[reference].expanded.copy())
+                else:
+                    expanded = exp.alias_(meta.expression, meta.name)
+
+                metric = Metric(**meta.dict(), expanded=expanded)
+                metric._path = meta._path
+                visiting.remove(meta.name)
+
+                if not stack:
+                    return metric
+
+                metrics[meta.name] = metric
 
 
 class Metric(MetricMeta, frozen=True):