Fix: Get rid of recursion when calculating skipped nodes (#4538)

Author: izeigerman

sqlmesh/utils/concurrency.py

@@ -105,16 +105,23 @@ def _skip_next_nodes(self, parent: H) -> None:
             self._finished_future.set_result(None)
             return
 
-        skipped_nodes = [node for node, deps in self._unprocessed_nodes.items() if parent in deps]
+        skipped_nodes = {node for node, deps in self._unprocessed_nodes.items() if parent in deps}
 
-        self._skipped_nodes.extend(skipped_nodes)
+        while skipped_nodes:
+            self._skipped_nodes.extend(skipped_nodes)
 
-        for skipped_node in skipped_nodes:
-            self._unprocessed_nodes_num -= 1
-            self._unprocessed_nodes.pop(skipped_node)
+            for skipped_node in skipped_nodes:
+                self._unprocessed_nodes_num -= 1
+                self._unprocessed_nodes.pop(skipped_node)
+
+            skipped_nodes = {
+                node
+                for node, deps in self._unprocessed_nodes.items()
+                if skipped_nodes.intersection(deps)
+            }
 
-        for skipped_node in skipped_nodes:
-            self._skip_next_nodes(skipped_node)
+        if not self._unprocessed_nodes_num:
+            self._finished_future.set_result(None)
 
     def _init_state(self) -> None:
         self._unprocessed_nodes = self.dag.graph

tests/utils/test_concurrency.py

@@ -139,7 +139,7 @@ def raise_(snapshot):
     assert len(errors) == 1
     assert errors[0].node == failed_snapshot.snapshot_id
 
-    assert skipped == [snapshot_a.snapshot_id, snapshot_b.snapshot_id, snapshot_c.snapshot_id]
+    assert set(skipped) == {snapshot_a.snapshot_id, snapshot_b.snapshot_id, snapshot_c.snapshot_id}
 
 
 @pytest.mark.parametrize("tasks_num", [1, 3])