Fix: Add workaround to ignore source dependency if fqn matches model

sqlmesh/dbt/basemodel.py

@@ -130,7 +130,7 @@ class BaseModelConfig(GeneralConfig):
     unique_id: str = ""
     name: str = ""
     package_name: str = ""
-    fqn: t.List[str] = []
+    fqn_: t.List[str] = Field(default_factory=list, alias="fqn")
     schema_: str = Field("", alias="schema")
     database: t.Optional[str] = None
     alias: t.Optional[str] = None
@@ -281,15 +281,17 @@ def remove_tests_with_invalid_refs(self, context: DbtContext) -> None:
             and all(source in context.sources for source in test.dependencies.sources)
         ]
 
+    @property
+    def fqn(self) -> str:
+        return ".".join(self.fqn_)
+
     @property
     def sqlmesh_config_fields(self) -> t.Set[str]:
         return {"description", "owner", "stamp", "storage_format"}
 
     @property
     def node_info(self) -> DbtNodeInfo:
-        return DbtNodeInfo(
-            unique_id=self.unique_id, name=self.name, fqn=".".join(self.fqn), alias=self.alias
-        )
+        return DbtNodeInfo(unique_id=self.unique_id, name=self.name, fqn=self.fqn, alias=self.alias)
 
     def sqlmesh_model_kwargs(
         self,
@@ -327,7 +329,14 @@ def sqlmesh_model_kwargs(
             "column_descriptions": column_descriptions_to_sqlmesh(self.columns) or None,
             "depends_on": {
                 model.canonical_name(context) for model in model_context.refs.values()
-            }.union({source.canonical_name(context) for source in model_context.sources.values()}),
+            }.union(
+                {
+                    source.canonical_name(context)
+                    for source in model_context.sources.values()
+                    if source.fqn not in context.model_fqns
+                    # Allow dbt projects to reference a model as a source without causing a cycle
+                },
+            ),
             "jinja_macros": jinja_macros,
             "path": self.path,
             "pre_statements": [d.jinja_statement(hook.sql) for hook in self.pre_hook],

sqlmesh/dbt/context.py

@@ -51,6 +51,7 @@ class DbtContext:
     _project_name: t.Optional[str] = None
     _variables: t.Dict[str, t.Any] = field(default_factory=dict)
     _models: t.Dict[str, ModelConfig] = field(default_factory=dict)
+    _model_fqns: t.Set[str] = field(default_factory=set)
     _seeds: t.Dict[str, SeedConfig] = field(default_factory=dict)
     _sources: t.Dict[str, SourceConfig] = field(default_factory=dict)
     _refs: t.Dict[str, t.Union[ModelConfig, SeedConfig]] = field(default_factory=dict)
@@ -144,13 +145,20 @@ def models(self) -> t.Dict[str, ModelConfig]:
     def models(self, models: t.Dict[str, ModelConfig]) -> None:
         self._models = {}
         self._refs = {}
+        self._model_fqns = set()
         self.add_models(models)
 
     def add_models(self, models: t.Dict[str, ModelConfig]) -> None:
         self._refs = {}
         self._models.update(models)
         self._jinja_environment = None
 
+    @property
+    def model_fqns(self) -> t.Set[str]:
+        if not self._model_fqns:
+            self._model_fqns = {model.fqn for model in self._models.values()}
+        return self._model_fqns
+
     @property
     def seeds(self) -> t.Dict[str, SeedConfig]:
         return self._seeds

sqlmesh/dbt/source.py

@@ -36,6 +36,7 @@ class SourceConfig(GeneralConfig):
     # DBT configuration fields
     name: str = ""
     source_name_: str = Field("", alias="source_name")
+    fqn_: t.List[str] = Field(default_factory=list, alias="fqn")
     database: t.Optional[str] = None
     schema_: t.Optional[str] = Field(None, alias="schema")
     identifier: t.Optional[str] = None
@@ -64,6 +65,10 @@ def table_name(self) -> t.Optional[str]:
     def config_name(self) -> str:
         return f"{self.source_name_}.{self.name}"
 
+    @property
+    def fqn(self) -> str:
+        return ".".join(self.fqn_)
+
     def canonical_name(self, context: DbtContext) -> str:
         if self._canonical_name is None:
             source = context.get_callable_macro("source")

tests/dbt/test_transformation.py

@@ -53,6 +53,7 @@
 )
 from sqlmesh.dbt.context import DbtContext
 from sqlmesh.dbt.model import Materialization, ModelConfig
+from sqlmesh.dbt.source import SourceConfig
 from sqlmesh.dbt.project import Project
 from sqlmesh.dbt.relation import Policy
 from sqlmesh.dbt.seed import SeedConfig
@@ -2678,3 +2679,31 @@ def test_selected_resources_context_variable(
 
     result = context.render(test_condition, selected_resources=selected_resources)
     assert result.strip() == "has_resources"
+
+
+def test_ignore_source_depends_on_when_also_model(dbt_dummy_postgres_config: PostgresConfig):
+    context = DbtContext()
+    context._target = dbt_dummy_postgres_config
+
+    source_a = SourceConfig(
+        name="source_a",
+        fqn=["package", "schema", "model_a"],
+    )
+    source_a._canonical_name = "schema.source_a"
+    source_b = SourceConfig(
+        name="source_b",
+        fqn=["package", "schema", "source_b"],
+    )
+    source_b._canonical_name = "schema.source_b"
+    context.sources = {"source_a": source_a, "source_b": source_b}
+
+    model = ModelConfig(
+        dependencies=Dependencies(sources={"source_a", "source_b"}),
+        fqn=["package", "schema", "test_model"],
+    )
+    context.models = {
+        "test_model": model,
+        "model_a": ModelConfig(name="model_a", fqn=["package", "schema", "model_a"]),
+    }
+
+    assert model.sqlmesh_model_kwargs(context)["depends_on"] == {"schema.source_b"}