Add test case to check semantic models between dataflow and parsing.

metricflow/execution/dataflow_to_execution.py

@@ -114,8 +114,7 @@ def visit_write_to_result_table_node(self, node: WriteToResultTableNode) -> Conv
 
     def convert_to_execution_plan(self, dataflow_plan: DataflowPlan) -> ConvertToExecutionPlanResult:
         """Convert the dataflow plan to an execution plan."""
-        assert len(dataflow_plan.sink_nodes) == 1, "Only 1 sink node in the plan is currently supported."
-        return dataflow_plan.sink_nodes[0].accept(self)
+        return dataflow_plan.sink_node.accept(self)
 
     @override
     def visit_source_node(self, node: ReadSqlSourceNode) -> ConvertToExecutionPlanResult:

tests_metricflow/integration/test_configured_cases.py

@@ -12,11 +12,8 @@
 from dbt_semantic_interfaces.implementations.elements.measure import PydanticMeasureAggregationParameters
 from dbt_semantic_interfaces.type_enums.date_part import DatePart
 from dbt_semantic_interfaces.type_enums.time_granularity import TimeGranularity
-from metricflow_semantics.protocols.query_parameter import DimensionOrEntityQueryParameter
-from metricflow_semantics.specs.query_param_implementations import DimensionOrEntityParameter, TimeDimensionParameter
-from metricflow_semantics.test_helpers.config_helpers import MetricFlowTestConfiguration
 
-from metricflow.engine.metricflow_engine import MetricFlowQueryRequest
+from metricflow.engine.metricflow_engine import MetricFlowQueryRequest, MetricFlowQueryResult
 from metricflow.plan_conversion.time_spine import TimeSpineSource
 from metricflow.protocols.sql_client import SqlClient
 from metricflow.sql.sql_exprs import (
@@ -32,7 +29,11 @@
     SqlStringExpression,
     SqlSubtractTimeIntervalExpression,
 )
+from metricflow_semantics.protocols.query_parameter import DimensionOrEntityQueryParameter
+from metricflow_semantics.specs.query_param_implementations import DimensionOrEntityParameter, TimeDimensionParameter
+from metricflow_semantics.test_helpers.config_helpers import MetricFlowTestConfiguration
 from tests_metricflow.fixtures.manifest_fixtures import MetricFlowEngineTestFixture, SemanticManifestSetup
+
 from tests_metricflow.integration.configured_test_case import (
     CONFIGURED_INTEGRATION_TESTS_REPOSITORY,
     IntegrationTestModel,
@@ -279,7 +280,7 @@ def test_case(
             group_by.append(TimeDimensionParameter(**kwargs))
         else:
             group_by.append(DimensionOrEntityParameter(**kwargs))
-    query_result = engine.query(
+    query_result: MetricFlowQueryResult = engine.query(
         MetricFlowQueryRequest.create_with_random_request_id(
             metric_names=case.metrics,
             group_by_names=case.group_bys if len(case.group_bys) > 0 else None,
@@ -343,3 +344,14 @@ def test_case(
     # If we sort, it's effectively not checking the order whatever order that the output was would be overwritten.
     assert actual is not None, "Did not get a result table from MetricFlow"
     assert_data_tables_equal(actual, expected, sort_columns=not case.check_order, allow_empty=case.allow_empty)
+
+    # Check that the parse result and the dataflow plan show the same semantic models read.
+    if name in {"itest_dimensions.yaml/distinct_values_query_with_metric_filter"}:
+        pytest.skip(
+            "Skipping the congruence check for semantic models queried by the parser vs. the dataflow as "
+            "metrics-in-filters is a WIP."
+        )
+    parse_query_result = query_result.explain_result.parse_query_result
+    dataflow_queried_semantic_models = query_result.explain_result.dataflow_plan.source_semantic_models
+
+    assert tuple(parse_query_result.queried_semantic_models) == tuple(dataflow_queried_semantic_models)