Fill nulls for multi-metric queries

metricflow/model/semantics/metric_lookup.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 import logging
-from typing import Dict, FrozenSet, List, Sequence
+from typing import Dict, FrozenSet, List, Optional, Sequence
 
+from dbt_semantic_interfaces.enum_extension import assert_values_exhausted
 from dbt_semantic_interfaces.implementations.filters.where_filter import PydanticWhereFilterIntersection
 from dbt_semantic_interfaces.implementations.metric import PydanticMetricTimeWindow
 from dbt_semantic_interfaces.protocols import WhereFilter
-from dbt_semantic_interfaces.protocols.metric import Metric, MetricType
+from dbt_semantic_interfaces.protocols.metric import Metric, MetricInputMeasure, MetricType
 from dbt_semantic_interfaces.protocols.semantic_manifest import SemanticManifest
 from dbt_semantic_interfaces.references import MetricReference
 
@@ -105,6 +106,23 @@ def add_metric(self, metric: Metric) -> None:
                 )
         self._metrics[metric_reference] = metric
 
+    def yaml_input_measure_for_metric(self, metric_reference: MetricReference) -> Optional[MetricInputMeasure]:
+        """Get input measure defined in the metric YAML, if exists.
+
+        When SemanticModel is constructed, input measures from input metrics are added to the list of input measures
+        for a metric. Here, use rules about metric types to determine which input measures were defined in the YAML:
+        - Simple & cumulative metrics require one input measure, and can't take any input metrics.
+        - Derived & ratio metrics take no input measures, only input metrics.
+        """
+        metric = self.get_metric(metric_reference=metric_reference)
+        if metric.type is MetricType.CUMULATIVE or metric.type is MetricType.SIMPLE:
+            assert len(metric.input_measures) == 1, "Simple and cumulative metrics should have one input measure."
+            return metric.input_measures[0]
+        elif metric.type is MetricType.RATIO or metric.type is MetricType.DERIVED:
+            return None
+        else:
+            assert_values_exhausted(metric.type)
+
     def measures_for_metric(
         self,
         metric_reference: MetricReference,

metricflow/plan_conversion/dataflow_to_sql.py

@@ -897,24 +897,16 @@ def _make_select_columns_for_multiple_metrics(
 
                 # At this point, the MetricSpec might have the alias in place of the element name, so we need to look
                 # back at where it was defined from to get the metric element name.
-                metric_name = metric_instance.defined_from.metric_name
-                input_measures = self._metric_lookup.measures_for_metric(
-                    metric_reference=MetricReference(metric_name),
-                    column_association_resolver=self._column_association_resolver,
-                )
-                # If multiple input measures, this is a query with a nested derived/ratio metric. In that case, we can
-                # skip this step because it has already occurred when rendering the nested metric.
-                # TODO: this logic might need updating for conversion metrics, which will allow multiple input measures
-                if len(input_measures) == 1:
-                    input_measure = input_measures[0]
-                    if input_measure.fill_nulls_with is not None:
-                        select_expression = SqlAggregateFunctionExpression(
-                            sql_function=SqlFunction.COALESCE,
-                            sql_function_args=[
-                                select_expression,
-                                SqlStringExpression(str(input_measure.fill_nulls_with)),
-                            ],
-                        )
+                metric_reference = MetricReference(element_name=metric_instance.defined_from.metric_name)
+                input_measure = self._metric_lookup.yaml_input_measure_for_metric(metric_reference=metric_reference)
+                if input_measure and input_measure.fill_nulls_with is not None:
+                    select_expression = SqlAggregateFunctionExpression(
+                        sql_function=SqlFunction.COALESCE,
+                        sql_function_args=[
+                            select_expression,
+                            SqlStringExpression(str(input_measure.fill_nulls_with)),
+                        ],
+                    )
 
                 select_columns.append(
                     SqlSelectColumn(