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

tests_metricflow/integration/test_configured_cases.py

@@ -20,7 +20,7 @@
 from metricflow_semantics.test_helpers.config_helpers import MetricFlowTestConfiguration
 from metricflow_semantics.test_helpers.time_helpers import ConfigurableTimeSource
 
-from metricflow.engine.metricflow_engine import MetricFlowEngine, MetricFlowQueryRequest
+from metricflow.engine.metricflow_engine import MetricFlowEngine, MetricFlowQueryRequest, MetricFlowQueryResult
 from metricflow.plan_conversion.time_spine import TimeSpineSource
 from metricflow.protocols.sql_client import SqlClient
 from metricflow.sql.sql_exprs import (
@@ -37,6 +37,7 @@
     SqlSubtractTimeIntervalExpression,
 )
 from tests_metricflow.compare_df import assert_dataframes_equal
+from tests_metricflow.dataflow.optimizer.source_scan.test_source_scan_optimizer import DataflowPlanLookup
 from tests_metricflow.integration.configured_test_case import (
     CONFIGURED_INTEGRATION_TESTS_REPOSITORY,
     IntegrationTestModel,
@@ -294,7 +295,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,
@@ -361,3 +362,16 @@ def test_case(
     )
     # If we sort, it's effectively not checking the order whatever order that the output was would be overwritten.
     assert_dataframes_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_plan_lookup = DataflowPlanLookup(query_result.dataflow_plan)
+    dataflow_queried_semantic_models = dataflow_plan_lookup.read_semantic_models()
+
+    assert tuple(parse_query_result.queried_semantic_models) == tuple(dataflow_queried_semantic_models)