Add compiled_sql to NodeRelation for ephemeral model support

.changes/unreleased/Features-20260313-140000.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add compiled_sql to NodeRelation to support ephemeral models in Semantic Layer definitions.
+time: 2026-03-13T14:00:00.000000-07:00
+custom:
+  Author: b-per
+  Issue: "1993"

metricflow-semantic-interfaces/metricflow_semantic_interfaces/implementations/node_relation.py

@@ -17,6 +17,7 @@ class PydanticNodeRelation(HashableBaseModel, ProtocolHint[NodeRelation]):
     schema_name: str
     database: Optional[str] = None
     relation_name: str = ""
+    compiled_sql: Optional[str] = None
 
     @override
     def _implements_protocol(self) -> NodeRelation:  # noqa: D102

metricflow-semantic-interfaces/metricflow_semantic_interfaces/parsing/generated_json_schemas/default_explicit_schema.json

@@ -587,6 +587,9 @@
                 "alias": {
                     "type": "string"
                 },
+                "compiled_sql": {
+                    "type": "string"
+                },
                 "database": {
                     "type": "string"
                 },

metricflow-semantic-interfaces/metricflow_semantic_interfaces/parsing/schemas.py

@@ -351,6 +351,7 @@
         "schema_name": {"type": "string"},
         "database": {"type": "string"},
         "relation_name": {"type": "string"},
+        "compiled_sql": {"type": "string"},
     },
     "additionalProperties": False,
     "required": ["alias", "schema_name"],

metricflow-semantic-interfaces/metricflow_semantic_interfaces/protocols/node_relation.py

@@ -26,3 +26,9 @@ def database(self) -> Optional[str]:  # noqa: D102
     @abstractmethod
     def relation_name(self) -> str:  # noqa: D102
         pass
+
+    @property
+    @abstractmethod
+    def compiled_sql(self) -> Optional[str]:
+        """The compiled SQL for ephemeral models. When set, this is used as the source instead of relation_name."""
+        pass

metricflow-semantic-interfaces/tests/implementations/test_node_relation.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from metricflow_semantic_interfaces.implementations.node_relation import PydanticNodeRelation
+
+
+def test_node_relation_compiled_sql_defaults_to_none() -> None:
+    """Test that compiled_sql is None by default for non-ephemeral models."""
+    node_relation = PydanticNodeRelation(schema_name="my_schema", alias="my_table")
+    assert node_relation.compiled_sql is None
+    assert node_relation.relation_name == "my_schema.my_table"
+
+
+def test_node_relation_with_compiled_sql() -> None:
+    """Test that compiled_sql is preserved when set (ephemeral model)."""
+    compiled_sql = "SELECT id, name FROM raw.source_table WHERE active = true"
+    node_relation = PydanticNodeRelation(
+        schema_name="my_schema",
+        alias="my_table",
+        compiled_sql=compiled_sql,
+    )
+    assert node_relation.compiled_sql == compiled_sql
+    assert node_relation.relation_name == "my_schema.my_table"
+
+
+def test_node_relation_from_string_has_no_compiled_sql() -> None:
+    """Test that from_string produces a node relation without compiled_sql."""
+    node_relation = PydanticNodeRelation.from_string("my_schema.my_table")
+    assert node_relation.compiled_sql is None
+
+
+def test_node_relation_with_database_and_compiled_sql() -> None:
+    """Test compiled_sql with a fully qualified relation (database.schema.table)."""
+    compiled_sql = "SELECT 1"
+    node_relation = PydanticNodeRelation(
+        database="my_db",
+        schema_name="my_schema",
+        alias="my_table",
+        compiled_sql=compiled_sql,
+    )
+    assert node_relation.compiled_sql == compiled_sql
+    assert node_relation.relation_name == "my_db.my_schema.my_table"

metricflow-semantic-interfaces/tests/parsing/test_semantic_model_parsing.py

@@ -97,6 +97,48 @@ def test_semantic_model_node_relation_parsing() -> None:
     assert semantic_model.node_relation.relation_name == "some_schema.source_table"
 
 
+def test_semantic_model_node_relation_with_compiled_sql() -> None:
+    """Test for parsing a semantic model with compiled_sql set on the node_relation (ephemeral model)."""
+    yaml_contents = textwrap.dedent(
+        """\
+        semantic_model:
+          name: ephemeral_test
+          node_relation:
+            alias: source_table
+            schema_name: some_schema
+            compiled_sql: "SELECT id, name FROM raw.source_table WHERE active = true"
+        """
+    )
+    file = YamlConfigFile(filepath="inline_for_test", contents=yaml_contents)
+
+    build_result = parse_yaml_files_to_semantic_manifest(files=[file, EXAMPLE_PROJECT_CONFIGURATION_YAML_CONFIG_FILE])
+
+    assert len(build_result.semantic_manifest.semantic_models) == 1
+    semantic_model = build_result.semantic_manifest.semantic_models[0]
+    assert semantic_model.node_relation.compiled_sql == "SELECT id, name FROM raw.source_table WHERE active = true"
+    assert semantic_model.node_relation.relation_name == "some_schema.source_table"
+
+
+def test_semantic_model_node_relation_without_compiled_sql() -> None:
+    """Test that compiled_sql defaults to None when not provided."""
+    yaml_contents = textwrap.dedent(
+        """\
+        semantic_model:
+          name: table_test
+          node_relation:
+            alias: source_table
+            schema_name: some_schema
+        """
+    )
+    file = YamlConfigFile(filepath="inline_for_test", contents=yaml_contents)
+
+    build_result = parse_yaml_files_to_semantic_manifest(files=[file, EXAMPLE_PROJECT_CONFIGURATION_YAML_CONFIG_FILE])
+
+    assert len(build_result.semantic_manifest.semantic_models) == 1
+    semantic_model = build_result.semantic_manifest.semantic_models[0]
+    assert semantic_model.node_relation.compiled_sql is None
+
+
 def test_base_semantic_model_entity_parsing() -> None:
     """Test parsing base attributes of PydanticEntity object."""
     label = "Base Test Entity"

metricflow/dataset/convert_semantic_model.py

@@ -49,9 +49,11 @@
 from metricflow.dataset.semantic_model_adapter import SemanticModelDataSet
 from metricflow.dataset.sql_dataset import SqlDataSet
 from metricflow.sql.sql_plan import (
+    SqlPlanNode,
     SqlSelectColumn,
 )
 from metricflow.sql.sql_select_node import SqlSelectStatementNode
+from metricflow.sql.sql_select_text_node import SqlSelectTextNode
 from metricflow.sql.sql_table_node import SqlTableNode
 
 logger = logging.getLogger(__name__)
@@ -511,7 +513,16 @@ def create_sql_source_data_set(self, model_reference: SemanticModelReference) ->
             all_select_columns.extend(select_columns)
 
         # Generate the "from" clause depending on whether it's an SQL query or an SQL table.
-        from_source = SqlTableNode.create(sql_table=SqlTable.from_string(semantic_model.node_relation.relation_name))
+        # TODO: Use `semantic_model.node_relation.compiled_sql` directly once the published
+        # dbt-semantic-interfaces package includes the `compiled_sql` field on `NodeRelation`.
+        compiled_sql: Optional[str] = getattr(semantic_model.node_relation, "compiled_sql", None)
+        from_source: SqlPlanNode
+        if compiled_sql is not None:
+            from_source = SqlSelectTextNode.create(select_query=compiled_sql)
+        else:
+            from_source = SqlTableNode.create(
+                sql_table=SqlTable.from_string(semantic_model.node_relation.relation_name)
+            )
 
         select_statement_node = SqlSelectStatementNode.create(
             description=f"Read Elements From Semantic Model '{semantic_model.name}'",

tests_metricflow/dataset/test_convert_semantic_model.py

@@ -5,11 +5,16 @@
 
 import pytest
 from _pytest.fixtures import FixtureRequest
+from dbt_semantic_interfaces.implementations.node_relation import PydanticNodeRelation
 from dbt_semantic_interfaces.references import SemanticModelReference
+from metricflow_semantics.sql.sql_table import SqlTable
 from metricflow_semantics.test_helpers.config_helpers import MetricFlowTestConfiguration
 from metricflow_semantics.test_helpers.snapshot_helpers import assert_spec_set_snapshot_equal
 
 from metricflow.protocols.sql_client import SqlClient
+from metricflow.sql.sql_plan import SqlPlanNode
+from metricflow.sql.sql_select_text_node import SqlSelectTextNode
+from metricflow.sql.sql_table_node import SqlTableNode
 from tests_metricflow.fixtures.manifest_fixtures import MetricFlowEngineTestFixture, SemanticManifestSetup
 from tests_metricflow.sql.compare_sql_plan import assert_rendered_sql_equal
 
@@ -100,3 +105,63 @@ def test_convert_query_semantic_model(  # noqa: D103
         sql_plan_node=bookings_data_set.checked_sql_select_node,
         sql_client=sql_client,
     )
+
+
+_DSI_HAS_COMPILED_SQL = hasattr(PydanticNodeRelation, "compiled_sql")
+
+
+@pytest.mark.skipif(not _DSI_HAS_COMPILED_SQL, reason="installed dbt-semantic-interfaces lacks compiled_sql")
+def test_from_source_uses_sql_table_node_for_table_models() -> None:
+    """When compiled_sql is not set, the from_source should be a SqlTableNode."""
+    node_relation = PydanticNodeRelation(schema_name="my_schema", alias="my_table")
+
+    assert getattr(node_relation, "compiled_sql", None) is None
+    from_source = SqlTableNode.create(sql_table=SqlTable.from_string(node_relation.relation_name))
+    assert isinstance(from_source, SqlTableNode)
+    assert from_source.sql_table == SqlTable.from_string("my_schema.my_table")
+
+
+@pytest.mark.skipif(not _DSI_HAS_COMPILED_SQL, reason="installed dbt-semantic-interfaces lacks compiled_sql")
+def test_from_source_uses_sql_select_text_node_for_ephemeral_models() -> None:
+    """When compiled_sql is set (ephemeral model), the from_source should be a SqlSelectTextNode."""
+    compiled_sql = "SELECT id, name FROM raw.source_table WHERE active = true"
+    node_relation = PydanticNodeRelation(
+        schema_name="my_schema",
+        alias="my_table",
+        compiled_sql=compiled_sql,
+    )
+
+    assert getattr(node_relation, "compiled_sql", None) is not None
+    from_source = SqlSelectTextNode.create(select_query=node_relation.compiled_sql)
+    assert isinstance(from_source, SqlSelectTextNode)
+    assert from_source.select_query == compiled_sql
+
+
+@pytest.mark.skipif(not _DSI_HAS_COMPILED_SQL, reason="installed dbt-semantic-interfaces lacks compiled_sql")
+def test_from_source_branching_logic() -> None:
+    """Test the branching logic used in the converter to select the right from_source node type."""
+    # Table-based model (no compiled_sql)
+    table_relation = PydanticNodeRelation(schema_name="my_schema", alias="my_table")
+    table_compiled_sql = getattr(table_relation, "compiled_sql", None)
+    table_from_source: SqlPlanNode
+    if table_compiled_sql is not None:
+        table_from_source = SqlSelectTextNode.create(select_query=table_compiled_sql)
+    else:
+        table_from_source = SqlTableNode.create(sql_table=SqlTable.from_string(table_relation.relation_name))
+    assert isinstance(table_from_source, SqlTableNode)
+
+    # Ephemeral model (with compiled_sql)
+    compiled_sql = "SELECT 1 AS id"
+    ephemeral_relation = PydanticNodeRelation(
+        schema_name="my_schema",
+        alias="my_table",
+        compiled_sql=compiled_sql,
+    )
+    ephemeral_compiled_sql = getattr(ephemeral_relation, "compiled_sql", None)
+    ephemeral_from_source: SqlPlanNode
+    if ephemeral_compiled_sql is not None:
+        ephemeral_from_source = SqlSelectTextNode.create(select_query=ephemeral_compiled_sql)
+    else:
+        ephemeral_from_source = SqlTableNode.create(sql_table=SqlTable.from_string(ephemeral_relation.relation_name))
+    assert isinstance(ephemeral_from_source, SqlSelectTextNode)
+    assert ephemeral_from_source.select_query == compiled_sql