remove dbt implementation

Author: eakmanrq

sqlmesh/dbt/basemodel.py

@@ -30,7 +30,6 @@
 from sqlmesh.dbt.test import TestConfig
 from sqlmesh.dbt.util import DBT_VERSION
 from sqlmesh.utils import AttributeDict
-from sqlmesh.utils.dag import find_path_with_dfs
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import field_validator
 
@@ -271,10 +270,9 @@ def remove_tests_with_invalid_refs(self, context: DbtContext) -> None:
 
     def fix_circular_test_refs(self, context: DbtContext) -> None:
         """
-        Checks for circular references between models and moves tests to break cycles.
-        This handles both direct circular references (A -> B -> A) and indirect circular
-        references (A -> B -> C -> A). Tests are moved to the model that appears latest
-        in the dependency chain to ensure the cycle is broken.
+        Checks for direct circular references between two models and moves the test to the downstream
+        model if found. This addresses the most common circular reference - relationship tests in both
+        directions. In the future, we may want to increase coverage by checking for indirect circular references.
 
         Args:
             context: The dbt context this model resides within.
@@ -286,91 +284,16 @@ def fix_circular_test_refs(self, context: DbtContext) -> None:
             for ref in test.dependencies.refs:
                 if ref == self.name or ref in self.dependencies.refs:
                     continue
-
-                # Check if moving this test would create or maintain a cycle
-                cycle_path = self._find_circular_path(ref, context, set())
-                if cycle_path:
-                    # Find the model in the cycle that should receive the test
-                    # We want to move to the model that appears latest in the dependency chain
-                    target_model_name = self._select_target_model_for_test(cycle_path, context)
-                    target_model = context.refs[target_model_name]
-
+                model = context.refs[ref]
+                if (
+                    self.name in model.dependencies.refs
+                    or self.name in model.tests_ref_source_dependencies.refs
+                ):
                     logger.info(
-                        f"Moving test '{test.name}' from model '{self.name}' to '{target_model_name}' "
-                        f"to avoid circular reference through path: {' -> '.join(cycle_path)}"
+                        f"Moving test '{test.name}' from model '{self.name}' to '{model.name}' to avoid circular reference."
                     )
-                    target_model.tests.append(test)
+                    model.tests.append(test)
                     self.tests.remove(test)
-                    break
-
-    def _find_circular_path(
-        self, ref: str, context: DbtContext, visited: t.Set[str]
-    ) -> t.Optional[t.List[str]]:
-        """
-        Find if there's a circular dependency path from ref back to this model.
-
-        Args:
-            ref: The model name to start searching from
-            context: The dbt context
-            visited: Set of model names already visited in this path
-
-        Returns:
-            List of model names forming the circular path, or None if no cycle exists
-        """
-        # Build a graph of all models and their dependencies from the context
-        graph: t.Dict[str, t.Set[str]] = {}
-
-        def build_graph_from_node(node_name: str, current_visited: t.Set[str]) -> None:
-            if node_name in current_visited or node_name in graph:
-                return
-            current_visited.add(node_name)
-
-            model = context.refs[node_name]
-            # Include both direct model dependencies and test dependencies
-            all_refs = model.dependencies.refs | model.tests_ref_source_dependencies.refs
-            graph[node_name] = all_refs.copy()
-
-            # Recursively build graph for dependencies
-            for dep in all_refs:
-                build_graph_from_node(dep, current_visited)
-
-        # Build the graph starting from the ref, including visited nodes to avoid infinite recursion
-        build_graph_from_node(ref, visited.copy())
-
-        # Add self.name to the graph if it's not already there
-        if self.name not in graph:
-            graph[self.name] = set()
-
-        # Use the shared DFS function to find path from ref to self.name
-        return find_path_with_dfs(graph, start_node=ref, target_node=self.name)
-
-    def _select_target_model_for_test(self, cycle_path: t.List[str], context: DbtContext) -> str:
-        """
-        Select which model in the cycle should receive the test.
-        We select the model that has the most downstream dependencies in the cycle
-
-        Args:
-            cycle_path: List of model names in the circular dependency path
-            context: The dbt context
-
-        Returns:
-            Name of the model that should receive the test
-        """
-        # Count how many other models in the cycle each model depends on
-        dependency_counts = {}
-
-        for model_name in cycle_path:
-            model = context.refs[model_name]
-            all_refs = model.dependencies.refs | model.tests_ref_source_dependencies.refs
-            count = len([ref for ref in all_refs if ref in cycle_path])
-            dependency_counts[model_name] = count
-
-        # Return the model with the fewest dependencies within the cycle
-        # (i.e., the most downstream model in the cycle)
-        if dependency_counts:
-            return min(dependency_counts, key=dependency_counts.get)  # type: ignore
-        # Fallback to the last model in the path
-        return cycle_path[-1]
 
     @property
     def sqlmesh_config_fields(self) -> t.Set[str]:

sqlmesh/utils/dag.py

@@ -184,7 +184,46 @@ def _find_cycle_path(self, nodes_in_cycle: t.Dict[T, t.Set[T]]) -> t.Optional[t.
         Returns:
             List of nodes forming the cycle path, or None if no cycle found
         """
-        return find_path_with_dfs(nodes_in_cycle)
+        if not nodes_in_cycle:
+            return None
+
+        # Use DFS to find a cycle path
+        visited: t.Set[T] = set()
+        rec_stack: t.Set[T] = set()
+        path: t.List[T] = []
+
+        def dfs(node: T) -> t.Optional[t.List[T]]:
+            if node in rec_stack:
+                # Found a cycle - extract the cycle path
+                cycle_start = path.index(node)
+                return path[cycle_start:] + [node]
+
+            if node in visited:
+                return None
+
+            visited.add(node)
+            rec_stack.add(node)
+            path.append(node)
+
+            # Only follow edges to nodes that are still in the unprocessed set
+            for neighbor in nodes_in_cycle.get(node, set()):
+                if neighbor in nodes_in_cycle:
+                    cycle = dfs(neighbor)
+                    if cycle:
+                        return cycle
+
+            rec_stack.remove(node)
+            path.pop()
+            return None
+
+        # Try starting DFS from each unvisited node
+        for start_node in nodes_in_cycle:
+            if start_node not in visited:
+                cycle = dfs(start_node)
+                if cycle:
+                    return cycle[:-1]  # Remove the duplicate node at the end
+
+        return None
 
     @property
     def roots(self) -> t.Set[T]:

tests/dbt/test_model.py

@@ -116,126 +116,6 @@ def test_model_test_circular_references() -> None:
     assert downstream_model.tests == [downstream_test, upstream_test]
 
 
-def test_model_test_indirect_circular_references() -> None:
-    """Test detection and resolution of indirect circular references through test dependencies"""
-    model_a = ModelConfig(name="model_a")  # No dependencies
-    model_b = ModelConfig(
-        name="model_b", dependencies=Dependencies(refs={"model_a"})
-    )  # B depends on A
-    model_c = ModelConfig(
-        name="model_c", dependencies=Dependencies(refs={"model_b"})
-    )  # C depends on B
-
-    context = DbtContext(_refs={"model_a": model_a, "model_b": model_b, "model_c": model_c})
-
-    # Test on model_a that references model_c (creates indirect cycle through test dependencies)
-    # The cycle would be: model_a (via test) -> model_c -> model_b -> model_a
-    test_a_refs_c = TestConfig(
-        name="test_a_refs_c",
-        sql="",
-        dependencies=Dependencies(refs={"model_a", "model_c"}),  # Test references both A and C
-    )
-
-    # Place tests that would create indirect cycles when combined with model dependencies
-    model_a.tests = [test_a_refs_c]
-    assert model_b.tests == []
-    assert model_c.tests == []
-
-    # Fix circular references on model_a
-    model_a.fix_circular_test_refs(context)
-    # The test should be moved from model_a to break the indirect cycle down to model c
-    assert model_a.tests == []
-    assert test_a_refs_c in model_c.tests
-
-
-def test_model_test_complex_indirect_circular_references() -> None:
-    """Test detection and resolution of more complex indirect circular references through test dependencies"""
-    # Create models with a longer linear dependency chain (no cycles in models themselves)
-    # A -> B -> C -> D (B depends on A, C depends on B, D depends on C)
-    model_a = ModelConfig(name="model_a")  # No dependencies
-    model_b = ModelConfig(
-        name="model_b", dependencies=Dependencies(refs={"model_a"})
-    )  # B depends on A
-    model_c = ModelConfig(
-        name="model_c", dependencies=Dependencies(refs={"model_b"})
-    )  # C depends on B
-    model_d = ModelConfig(
-        name="model_d", dependencies=Dependencies(refs={"model_c"})
-    )  # D depends on C
-
-    context = DbtContext(
-        _refs={"model_a": model_a, "model_b": model_b, "model_c": model_c, "model_d": model_d}
-    )
-
-    # Test on model_a that references model_d (creates long indirect cycle through test dependencies)
-    # The cycle would be: model_a (via test) -> model_d -> model_c -> model_b -> model_a
-    test_a_refs_d = TestConfig(
-        name="test_a_refs_d",
-        sql="",
-        dependencies=Dependencies(refs={"model_a", "model_d"}),  # Test references both A and D
-    )
-
-    # Place tests that would create indirect cycles when combined with model dependencies
-    model_a.tests = [test_a_refs_d]
-    model_b.tests = []
-    assert model_c.tests == []
-    assert model_d.tests == []
-
-    # Fix circular references on model_a
-    model_a.fix_circular_test_refs(context)
-    # The test should be moved from model_a to break the long indirect cycle down to model_d
-    assert model_a.tests == []
-    assert model_d.tests == [test_a_refs_d]
-
-    # Test on model_b that references model_d (creates indirect cycle through test dependencies)
-    # The cycle would be: model_b (via test) -> model_d -> model_c -> model_b
-    test_b_refs_d = TestConfig(
-        name="test_b_refs_d",
-        sql="",
-        dependencies=Dependencies(refs={"model_b", "model_d"}),  # Test references both B and D
-    )
-    model_a.tests = []
-    model_b.tests = [test_b_refs_d]
-    model_c.tests = []
-    model_d.tests = []
-
-    model_b.fix_circular_test_refs(context)
-    assert model_a.tests == []
-    assert model_b.tests == []
-    assert model_c.tests == []
-    assert model_d.tests == [test_b_refs_d]
-
-    # Do both at the same time
-    model_a.tests = [test_a_refs_d]
-    model_b.tests = [test_b_refs_d]
-    model_c.tests = []
-    model_d.tests = []
-
-    model_a.fix_circular_test_refs(context)
-    model_b.fix_circular_test_refs(context)
-    assert model_a.tests == []
-    assert model_b.tests == []
-    assert model_c.tests == []
-    assert model_d.tests == [test_a_refs_d, test_b_refs_d]
-
-    # Test A -> B -> C cycle and make sure test ends up with C
-    test_a_refs_c = TestConfig(
-        name="test_a_refs_c",
-        sql="",
-        dependencies=Dependencies(refs={"model_a", "model_c"}),  # Test references both A and C
-    )
-    model_a.tests = [test_a_refs_c]
-    model_b.tests = []
-    model_c.tests = []
-    model_d.tests = []
-
-    model_a.fix_circular_test_refs(context)
-    assert model_a.tests == []
-    assert model_b.tests == []
-    assert model_c.tests == [test_a_refs_c]
-    assert model_d.tests == []
-
-
 @pytest.mark.slow
 def test_load_invalid_ref_audit_constraints(
     tmp_path: Path, caplog, dbt_dummy_postgres_config: PostgresConfig, create_empty_project