add the dbt node name

Author: themisvaltinos

sqlmesh/core/context.py

@@ -1677,7 +1677,11 @@ def plan_builder(
             end_override_per_model=max_interval_end_per_model,
             console=self.console,
             user_provided_flags=user_provided_flags,
-            selected_models=model_selector.expand_model_selections(select_models or "*"),
+            selected_models={
+                node_name
+                for model in model_selector.expand_model_selections(select_models or "*")
+                if (node_name := snapshots[model].node.node_name)
+            },
             explain=explain or False,
             ignore_cron=ignore_cron or False,
         )

sqlmesh/core/node.py

@@ -199,6 +199,7 @@ class _Node(PydanticModel):
     interval_unit_: t.Optional[IntervalUnit] = Field(alias="interval_unit", default=None)
     tags: t.List[str] = []
     stamp: t.Optional[str] = None
+    node_name: t.Optional[str] = None  # dbt node name
     _path: t.Optional[Path] = None
     _data_hash: t.Optional[str] = None
     _metadata_hash: t.Optional[str] = None

sqlmesh/core/scheduler.py

@@ -812,7 +812,7 @@ def _run_or_audit(
             run_environment_statements=run_environment_statements,
             audit_only=audit_only,
             auto_restatement_triggers=auto_restatement_triggers,
-            selected_models=selected_snapshots or {s.name for s in merged_intervals},
+            selected_models={s.node.node_name for s in merged_intervals if s.node.node_name},
         )
 
         return CompletionStatus.FAILURE if errors else CompletionStatus.SUCCESS

sqlmesh/dbt/adapter.py

@@ -181,10 +181,6 @@ def graph(self) -> t.Any:
             }
         )
 
-    @property
-    def selected_resources(self) -> t.List[str]:
-        return []
-
 
 class ParsetimeAdapter(BaseAdapter):
     def get_relation(self, database: str, schema: str, identifier: str) -> t.Optional[BaseRelation]:
@@ -505,8 +501,3 @@ def _normalize(self, input_table: exp.Table) -> exp.Table:
             normalized_table.set("db", normalized_table.this)
             normalized_table.set("this", None)
         return normalized_table
-
-    def _dbt_model_id(self, sqlmesh_model_name: str) -> str:
-        # Model prefix is needed to correspond to the key in the nodes within the dbt context variable
-        parts = [part.strip('"') for part in sqlmesh_model_name.split(".")]
-        return f"model.{parts[0]}.{parts[-1]}"

sqlmesh/dbt/builtin.py

@@ -545,7 +545,7 @@ def create_builtin_globals(
             "run_query": sql_execution.run_query,
             "statement": sql_execution.statement,
             "graph": adapter.graph,
-            "selected_resources": adapter.selected_resources,
+            "selected_resources": list(jinja_globals.get("selected_models") or []),
         }
     )
 

sqlmesh/dbt/model.py

@@ -689,6 +689,7 @@ def to_sqlmesh(
             extract_dependencies_from_query=False,
             allow_partials=allow_partials,
             virtual_environment_mode=virtual_environment_mode,
+            node_name=self.node_name,
             **optional_kwargs,
             **model_kwargs,
         )

sqlmesh/dbt/seed.py

@@ -92,6 +92,7 @@ def to_sqlmesh(
             audit_definitions=audit_definitions,
             virtual_environment_mode=virtual_environment_mode,
             start=self.start or context.sqlmesh_config.model_defaults.start,
+            node_name=self.node_name,
             **kwargs,
         )
 

sqlmesh/migrations/v0097_add_node_name.py

@@ -0,0 +1,9 @@
+"""Add 'node_name' property to node definition."""
+
+
+def migrate_schemas(state_sync, **kwargs):  # type: ignore
+    pass
+
+
+def migrate_rows(state_sync, **kwargs):  # type: ignore
+    pass

tests/dbt/test_model.py

@@ -201,7 +201,7 @@ def test_load_microbatch_all_defined(
             concurrent_batches=true
         )
     }}
-    
+
     SELECT 1 as cola, '2025-01-01' as ds
     """
     microbatch_model_file = model_dir / "microbatch.sql"
@@ -633,3 +633,80 @@ def test_dbt_jinja_macro_undefined_variable_error(create_empty_project):
     assert "Failed to update model schemas" in error_message
     assert "Could not render jinja for" in error_message
     assert "Undefined macro/variable: 'columns' in macro: 'select_columns'" in error_message
+
+
+@pytest.mark.slow
+def test_node_name_populated_for_dbt_models(dbt_dummy_postgres_config: PostgresConfig) -> None:
+    model_config = ModelConfig(
+        name="test_model",
+        package_name="test_package",
+        sql="SELECT 1 as id",
+        database="test_db",
+        schema_="test_schema",
+        alias="test_model",
+    )
+
+    context = DbtContext()
+    context.project_name = "test_project"
+    context.target = dbt_dummy_postgres_config
+
+    # check after convert to SQLMesh model that node_name is populated correctly
+    sqlmesh_model = model_config.to_sqlmesh(context)
+    assert sqlmesh_model.node_name == "model.test_package.test_model"
+
+
+@pytest.mark.slow
+def test_load_model_dbt_node_name(tmp_path: Path) -> None:
+    yaml = YAML()
+    dbt_project_dir = tmp_path / "dbt"
+    dbt_project_dir.mkdir()
+    dbt_model_dir = dbt_project_dir / "models"
+    dbt_model_dir.mkdir()
+
+    model_contents = "SELECT 1 as id, 'test' as name"
+    model_file = dbt_model_dir / "simple_model.sql"
+    with open(model_file, "w", encoding="utf-8") as f:
+        f.write(model_contents)
+
+    dbt_project_config = {
+        "name": "test_project",
+        "version": "1.0.0",
+        "config-version": 2,
+        "profile": "test",
+        "model-paths": ["models"],
+    }
+    dbt_project_file = dbt_project_dir / "dbt_project.yml"
+    with open(dbt_project_file, "w", encoding="utf-8") as f:
+        yaml.dump(dbt_project_config, f)
+
+    sqlmesh_config = {
+        "model_defaults": {
+            "start": "2025-01-01",
+        }
+    }
+    sqlmesh_config_file = dbt_project_dir / "sqlmesh.yaml"
+    with open(sqlmesh_config_file, "w", encoding="utf-8") as f:
+        yaml.dump(sqlmesh_config, f)
+
+    dbt_data_dir = tmp_path / "dbt_data"
+    dbt_data_dir.mkdir()
+    dbt_data_file = dbt_data_dir / "local.db"
+    dbt_profile_config = {
+        "test": {
+            "outputs": {"duckdb": {"type": "duckdb", "path": str(dbt_data_file)}},
+            "target": "duckdb",
+        }
+    }
+    db_profile_file = dbt_project_dir / "profiles.yml"
+    with open(db_profile_file, "w", encoding="utf-8") as f:
+        yaml.dump(dbt_profile_config, f)
+
+    context = Context(paths=dbt_project_dir)
+
+    # find the model by its sqlmesh fully qualified name
+    model_fqn = '"local"."main"."simple_model"'
+    assert model_fqn in context.snapshots
+
+    # Verify that node_name is the equivalent dbt one
+    model = context.snapshots[model_fqn].model
+    assert model.node_name == "model.test_project.simple_model"

tests/dbt/test_transformation.py

@@ -5,8 +5,6 @@
 import typing as t
 from pathlib import Path
 from unittest.mock import patch
-from sqlmesh.dbt.adapter import RuntimeAdapter
-from sqlmesh.utils.jinja import JinjaMacroRegistry
 
 from sqlmesh.dbt.util import DBT_VERSION
 
@@ -2417,7 +2415,7 @@ def test_selected_resources_context_variable(
 ):
     context = sushi_test_project.context
 
-    # should be empty list during parse time
+    # empty selected resources
     direct_access = context.render("{{ selected_resources }}")
     assert direct_access == "[]"
 
@@ -2443,32 +2441,16 @@ def test_selected_resources_context_variable(
     result = context.render(test_condition)
     assert result.strip() == "no_resources"
 
-    # Test 4: Test with runtime adapter (simulating runtime execution)
-    runtime_adapter = RuntimeAdapter(
-        engine_adapter=sushi_test_dbt_context.engine_adapter,
-        jinja_macros=JinjaMacroRegistry(),
-        jinja_globals={
-            "selected_models": {
-                '"jaffle_shop"."main"."customers"',
-                '"jaffle_shop"."main"."orders"',
-                '"jaffle_shop"."main"."items"',
-            },
-        },
-    )
-
-    # it should return correct selected resources in dbt format
-    selected_resources = runtime_adapter.selected_resources
-    assert len(selected_resources) == 3
-    assert "model.jaffle_shop.customers" in selected_resources
-    assert "model.jaffle_shop.orders" in selected_resources
-    assert "model.jaffle_shop.items" in selected_resources
+    #  selected resources in dbt format
+    selected_resources = [
+        "model.jaffle_shop.customers",
+        "model.jaffle_shop.items",
+        "model.jaffle_shop.orders",
+    ]
 
     # check the jinja macros rendering
     result = context.render("{{ selected_resources }}", selected_resources=selected_resources)
-    assert (
-        result
-        == "['model.jaffle_shop.customers', 'model.jaffle_shop.items', 'model.jaffle_shop.orders']"
-    )
+    assert result == selected_resources.__repr__()
 
     result = context.render(test_jinja, selected_resources=selected_resources)
     assert result.strip() == "3"