Repository: TobikoData/sqlmesh
Pull request #2022 (2024-01-25 03:14:17)

fix: further improve performance of dag by introducing a dummy node
nodes are now connected to a terminal node to avoid many to many edges

before:

![Screenshot_2024-01-24_19 15 44](https://github.com/TobikoData/sqlmesh/assets/8205034/762992e3-d742-47fd-ac76-60fc48a81ca0)

after:

![Screenshot_2024-01-24_19 14 38](https://github.com/TobikoData/sqlmesh/assets/8205034/c7357e02-1481-451d-8355-58e3dc17e8b9)

104 edges -> 60 edges

izeigerman: Do we need a dummy node for snapshots that only have a single interval to backfill?
tobymao: reduced to 58 edges

![Screenshot_2024-01-25_16 50 42](https://github.com/TobikoData/sqlmesh/assets/8205034/ed2f434f-d380-41a9-8469-56b751e0e019)


Patch:
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index b7c601fa..36420fcf 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -265,6 +265,8 @@ class Scheduler:
                 raise CircuitBreakerError()
 
             snapshot_name, ((start, end), batch_idx) = node
+            if batch_idx == -1:
+                return
             snapshot = snapshots_by_name[snapshot_name]
 
             self.console.start_snapshot_evaluation_progress(snapshot)
@@ -323,35 +325,38 @@ class Scheduler:
             A DAG of snapshot intervals to be evaluated.
         """
 
-        intervals_per_snapshot_version = {
-            (snapshot.name, snapshot.version_get_or_generate()): intervals
-            for snapshot, intervals in batches.items()
+        intervals_per_snapshot = {
+            snapshot.name: intervals for snapshot, intervals in batches.items()
         }
 
         dag = DAG[SchedulingUnit]()
+        terminal_node = ((to_datetime(0), to_datetime(0)), -1)
+
         for snapshot, intervals in batches.items():
             if not intervals:
                 continue
-            upstream_dependencies = [
-                (p_sid.name, (interval, i))
-                for p_sid in snapshot.parents
-                if p_sid in self.snapshots
-                for i, interval in enumerate(
-                    intervals_per_snapshot_version.get(
-                        (
-                            self.snapshots[p_sid].name,
-                            self.snapshots[p_sid].version_get_or_generate(),
-                        ),
-                        [],
-                    )
-                )
-            ]
+
+            upstream_dependencies = []
+
+            for p_sid in snapshot.parents:
+                if p_sid in self.snapshots:
+                    p_intervals = intervals_per_snapshot.get(p_sid.name, [])
+
+                    if len(p_intervals) > 1:
+                        upstream_dependencies.append((p_sid.name, terminal_node))
+                    else:
+                        for (i, interval) in enumerate(p_intervals):
+                            upstream_dependencies.append((p_sid.name, (interval, i)))
+
             for i, interval in enumerate(intervals):
-                dag.add((snapshot.name, (interval, i)), upstream_dependencies)
+                node = (snapshot.name, (interval, i))
+                dag.add(node, upstream_dependencies)
+
+                if len(intervals) > 1:
+                    dag.add((snapshot.name, terminal_node), [node])
+
                 if snapshot.depends_on_past and i > 0:
-                    dag.add(
-                        (snapshot.name, (interval, i)), [(snapshot.name, (intervals[i - 1], i - 1))]
-                    )
+                    dag.add(node, [(snapshot.name, (intervals[i - 1], i - 1))])
         return dag
 
 


Test patch:
diff --git a/tests/core/test_scheduler.py b/tests/core/test_scheduler.py
index 3635a0c0..47d40e93 100644
--- a/tests/core/test_scheduler.py
+++ b/tests/core/test_scheduler.py
@@ -127,7 +127,6 @@ def test_incremental_by_unique_key_kind_dag(mocker: MockerFixture, make_snapshot
     batches = scheduler.batches(start, end, end)
     dag = scheduler._dag(batches)
     assert dag.graph == {
-        # Depends on no one
         (
             unique_by_key_snapshot.name,
             ((to_datetime("2023-01-01"), to_datetime("2023-01-07")), 0),
@@ -200,6 +199,26 @@ def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapsho
                 ((to_datetime("2023-01-04"), to_datetime("2023-01-05")), 2),
             ),
         },
+        (incremental_self_snapshot.name, ((to_datetime(0), to_datetime(0)), -1),): set(
+            [
+                (
+                    incremental_self_snapshot.name,
+                    ((to_datetime("2023-01-01"), to_datetime("2023-01-02")), 0),
+                ),
+                (
+                    incremental_self_snapshot.name,
+                    ((to_datetime("2023-01-03"), to_datetime("2023-01-04")), 1),
+                ),
+                (
+                    incremental_self_snapshot.name,
+                    ((to_datetime("2023-01-04"), to_datetime("2023-01-05")), 2),
+                ),
+                (
+                    incremental_self_snapshot.name,
+                    ((to_datetime("2023-01-06"), to_datetime("2023-01-07")), 3),
+                ),
+            ]
+        ),
     }