Repository: TobikoData/sqlmesh
Pull request #2874 (2024-07-05 21:26:54)

Feat: Support dbt model versioning
Introduces support for the dbt model versioning feature as described here: https://docs.getdbt.com/docs/collaborate/govern/model-versions

izeigerman: @tobymao for all intents and purposes each version is treated as a separate model. 

This is also consistent with dbt docs:
> How is this different from just creating a new model?
Honestly, it's only a little bit different! There isn't much magic here, and that's by design.
sungchun12: @tobymao it creates a whole new physical table specific to that version. 


```shell
dbt.exceptions.AmbiguousAliasError: Compilation Error
  dbt found two resources with the database representation "database.schema.model_name".
  dbt cannot create two resources with identical database representations. To fix this,
  change the configuration of one of these resources:
  - model.project_name.model_name.v1 (models/.../model_name.sql)
  - model.project_name.model_name.v2 (models/.../model_name_v2.sql)
```

Patch:
diff --git a/examples/sushi_dbt/models/schema.yml b/examples/sushi_dbt/models/schema.yml
index 3bb0c047..eb5d288d 100644
--- a/examples/sushi_dbt/models/schema.yml
+++ b/examples/sushi_dbt/models/schema.yml
@@ -77,3 +77,6 @@ models:
               amount: 10000
       - name: ds
         description: Date
+    versions:
+      - v: 1
+      - v: 2
diff --git a/examples/sushi_dbt/models/top_waiters.sql b/examples/sushi_dbt/models/top_waiters.sql
index b08eca97..f839b31d 100644
--- a/examples/sushi_dbt/models/top_waiters.sql
+++ b/examples/sushi_dbt/models/top_waiters.sql
@@ -7,12 +7,12 @@
 SELECT
   waiter_id::INT AS waiter_id,
   revenue::DOUBLE AS revenue
-FROM {{ ref('waiter_revenue_by_day') }}
+FROM {{ ref('waiter_revenue_by_day', version=1) }}
 WHERE
   ds = (
     SELECT
       MAX(ds)
-    FROM {{ ref('waiter_revenue_by_day') }}
+    FROM {{ ref('waiter_revenue_by_day', v=1) }}
   )
 ORDER BY
   revenue DESC
diff --git a/examples/sushi_dbt/models/waiter_revenue_by_day_v1.sql b/examples/sushi_dbt/models/waiter_revenue_by_day_v1.sql
new file mode 100644
index 00000000..d430c612
--- /dev/null
+++ b/examples/sushi_dbt/models/waiter_revenue_by_day_v1.sql
@@ -0,0 +1,29 @@
+{{
+  config(
+    materialized='incremental',
+    incremental_strategy='delete+insert',
+    cluster_by=['ds'],
+    time_column='ds',
+  )
+}}
+
+SELECT
+  o.waiter_id::INT AS waiter_id, /* Waiter id */
+  SUM(oi.quantity * i.price)::DOUBLE AS revenue, /* Revenue from orders taken by this waiter */
+  o.ds::TEXT AS ds /* Date */
+FROM {{ ref('orders') }} AS o
+LEFT JOIN {{ ref('order_items') }} AS oi
+  ON o.id = oi.order_id AND o.ds = oi.ds
+LEFT JOIN {{ ref('items') }} AS i
+  ON oi.item_id = i.id AND oi.ds = i.ds
+{% if is_incremental() %}
+  WHERE
+    o.ds > (select max(ds) from {{ this }})
+{% endif %}
+{% if sqlmesh_incremental is defined %}
+  WHERE
+      o.ds BETWEEN '{{ start_ds }}' AND '{{ end_ds }}'
+{% endif %}
+GROUP BY
+  o.waiter_id,
+  o.ds
diff --git a/sqlmesh/core/config/scheduler.py b/sqlmesh/core/config/scheduler.py
index 093bdd4d..14159f65 100644
--- a/sqlmesh/core/config/scheduler.py
+++ b/sqlmesh/core/config/scheduler.py
@@ -136,6 +136,7 @@ class BuiltInSchedulerConfig(_EngineAdapterStateSyncSchedulerConfig, BaseConfig)
             backfill_concurrent_tasks=context.concurrent_tasks,
             console=context.console,
             notification_target_manager=context.notification_target_manager,
+            signal_factory=context._signal_factory,
         )
 
     def get_default_catalog(self, context: GenericContext) -> t.Optional[str]:
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index b66e9a5d..5c0fd8c6 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -80,7 +80,7 @@ from sqlmesh.core.notification_target import (
 )
 from sqlmesh.core.plan import Plan, PlanBuilder
 from sqlmesh.core.reference import ReferenceGraph
-from sqlmesh.core.scheduler import Scheduler
+from sqlmesh.core.scheduler import Scheduler, SignalFactory
 from sqlmesh.core.schema_loader import create_external_models_file
 from sqlmesh.core.selector import Selector
 from sqlmesh.core.snapshot import (
@@ -308,6 +308,7 @@ class GenericContext(BaseContext, t.Generic[C]):
         load: bool = True,
         console: t.Optional[Console] = None,
         users: t.Optional[t.List[User]] = None,
+        signal_factory: t.Optional[SignalFactory] = None,
     ):
         self.configs = (
             config if isinstance(config, dict) else load_configs(config, self.CONFIG_TYPE, paths)
@@ -354,6 +355,7 @@ class GenericContext(BaseContext, t.Generic[C]):
         )
 
         self._snapshot_evaluator: t.Optional[SnapshotEvaluator] = None
+        self._signal_factory = signal_factory
 
         self._provided_state_sync: t.Optional[StateSync] = state_sync
         self._state_sync: t.Optional[StateSync] = None
@@ -474,6 +476,7 @@ class GenericContext(BaseContext, t.Generic[C]):
             max_workers=self.concurrent_tasks,
             console=self.console,
             notification_target_manager=self.notification_target_manager,
+            signal_factory=self._signal_factory,
         )
 
     @property
diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py
index 0e5e98e0..5d408154 100644
--- a/sqlmesh/core/plan/evaluator.py
+++ b/sqlmesh/core/plan/evaluator.py
@@ -26,7 +26,7 @@ from sqlmesh.core.notification_target import (
     NotificationTargetManager,
 )
 from sqlmesh.core.plan.definition import Plan
-from sqlmesh.core.scheduler import Scheduler
+from sqlmesh.core.scheduler import Scheduler, SignalFactory
 from sqlmesh.core.snapshot import DeployabilityIndex, Snapshot, SnapshotEvaluator
 from sqlmesh.core.state_sync import StateSync
 from sqlmesh.core.state_sync.base import PromotionResult
@@ -65,6 +65,7 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         backfill_concurrent_tasks: int = 1,
         console: t.Optional[Console] = None,
         notification_target_manager: t.Optional[NotificationTargetManager] = None,
+        signal_factory: t.Optional[SignalFactory] = None,
     ):
         self.state_sync = state_sync
         self.snapshot_evaluator = snapshot_evaluator
@@ -72,6 +73,7 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         self.backfill_concurrent_tasks = backfill_concurrent_tasks
         self.console = console or get_console()
         self.notification_target_manager = notification_target_manager
+        self.signal_factory = signal_factory
 
     def evaluate(
         self,
@@ -159,6 +161,7 @@ class BuiltInPlanEvaluator(PlanEvaluator):
             max_workers=self.backfill_concurrent_tasks,
             console=self.console,
             notification_target_manager=self.notification_target_manager,
+            signal_factory=self.signal_factory,
         )
         is_run_successful = scheduler.run(
             plan.environment_naming_info,
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index f0ac453a..6364b842 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import abc
 import logging
 import traceback
 import typing as t
@@ -30,6 +31,7 @@ from sqlmesh.utils.date import (
     now,
     now_timestamp,
     to_datetime,
+    to_timestamp,
     validate_date_range,
 )
 from sqlmesh.utils.errors import AuditError, CircuitBreakerError, SQLMeshError
@@ -43,6 +45,39 @@ SnapshotToBatches = t.Dict[Snapshot, Batch]
 SchedulingUnit = t.Tuple[str, t.Tuple[Interval, int]]
 
 
+class Signal(abc.ABC):
+    @abc.abstractmethod
+    def check_intervals(self, batch: Batch) -> t.Union[bool, Batch]:
+        """Returns which intervals are ready from a list of scheduled intervals.
+
+        When SQLMesh wishes to execute a batch of intervals, say between `a` and `d`, then
+        the `batch` parameter will contain each individual interval within this batch,
+        i.e.: `[a,b),[b,c),[c,d)`.
+
+        This function may return `True` to indicate that the whole batch is ready,
+        `False` to indicate none of the batch's intervals are ready, or a list of
+        intervals (a batch) to indicate exactly which ones are ready.
+
+        When returning a batch, the function is expected to return a subset of
+        the `batch` parameter, e.g.: `[a,b),[b,c)`. Note that it may return
+        gaps, e.g.: `[a,b),[c,d)`, but it may not alter the bounds of any of the
+        intervals.
+
+        The interface allows an implementation to check batches of intervals without
+        having to actually compute individual intervals itself.
+
+        Args:
+            batch: the list of intervals that are missing and scheduled to run.
+
+        Returns:
+            Either `True` to indicate all intervals are ready, `False` to indicate none are
+            ready or a list of intervals to indicate exactly which ones are ready.
+        """
+
+
+SignalFactory = t.Callable[[t.Dict[str, t.Union[str, int, float, bool]]], Signal]
+
+
 class Scheduler:
     """Schedules and manages the evaluation of snapshots.
 
@@ -58,6 +93,7 @@ class Scheduler:
         state_sync: The state sync to pull saved snapshots.
         max_workers: The maximum number of parallel queries to run.
         console: The rich instance used for printing scheduling information.
+        signal_factory: A factory method for building Signal instances from model signal configuration.
     """
 
     def __init__(
@@ -69,6 +105,7 @@ class Scheduler:
         max_workers: int = 1,
         console: t.Optional[Console] = None,
         notification_target_manager: t.Optional[NotificationTargetManager] = None,
+        signal_factory: t.Optional[SignalFactory] = None,
     ):
         self.state_sync = state_sync
         self.snapshots = {s.snapshot_id: s for s in snapshots}
@@ -80,6 +117,7 @@ class Scheduler:
         self.notification_target_manager = (
             notification_target_manager or NotificationTargetManager()
         )
+        self.signal_factory = signal_factory
 
     def batches(
         self,
@@ -131,6 +169,7 @@ class Scheduler:
             restatements=restatements,
             ignore_cron=ignore_cron,
             end_bounded=end_bounded,
+            signal_factory=self.signal_factory,
         )
 
     def evaluate(
@@ -393,6 +432,7 @@ def compute_interval_params(
     restatements: t.Optional[t.Dict[SnapshotId, SnapshotInterval]] = None,
     ignore_cron: bool = False,
     end_bounded: bool = False,
+    signal_factory: t.Optional[SignalFactory] = None,
 ) -> SnapshotToBatches:
     """Find the optimal date interval paramaters based on what needs processing and maximal batch size.
 
@@ -431,6 +471,15 @@ def compute_interval_params(
         ignore_cron=ignore_cron,
         end_bounded=end_bounded,
     ).items():
+        if signal_factory:
+            for signal in snapshot.model.render_signals(
+                start=start, end=end, execution_time=execution_time
+            ):
+                intervals = _check_ready_intervals(
+                    signal=signal_factory(signal),
+                    intervals=intervals,
+                )
+
         batches = []
         batch_size = snapshot.node.batch_size
         next_batch: t.List[t.Tuple[int, int]] = []
@@ -465,3 +514,53 @@ def _resolve_one_snapshot_per_version(
                 snapshot_per_version[key] = snapshot
 
     return snapshot_per_version
+
+
+def _contiguous_intervals(
+    intervals: t.List[SnapshotInterval],
+) -> t.List[t.List[SnapshotInterval]]:
+    """Given a list of intervals with gaps, returns a list of sequences of contiguous intervals."""
+    contiguous_intervals = []
+    current_batch: t.List[SnapshotInterval] = []
+    for interval in intervals:
+        if len(current_batch) == 0 or interval[0] == current_batch[-1][-1]:
+            current_batch.append(interval)
+        else:
+            contiguous_intervals.append(current_batch)
+            current_batch = [interval]
+
+    if len(current_batch) > 0:
+        contiguous_intervals.append(current_batch)
+
+    return contiguous_intervals
+
+
+def _check_ready_intervals(
+    signal: Signal,
+    intervals: t.List[SnapshotInterval],
+) -> t.List[SnapshotInterval]:
+    """Returns a list of intervals that are considered ready by the provided signal.
+
+    Note that this will handle gaps in the provided intervals. The returned intervals
+    may introduce new gaps.
+    """
+    checked_intervals = []
+    for interval_batch in _contiguous_intervals(intervals):
+        batch = [(to_datetime(start), to_datetime(end)) for start, end in interval_batch]
+
+        ready_intervals = signal.check_intervals(batch=batch)
+        if isinstance(ready_intervals, bool):
+            if not ready_intervals:
+                batch = []
+        elif isinstance(ready_intervals, list):
+            for i in ready_intervals:
+                if i not in batch:
+                    raise RuntimeError(f"Signal returned unknown interval {i}")
+            batch = ready_intervals
+        else:
+            raise ValueError(
+                f"unexpected return value from signal, expected bool | list, got {type(ready_intervals)}"
+            )
+
+        checked_intervals.extend([(to_timestamp(start), to_timestamp(end)) for start, end in batch])
+    return checked_intervals
diff --git a/sqlmesh/dbt/basemodel.py b/sqlmesh/dbt/basemodel.py
index 0e670531..40a07ea9 100644
--- a/sqlmesh/dbt/basemodel.py
+++ b/sqlmesh/dbt/basemodel.py
@@ -120,6 +120,9 @@ class BaseModelConfig(GeneralConfig):
     columns: t.Dict[str, ColumnConfig] = {}
     quoting: t.Dict[str, t.Optional[bool]] = {}
 
+    version: t.Optional[int] = None
+    latest_version: t.Optional[int] = None
+
     _canonical_name: t.Optional[str] = None
 
     @field_validator("pre_hook", "post_hook", mode="before")
@@ -184,9 +187,9 @@ class BaseModelConfig(GeneralConfig):
     @property
     def config_name(self) -> str:
         """
-        Get the model's config name (package_name.name)
+        Get the model's config name (package_name.alias)
         """
-        return f"{self.package_name}.{self.name}"
+        return f"{self.package_name}.{self.alias}"
 
     def dialect(self, context: DbtContext) -> str:
         return self.dialect_ or context.default_dialect
diff --git a/sqlmesh/dbt/builtin.py b/sqlmesh/dbt/builtin.py
index d6d62db4..00cd897a 100644
--- a/sqlmesh/dbt/builtin.py
+++ b/sqlmesh/dbt/builtin.py
@@ -171,11 +171,33 @@ def log(msg: str, info: bool = False) -> str:
 
 
 def generate_ref(refs: t.Dict[str, t.Any], api: Api) -> t.Callable:
-    def ref(package: str, name: t.Optional[str] = None) -> t.Optional[BaseRelation]:
+    def ref(
+        package: str, name: t.Optional[str] = None, **kwargs: t.Any
+    ) -> t.Optional[BaseRelation]:
+        version = kwargs.get("version", kwargs.get("v"))
         ref_name = f"{package}.{name}" if name else package
-        relation_info = refs.get(ref_name)
+
+        if version is not None:
+            relation_info = refs.get(f"{ref_name}_v{version}")
+            if relation_info is None:
+                logger.warning(
+                    "Could not resolve ref '%s' with version '%s'. Falling back to unversioned reference",
+                    ref_name,
+                    version,
+                )
+                relation_info = refs.get(ref_name)
+        else:
+            relation_info = refs.get(ref_name)
+            if not relation_info:
+                versioned_infos = sorted(
+                    [(r, info) for r, info in refs.items() if r.startswith(f"{ref_name}_v")],
+                    key=lambda i: i[0],
+                )
+                if versioned_infos:
+                    relation_info = versioned_infos[-1][1]
+
         if relation_info is None:
-            logger.debug("Could not resolve ref '%s'", ref_name)
+            logger.debug("Could not resolve ref '%s', version '%s'", ref_name, version)
             return None
 
         return _relation_info_to_relation(relation_info, api.Relation, api.quote_policy)
diff --git a/sqlmesh/dbt/context.py b/sqlmesh/dbt/context.py
index d6a41286..14e14efa 100644
--- a/sqlmesh/dbt/context.py
+++ b/sqlmesh/dbt/context.py
@@ -185,8 +185,11 @@ class DbtContext:
             for model in t.cast(
                 t.Dict[str, t.Union[ModelConfig, SeedConfig]], {**self._seeds, **self._models}
             ).values():
-                self._refs[model.name] = model
+                self._refs[model.alias or model.name] = model
                 self._refs[model.config_name] = model
+                if model.version is not None and model.version == model.latest_version:
+                    self._refs[model.name] = model
+                    self._refs[f"{model.package_name}.{model.name}"] = model
         return self._refs
 
     @property
@@ -270,7 +273,6 @@ class DbtContext:
         dependency_context.seeds = seeds
         dependency_context.models = models
         dependency_context.variables = variables
-        dependency_context._refs = {**dependency_context._seeds, **dependency_context._models}  # type: ignore
 
         return dependency_context
 
diff --git a/sqlmesh/dbt/manifest.py b/sqlmesh/dbt/manifest.py
index 0bac148a..f62dbc38 100644
--- a/sqlmesh/dbt/manifest.py
+++ b/sqlmesh/dbt/manifest.py
@@ -223,14 +223,15 @@ class ManifestHelper:
                     self._flatten_dependencies_from_macros(dependencies.macros, node.package_name)
                 )
 
-                self._models_per_package[node.package_name][node.name] = ModelConfig(
+                # Using the alias instead of the name because the alias captures the version of the model.
+                self._models_per_package[node.package_name][node.alias] = ModelConfig(
                     sql=sql,
                     dependencies=dependencies,
                     tests=tests,
                     **node_config,
                 )
             else:
-                self._seeds_per_package[node.package_name][node.name] = SeedConfig(
+                self._seeds_per_package[node.package_name][node.alias] = SeedConfig(
                     dependencies=Dependencies(macros=macro_references),
                     tests=tests,
                     **node_config,
@@ -451,7 +452,13 @@ def _macro_references(
 
 def _refs(node: ManifestNode) -> t.Set[str]:
     if DBT_VERSION >= (1, 5):
-        return {f"{r.package}.{r.name}" if r.package else r.name for r in node.refs}  # type: ignore
+        result = set()
+        for r in node.refs:
+            ref_name = f"{r.package}.{r.name}" if r.package else r.name
+            if getattr(r, "version", None):
+                ref_name = f"{ref_name}_v{r.version}"
+            result.add(ref_name)
+        return result
     else:
         return {".".join(r) for r in node.refs}  # type: ignore
 
diff --git a/sqlmesh/schedulers/airflow/plan.py b/sqlmesh/schedulers/airflow/plan.py
index 1a2dc553..e5941f54 100644
--- a/sqlmesh/schedulers/airflow/plan.py
+++ b/sqlmesh/schedulers/airflow/plan.py
@@ -134,6 +134,7 @@ def create_plan_dag_spec(
             deployability_index=deployability_index_for_evaluation,
             restatements=restatements,
             end_bounded=request.end_bounded,
+            signal_factory=None,
         )
     else:
         backfill_batches = {}


Test patch:
diff --git a/tests/core/test_scheduler.py b/tests/core/test_scheduler.py
index c68dbd1b..a0ef5a89 100644
--- a/tests/core/test_scheduler.py
+++ b/tests/core/test_scheduler.py
@@ -458,3 +458,91 @@ def test_external_model_audit(mocker, make_snapshot):
     )
 
     spy.assert_called_once()
+
+
+def test_contiguous_intervals():
+    from sqlmesh.core.scheduler import _contiguous_intervals as ci
+
+    assert ci([]) == []
+    assert ci([(0, 1)]) == [[(0, 1)]]
+    assert ci([(0, 1), (1, 2), (2, 3)]) == [[(0, 1), (1, 2), (2, 3)]]
+    assert ci([(0, 1), (3, 4), (4, 5), (6, 7)]) == [
+        [(0, 1)],
+        [(3, 4), (4, 5)],
+        [(6, 7)],
+    ]
+
+
+def test_check_ready_intervals(mocker: MockerFixture):
+    from sqlmesh.core.scheduler import _check_ready_intervals, Interval
+
+    def const_signal(const):
+        signal_mock = mocker.Mock()
+        signal_mock.check_intervals = mocker.MagicMock(return_value=const)
+        return signal_mock
+
+    def assert_always_signal(intervals):
+        _check_ready_intervals(const_signal(True), intervals) == intervals
+
+    assert_always_signal([])
+    assert_always_signal([(0, 1)])
+    assert_always_signal([(0, 1), (1, 2)])
+    assert_always_signal([(0, 1), (2, 3)])
+
+    def assert_never_signal(intervals):
+        _check_ready_intervals(const_signal(False), intervals) == []
+
+    assert_never_signal([])
+    assert_never_signal([(0, 1)])
+    assert_never_signal([(0, 1), (1, 2)])
+    assert_never_signal([(0, 1), (2, 3)])
+
+    def to_intervals(values: t.List[t.Tuple[int, int]]) -> t.List[Interval]:
+        return [(to_datetime(s), to_datetime(e)) for s, e in values]
+
+    def assert_check_intervals(
+        intervals: t.List[t.Tuple[int, int]],
+        ready: t.List[t.List[t.Tuple[int, int]]],
+        expected: t.List[t.Tuple[int, int]],
+    ):
+        signal_mock = mocker.Mock()
+        signal_mock.check_intervals = mocker.MagicMock(side_effect=[to_intervals(r) for r in ready])
+        _check_ready_intervals(signal_mock, intervals) == expected
+
+    assert_check_intervals([], [], [])
+    assert_check_intervals([(0, 1)], [[]], [])
+    assert_check_intervals(
+        [(0, 1)],
+        [[(0, 1)]],
+        [(0, 1)],
+    )
+    assert_check_intervals(
+        [(0, 1), (1, 2)],
+        [[(0, 1)]],
+        [(0, 1)],
+    )
+    assert_check_intervals(
+        [(0, 1), (1, 2)],
+        [[(1, 2)]],
+        [(1, 2)],
+    )
+    assert_check_intervals(
+        [(0, 1), (1, 2)],
+        [[(0, 1), (1, 2)]],
+        [(0, 1), (1, 2)],
+    )
+    assert_check_intervals(
+        [(0, 1), (1, 2), (3, 4)],
+        [[], []],
+        [],
+    )
+    assert_check_intervals(
+        [(0, 1), (1, 2), (3, 4)],
+        [[(0, 1)], []],
+        [(0, 1)],
+    )
+    assert_check_intervals(
+        [(0, 1), (1, 2), (3, 4)],
+        [[(0, 1)], [(3, 4)]],
+        [(0, 1), (3, 4)],
+    )
diff --git a/tests/dbt/test_manifest.py b/tests/dbt/test_manifest.py
index bb81d193..59b4d0ff 100644
--- a/tests/dbt/test_manifest.py
+++ b/tests/dbt/test_manifest.py
@@ -25,38 +25,45 @@ def test_manifest_helper(caplog):
         variable_overrides={"start": "2020-01-01"},
     )
 
-    assert helper.models()["top_waiters"].dependencies == Dependencies(
+    models = helper.models()
+
+    assert models["top_waiters"].dependencies == Dependencies(
         refs={"sushi.waiter_revenue_by_day", "waiter_revenue_by_day"},
         variables={"top_waiters:revenue", "top_waiters:limit"},
         macros=[MacroReference(name="ref"), MacroReference(name="var")],
     )
-    assert helper.models()["top_waiters"].materialized == "view"
-    assert helper.models()["top_waiters"].dialect_ == "postgres"
+    assert models["top_waiters"].materialized == "view"
+    assert models["top_waiters"].dialect_ == "postgres"
 
-    assert helper.models()["waiters"].dependencies == Dependencies(
+    assert models["waiters"].dependencies == Dependencies(
         macros={MacroReference(name="incremental_by_time"), MacroReference(name="source")},
         sources={"streaming.orders"},
     )
-    assert helper.models()["waiters"].materialized == "ephemeral"
-    assert helper.models()["items_snapshot"].materialized == "snapshot"
-    assert helper.models()["items_snapshot"].updated_at == "ds"
-    assert helper.models()["items_snapshot"].unique_key == ["id"]
-    assert helper.models()["items_snapshot"].strategy == "timestamp"
-    assert helper.models()["items_snapshot"].table_schema == "snapshots"
-    assert helper.models()["items_snapshot"].invalidate_hard_deletes is True
-    assert helper.models()["items_check_snapshot"].materialized == "snapshot"
-    assert helper.models()["items_check_snapshot"].check_cols == ["ds"]
-    assert helper.models()["items_check_snapshot"].unique_key == ["id"]
-    assert helper.models()["items_check_snapshot"].strategy == "check"
-    assert helper.models()["items_check_snapshot"].table_schema == "snapshots"
-    assert helper.models()["items_check_snapshot"].invalidate_hard_deletes is True
-    assert helper.models()["items_no_hard_delete_snapshot"].materialized == "snapshot"
-    assert helper.models()["items_no_hard_delete_snapshot"].unique_key == ["id"]
-    assert helper.models()["items_no_hard_delete_snapshot"].strategy == "timestamp"
-    assert helper.models()["items_no_hard_delete_snapshot"].table_schema == "snapshots"
-    assert helper.models()["items_no_hard_delete_snapshot"].invalidate_hard_deletes is False
-
-    waiter_as_customer_by_day_config = helper.models()["waiter_as_customer_by_day"]
+    assert models["waiters"].materialized == "ephemeral"
+    assert models["items_snapshot"].materialized == "snapshot"
+    assert models["items_snapshot"].updated_at == "ds"
+    assert models["items_snapshot"].unique_key == ["id"]
+    assert models["items_snapshot"].strategy == "timestamp"
+    assert models["items_snapshot"].table_schema == "snapshots"
+    assert models["items_snapshot"].invalidate_hard_deletes is True
+    assert models["items_check_snapshot"].materialized == "snapshot"
+    assert models["items_check_snapshot"].check_cols == ["ds"]
+    assert models["items_check_snapshot"].unique_key == ["id"]
+    assert models["items_check_snapshot"].strategy == "check"
+    assert models["items_check_snapshot"].table_schema == "snapshots"
+    assert models["items_check_snapshot"].invalidate_hard_deletes is True
+    assert models["items_no_hard_delete_snapshot"].materialized == "snapshot"
+    assert models["items_no_hard_delete_snapshot"].unique_key == ["id"]
+    assert models["items_no_hard_delete_snapshot"].strategy == "timestamp"
+    assert models["items_no_hard_delete_snapshot"].table_schema == "snapshots"
+    assert models["items_no_hard_delete_snapshot"].invalidate_hard_deletes is False
+
+    # Test versioned models
+    assert models["waiter_revenue_by_day_v1"].version == 1
+    assert models["waiter_revenue_by_day_v2"].version == 2
+    assert "waiter_revenue_by_day" not in models
+
+    waiter_as_customer_by_day_config = models["waiter_as_customer_by_day"]
     assert waiter_as_customer_by_day_config.dependencies == Dependencies(
         refs={"waiters", "waiter_names", "customers"},
         macros=[MacroReference(name="ref")],
@@ -66,7 +73,7 @@ def test_manifest_helper(caplog):
     assert waiter_as_customer_by_day_config.cluster_by == ["ds"]
     assert waiter_as_customer_by_day_config.time_column == "ds"
 
-    waiter_revenue_by_day_config = helper.models()["waiter_revenue_by_day"]
+    waiter_revenue_by_day_config = models["waiter_revenue_by_day_v2"]
     assert waiter_revenue_by_day_config.dependencies == Dependencies(
         macros={
             MacroReference(name="log_value"),
@@ -96,12 +103,14 @@ def test_manifest_helper(caplog):
 
     assert helper.seeds()["waiter_names"].path == Path("seeds/waiter_names.csv")
 
-    assert helper.sources()["streaming.items"].table_name == "items"
-    assert helper.sources()["streaming.items"].schema_ == "raw"
-    assert helper.sources()["streaming.orders"].table_name == "orders"
-    assert helper.sources()["streaming.orders"].schema_ == "raw"
-    assert helper.sources()["streaming.order_items"].table_name == "order_items"
-    assert helper.sources()["streaming.order_items"].schema_ == "raw"
+    sources = helper.sources()
+
+    assert sources["streaming.items"].table_name == "items"
+    assert sources["streaming.items"].schema_ == "raw"
+    assert sources["streaming.orders"].table_name == "orders"
+    assert sources["streaming.orders"].schema_ == "raw"
+    assert sources["streaming.order_items"].table_name == "order_items"
+    assert sources["streaming.order_items"].schema_ == "raw"
 
 
 @pytest.mark.xdist_group("dbt_manifest")
diff --git a/tests/fixtures/dbt/sushi_test/models/schema.yml b/tests/fixtures/dbt/sushi_test/models/schema.yml
index d6119c53..78d6c3e9 100644
--- a/tests/fixtures/dbt/sushi_test/models/schema.yml
+++ b/tests/fixtures/dbt/sushi_test/models/schema.yml
@@ -7,6 +7,9 @@ models:
   - name: waiters
   - name: waiter_as_customer_by_day
   - name: waiter_revenue_by_day
+    versions:
+      - v: 1
+      - v: 2
   - name: disabled_model
     columns:
       - name: one
diff --git a/tests/fixtures/dbt/sushi_test/models/waiter_revenue_by_day_v1.sql b/tests/fixtures/dbt/sushi_test/models/waiter_revenue_by_day_v1.sql
new file mode 100644
index 00000000..335e7ab7
--- /dev/null
+++ b/tests/fixtures/dbt/sushi_test/models/waiter_revenue_by_day_v1.sql
@@ -0,0 +1,40 @@
+{{
+    config(
+        materialized='incremental',
+        incremental_strategy='delete+insert',
+        cluster_by=['ds'],
+        time_column='ds',
+        dialect="bigquery"
+    )
+}}
+
+{{ log_value(5) }}
+{{ log(adapter.dispatch('current_engine', 'customers')()) }}
+
+{{ test_dependencies() }}
+
+{% set results = run_query('select 1 as constant') %}
+
+SELECT
+  o.waiter_id::INT AS waiter_id, /* Waiter id */
+  SUM(oi.quantity * i.price)::DOUBLE AS revenue, /* Revenue from orders taken by this waiter */
+  o.ds::TEXT AS ds, /* Date */
+  {% if execute %}
+    {{ results.columns[0].values()[0] }}::INT AS constant /* Constant */
+  {% endif %}
+FROM {{ source('streaming', 'orders') }} AS o
+LEFT JOIN {{ source('streaming', 'order_items') }} AS oi
+  ON o.id = oi.order_id AND o.ds = oi.ds
+LEFT JOIN {{ source('streaming', 'items') }} AS i
+  ON oi.item_id = i.id AND oi.ds = i.ds
+{% if is_incremental() %}
+  WHERE
+    o.ds > (select max(ds) from {{ this }})
+{% endif %}
+{% if sqlmesh_incremental is defined %}
+  WHERE
+      o.ds BETWEEN '{{ start_ds }}' AND '{{ end_ds }}'
+{% endif %}
+GROUP BY
+  o.waiter_id,
+  o.ds