fix(dbt): substitute DERIVED metric references in a single pass

converters/dbt/src/ossie_dbt/msi_to_ossie.py

@@ -338,17 +338,50 @@ def _resolve_derived(
         """Resolve a DERIVED metric by substituting each input metric's expression into the expr string.
 
         Compound sub-expressions (DERIVED/RATIO) are wrapped in parentheses to preserve operator precedence.
+
+        All references are substituted in a single pass. Substituting them one at a
+        time would re-scan text inserted by an earlier reference, so a metric named
+        after a column appearing in an already-inlined expression would be expanded
+        twice. The replacement is a callback rather than a string so that backslashes
+        in the resolved SQL (e.g. from a `LIKE 'a\\b'` filter) are inserted verbatim
+        instead of being interpreted as `re.sub` template escapes.
+
+        Listing the same input metric twice under one reference is rejected when the
+        two occurrences resolve differently (e.g. distinct per-input filters and no
+        aliases): the expression has a single token for them, so either resolution
+        would be an arbitrary choice. MetricFlow does not reject this shape upstream —
+        `DerivedMetricRule._validate_alias_collision` only compares entries that set an
+        alias. Occurrences that resolve identically are redundant rather than ambiguous
+        and are accepted.
         """
         expr = metric.type_params.expr or ""
+        replacements: Dict[str, str] = {}
         for input_metric in metric.type_params.metrics or []:
             ref = input_metric.alias if input_metric.alias else input_metric.name
             dep_metric = self._lookup_metric(metric_index, input_metric.name, f"DERIVED metric '{metric.name}'")
             input_filter = _merge_filter_sqls(filter_sql, _collect_filter_sql(input_metric.filter))
             resolved = self._resolve_metric_expression(dep_metric, metric_index, cache, input_filter)
             if dep_metric.type in (MetricType.DERIVED, MetricType.RATIO):
                 resolved = f"({resolved})"
-            expr = re.sub(rf"\b{re.escape(ref)}\b", resolved, expr)
-        return expr
+            previous = replacements.get(ref)
+            if previous is not None and previous != resolved:
+                raise ValueError(
+                    "DERIVED metric references an input metric that is listed more than once with "
+                    "differing resolutions, making the reference ambiguous; give each occurrence a "
+                    f"distinct alias: metric_name={metric.name!r}, reference={ref!r}"
+                )
+            replacements[ref] = resolved
+
+        if not replacements:
+            return expr
+
+        # The `\b` anchors already stop a short reference from matching inside a
+        # longer identifier; sorting by length (then name) keeps the alternation order stable
+        # and independent of the order metrics happen to be declared in.
+        pattern = re.compile(
+            r"\b(" + "|".join(re.escape(ref) for ref in sorted(replacements, key=lambda ref: (-len(ref), ref))) + r")\b"
+        )
+        return pattern.sub(lambda match: replacements[match.group(0)], expr)
 
     @staticmethod
     def _build_entity_index(

converters/dbt/tests/test_msi_to_ossie.py

@@ -644,6 +644,139 @@ def test_derived_metric_uses_alias_for_substitution(self) -> None:
         profit_ossie = next(m for m in _ossie_metrics(result) if m.name == "profit")
         assert profit_ossie.expression.dialects[0].expression == "SUM(orders.amount) - SUM(orders.cost_amount)"
 
+    def test_derived_metric_does_not_re_expand_an_inlined_reference(self) -> None:
+        """A reference is substituted once, even when its name appears in already-inlined text.
+
+        `gross` inlines to `SUM(orders.net)`, which contains the name of the second
+        input metric. Substituting references one at a time would expand `net` inside
+        that text as well.
+        """
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[
+                _measure("gross", agg=AggregationType.SUM, expr="net"),
+                _measure("net", agg=AggregationType.SUM, expr="net_amount"),
+            ],
+        )
+        gross_m = _simple_metric("gross", "gross")
+        net_m = _simple_metric("net", "net")
+        margin = PydanticMetric(
+            name="margin",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="gross - net",
+                metrics=[
+                    PydanticMetricInput(name="gross"),
+                    PydanticMetricInput(name="net"),
+                ],
+            ),
+            filter=None,
+            metadata=default_meta(),
+            config=None,
+        )
+        result = (
+            MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[gross_m, net_m, margin])).output
+        )
+
+        margin_ossie = next(m for m in _ossie_metrics(result) if m.name == "margin")
+        assert margin_ossie.expression.dialects[0].expression == "SUM(orders.net) - SUM(orders.net_amount)"
+
+    def test_derived_metric_preserves_backslashes_from_a_filter(self) -> None:
+        """Backslashes in an inlined expression survive substitution verbatim.
+
+        A resolved expression is inserted as a literal, not as a `re.sub` replacement
+        template, so an escape sequence such as `\\b` in a filter's SQL is not
+        reinterpreted (`\\b` would otherwise become a backspace character).
+        """
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")],
+        )
+        revenue_m = _simple_metric("revenue", "revenue")
+        revenue_m.filter = _filter(r"{{ Dimension('order__path') }} LIKE 'a\b'")
+        scaled = PydanticMetric(
+            name="scaled",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="revenue * 2",
+                metrics=[PydanticMetricInput(name="revenue")],
+            ),
+            filter=None,
+            metadata=default_meta(),
+            config=None,
+        )
+        result = MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, scaled])).output
+
+        revenue_ossie = next(m for m in _ossie_metrics(result) if m.name == "revenue")
+        scaled_ossie = next(m for m in _ossie_metrics(result) if m.name == "scaled")
+        assert revenue_ossie.expression.dialects[0].expression == (
+            r"SUM(CASE WHEN order__path LIKE 'a\b' THEN orders.amount END)"
+        )
+        assert scaled_ossie.expression.dialects[0].expression == (
+            r"SUM(CASE WHEN order__path LIKE 'a\b' THEN orders.amount END) * 2"
+        )
+
+    def test_derived_metric_rejects_a_reference_listed_twice_with_differing_filters(self) -> None:
+        """An input metric listed twice under one reference, resolving differently, is ambiguous.
+
+        MetricFlow accepts this shape — `DerivedMetricRule._validate_alias_collision`
+        only compares entries that set an alias — so the converter has to reject it
+        rather than silently pick one of the two filters.
+        """
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")],
+        )
+        revenue_m = _simple_metric("revenue", "revenue")
+        both = PydanticMetric(
+            name="both",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="revenue",
+                metrics=[
+                    PydanticMetricInput(name="revenue", filter=_filter("{{ Dimension('order__region') }} = 'EU'")),
+                    PydanticMetricInput(name="revenue", filter=_filter("{{ Dimension('order__region') }} = 'US'")),
+                ],
+            ),
+            filter=None,
+            metadata=default_meta(),
+            config=None,
+        )
+        with pytest.raises(ValueError, match="listed more than once"):
+            MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, both]))
+
+    def test_derived_metric_accepts_a_reference_listed_twice_resolving_identically(self) -> None:
+        """A redundant duplicate is not ambiguous: both occurrences resolve to the same SQL."""
+        sm = semantic_model_with_guaranteed_meta(
+            name="orders",
+            measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")],
+        )
+        revenue_m = _simple_metric("revenue", "revenue")
+        doubled = PydanticMetric(
+            name="doubled",
+            description=None,
+            type=MetricType.DERIVED,
+            type_params=PydanticMetricTypeParams(
+                expr="revenue + revenue",
+                metrics=[
+                    PydanticMetricInput(name="revenue"),
+                    PydanticMetricInput(name="revenue"),
+                ],
+            ),
+            filter=None,
+            metadata=default_meta(),
+            config=None,
+        )
+        result = (
+            MSIToOssieConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, doubled])).output
+        )
+
+        doubled_ossie = next(m for m in _ossie_metrics(result) if m.name == "doubled")
+        assert doubled_ossie.expression.dialects[0].expression == "SUM(orders.amount) + SUM(orders.amount)"
+
     def test_derived_metric_nested(self, snapshot: SnapshotAssertion) -> None:
         sm = semantic_model_with_guaranteed_meta(
             name="orders",