Repository: TobikoData/sqlmesh
Pull request #2962 (2024-07-30 01:52:36)

Fix!: Handle lookback correctly on incremental models with an end date defined
Slack context: https://tobiko-data.slack.com/archives/C044BRE5W4S/p1721917406454879

This PR allows `lookback` to return the correct missing intervals if the execution date is beyond the model's end date. Prior this, `sqlmesh run` would always show a phantom interval for these models even if the current date was well beyond the model end date + lookback period.

Assuming an incremental model with an end date of `2023-01-05`, `@daily` intervals and a lookback of `2` days:
 - The user does not want data from the source system beyond `2023-01-05` so they don't want _new_ intervals after 2023-01-05
 - However, they do want already run intervals to re-run after `2023-01-05` if they are within the lookback period
 - So the following intervals would run on the following days:
     - `2023-01-05` -> `2023-01-05`, `2023-01-04`, `2023-01-03`
     - `2023-01-06` -> `2023-01-04`, `2023-01-05`
     - `2023-01-07` -> `2023-01-05`
     - `2023-01-08` -> (none, we are done)


Patch:
diff --git a/docs/faq/faq.md b/docs/faq/faq.md
index 74014753..ed072bb7 100644
--- a/docs/faq/faq.md
+++ b/docs/faq/faq.md
@@ -57,7 +57,7 @@
 ??? question "What is semantic understanding of SQL?"
     Semantic understanding is the result of analyzing SQL code to determine what it does at a granular level. SQLMesh uses the free, open-source Python library [SQLGlot](https://github.com/tobymao/sqlglot) to parse the SQL code and build the semantic understanding.
 
-    Semantic understanding allows SQLMesh to do things like transpilation (executing one SQL dialect on an engine running another dialect) and protecting incremental loading queries from duplicating data.
+    Semantic understanding allows SQLMesh to do things like transpilation (executing one SQL dialect on an engine running another dialect) and preventing incremental loading queries from duplicating data.
 
 ??? question "Does SQLMesh work like Terraform?"
     SQLMesh was inspired by Terraform, but its commands are not equivalent.
@@ -74,7 +74,7 @@
     SQLMesh is a Python library. After ensuring you have [an appropriate Python runtime](../prerequisites.md), install it [with `pip`](../installation.md).
 
 ??? question "How do I use SQLMesh?"
-    SQLMesh has three interfaces: [command line](../reference/cli.md), [Jupyter or Databricks notebook](../reference/notebook.md), and graphical user interface.
+    SQLMesh has three interfaces: [command line](../reference/cli.md), [Jupyter or Databricks notebook](../reference/notebook.md), and [graphical user interface](../guides/ui.md).
 
     The [quickstart guide](../quick_start.md) demonstrates an example project in each of the interfaces.
 
@@ -85,7 +85,9 @@
     SQLMesh creates schemas for two reasons:
 
     - SQLMesh stores state/metadata information about a project in the `sqlmesh` schema. This schema is created in the project's default gateway, or you can [specify a different location](../reference/configuration.md#state-connection).
-    - SQLMesh uses [Virtual Data Environments](https://tobikodata.com/virtual-data-environments.html) to prevent duplicative computation whenever possible.
+    - SQLMesh uses [Virtual Data Environments](https://tobikodata.com/virtual-data-environments.html) to prevent duplicative computation whenever possible, and stores environment-specific objects in separate schemas by default.
+
+    How Virtual Data Environments work:
 
     Virtual Data Environments work by maintaining a *virtual layer* of views that users interact with when building models and a *physical layer* of tables that stores the actual data.
 
@@ -102,8 +104,12 @@
 ??? question "What's the difference between a `test` and an `audit`?"
     A SQLMesh [`test`](../concepts/tests.md) is analogous to a "unit test" in software engineering. It tests *code* based on known inputs and outputs. In SQLMesh, the inputs and outputs are specified in a YAML file, and SQLMesh automatically runs them when `sqlmesh plan` is executed.
 
+    Writing YAML is annoying and error-prone, so SQLMesh's [`create_test` command](../concepts/tests.md#automatic-test-generation) allows you to automatically generate YAML test files based on queries of existing data tables.
+
     A SQLMesh [`audit`](../concepts/audits.md) validates that transformed *data* meet some criteria. For example, an `audit` might verify that a column contains no `NULL` values or has no duplicated values. SQLMesh automatically runs audits when a `sqlmesh plan` is executed and the plan is applied or when `sqlmesh run` is executed.
 
+    When the `sqlmesh plan` command is executed, SQLMesh `test`s run _before_ any model's code is executed. A SQLMesh model's `audit`s run _after_ the model's code is executed to validate the data output by the model.
+
 ??? question "How does a model know when to run?"
     A SQLMesh model determines when to run based on its [`cron`](#cron-question) parameter and how much time has elapsed since its previous run.
 
@@ -221,14 +227,14 @@
 
     SQLMesh always maintains state about the project structure, contents, and past runs. State information enables powerful SQLMesh features like virtual data environments and easy incremental loads.
 
-    State information is stored by default - you do not need to take any action to maintain or to use it when executing models. As the dbt caveats page says, state information is powerful but complex. SQLMesh handles that complexity for you so you don't need to learn about or understand the underlying mechanics.
+    State information is stored by default - you do not need to take any action to maintain or to use it when executing models. As the dbt caveats page says, state information is powerful but complex. SQLMesh handles that complexity for you so you don't need to worry about the underlying mechanics.
 
-    SQLMesh stores state information in database tables. By default, it stores this information in the same [database/connection where your project models run](../reference/configuration.md#gateways). You can specify a [different database/connection](../reference/configuration.md#state-connection) if you would prefer to store state information somewhere else.
+    SQLMesh stores state information in database tables. By default, it stores this information in the same [database/connection where your project models run](../reference/configuration.md#gateways). You can specify a [different database/connection](../reference/configuration.md#state-connection) if you would prefer to store state information somewhere else. We recommend using a separate connection for storing state in production deployments.
 
     SQLMesh adds information to the state tables via transactions, and some databases like BigQuery are not optimized to execute transactions. Changing the state connection to another database like PostgreSQL can alleviate performance issues you may encounter due to state transactions.
 
 ??? question "How do I get column-level lineage for my dbt project?"
-    SQLMesh can run dbt projects with its [dbt adapter](../integrations/dbt.md). After configuring the dbt project to work with SQLMesh, you can view the column-level lineage in the SQLMesh browser UI:
+    SQLMesh can run dbt projects with its [dbt adapter](../integrations/dbt.md). After configuring the dbt project to work with SQLMesh, you can view the column-level lineage in the [SQLMesh browser UI](../guides/ui.md):
 
     ![SQLMesh UI displaying column-level lineage for Sushi dbt example](./faq/ui-colum-lineage_sushi-dbt.png)
 
@@ -240,7 +246,7 @@
 ??? question "How do incremental models determine which dates to ingest?"
     dbt uses the "most recent record" approach to determine which dates should be included in an incremental load. It works by querying the existing data for the most recent date it contains, then ingesting all records after that date from the source system in a single query.
 
-    SQLMesh uses the "intervals" approach instead. It divides time into disjoint intervals based on a model's `cron` parameter then records which intervals have ever been processed. It ingests source records from only unprocessed intervals. The intervals approach enables features like loading in batches.
+    SQLMesh uses the ["intervals" approach](https://tobikodata.com/data_load_patterns_101.html) instead. It divides time into disjoint intervals based on a model's `cron` parameter then records which intervals have ever been processed. It ingests source records from only unprocessed intervals. The intervals approach enables features like loading in batches.
 
 ??? question "How do I run an append only model in SQLMesh?"
     SQLMesh does not support append-only models as implemented in dbt. You can achieve a similar outcome by defining a time column and using an [incremental by time range](../concepts/models/model_kinds.md#incremental_by_time_range) model or specifying a unique key and using an [incremental by unique key](../concepts/models/model_kinds.md#incremental_by_unique_key) model.
@@ -249,7 +255,7 @@
 
 ??? question "How does Tobiko Data make money?"
 
-    - Model execution observability and monitoring tools (in development)
+    - Tobiko Cloud: learn more [here](https://tobikodata.com/product.html)
     - Enterprise Github Actions CI/CD App (in development)
         - Advanced version of [open source CI/CD bot](../integrations/github.md)
     - Providing hands-on support for companies' SQLMesh projects
diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md
index cad840d5..8f5b12cd 100644
--- a/docs/guides/configuration.md
+++ b/docs/guides/configuration.md
@@ -489,12 +489,13 @@ These pages describe the connection configuration options for each execution eng
 
 Configuration for the state backend connection if different from the data warehouse connection.
 
-The data warehouse connection is used to store SQLMesh state if the `state_connection` key is not specified, unless the configuration uses an Airflow or Google Cloud Composer scheduler. If using one of those schedulers and the `state_connection` key is not specified, the scheduler's database is used (not the data warehouse).
-
-**NOTE: Using the same connection for data warehouse and state is only recommended for non-production deployments of SQLMesh.**
+The data warehouse connection is used to store SQLMesh state if the `state_connection` key is not specified, unless the configuration uses an Airflow or Google Cloud Composer scheduler. If using one of those schedulers, the scheduler's database is used (not the data warehouse) unless an [Airflow Connection has been configured](../integrations/airflow.md#state-connection).
 
 Unlike data transformations, storing state information requires database transactions. Data warehouses aren’t optimized for executing transactions, and storing state information in them can slow down your project or produce corrupted data due to simultaneous writes to the same table. Therefore, production SQLMesh deployments should use a dedicated state connection.
 
+!!! note
+    Using the same connection for data warehouse and state is not recommended for production deployments of SQLMesh.
+
 The easiest and most reliable way to manage your state connection is for [Tobiko Cloud](https://tobikodata.com/product.html) to do it for you. If you'd rather handle it yourself, we list recommended and unsupported state engines below.
 
 Recommended state engines for production deployments:
diff --git a/docs/integrations/airflow.md b/docs/integrations/airflow.md
index 5275728a..a3c2e9e4 100644
--- a/docs/integrations/airflow.md
+++ b/docs/integrations/airflow.md
@@ -11,7 +11,8 @@ SQLMesh provides first-class support for Airflow with the following capabilities
 ## Airflow cluster configuration
 To enable SQLMesh support on a target Airflow cluster, the SQLMesh package should first be installed on that cluster. Ensure it is installed with the extras for your engine if needed; for example: `sqlmesh[databricks]` for Databricks. Check [setup.py](https://github.com/TobikoData/sqlmesh/blob/main/setup.py) for a list of extras.
 
-**Note:** The Airflow Webserver instance(s) must be restarted after **installation** and every time the SQLMesh package is **upgraded**.
+!!! note
+    The Airflow Webserver instance(s) must be restarted after **installation** and every time the SQLMesh package is **upgraded**.
 
 Once the package is installed, the following Python module must be created in the `dags/` folder of the target DAG repository with the following contents:
 
diff --git a/docs/integrations/engines/bigquery.md b/docs/integrations/engines/bigquery.md
index 9acf2bf9..557e9e4f 100644
--- a/docs/integrations/engines/bigquery.md
+++ b/docs/integrations/engines/bigquery.md
@@ -98,5 +98,6 @@ sqlmesh_airflow = SQLMeshAirflow(
 
 ## Permissions Required
 With any of the above connection methods, ensure these BigQuery permissions are enabled to allow SQLMesh to work correctly.
+
 - [`BigQuery Data Editor`](https://cloud.google.com/bigquery/docs/access-control#bigquery.dataEditor)
 - [`BigQuery User`](https://cloud.google.com/bigquery/docs/access-control#bigquery.user)
diff --git a/docs/stylesheets/extra.css b/docs/stylesheets/extra.css
index 0304b175..498160fd 100644
--- a/docs/stylesheets/extra.css
+++ b/docs/stylesheets/extra.css
@@ -3,7 +3,7 @@
     border-color: rgb(68, 138, 255);
 }
 
-.md-typeset .question > .admonition-title, 
+.md-typeset .question > .admonition-title,
 .md-typeset .question > summary {
     background-color: rgba(68, 138, 255, 0.1);
 
@@ -19,3 +19,7 @@
 .md-nav__link {
     word-break: break-word;
 }
+
+:root {
+  --md-primary-fg-color: #3C64E2;
+}
\ No newline at end of file
diff --git a/mkdocs.yml b/mkdocs.yml
index 32f6ce48..d89d1a1a 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -97,16 +97,16 @@ theme:
   palette:
     - media: "(prefers-color-scheme: light)"
       scheme: default
-      primary: #091c79
-      accent: #cce8fe
+      primary: #3C64E2
+      accent: #3C64E2
       toggle:
         icon: material/weather-sunny
         name: Switch to dark mode
     # Palette toggle for dark mode
     - media: "(prefers-color-scheme: dark)"
       scheme: slate
-      primary: #cce8fe
-      accent: #091c79
+      primary: #3C64E2
+      accent: #3C64E2
       toggle:
         icon: material/weather-night
         name: Switch to light mode
diff --git a/setup.py b/setup.py
index 739e9eb3..0aef17b5 100644
--- a/setup.py
+++ b/setup.py
@@ -47,7 +47,7 @@ setup(
         "requests",
         "rich[jupyter]",
         "ruamel.yaml",
-        "sqlglot[rs]~=25.8.0",
+        "sqlglot[rs]~=25.8.1",
     ],
     extras_require={
         "bigquery": [
diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
index 68fd73ed..004602b2 100644
--- a/sqlmesh/core/dialect.py
+++ b/sqlmesh/core/dialect.py
@@ -454,7 +454,13 @@ def _parse_if(self: Parser) -> t.Optional[exp.Expression]:
         else:
             self.raise_error("Expecting )")
 
-        return exp.Anonymous(this="IF", expressions=[cond, self._parse_statement()])
+        index = self._index
+        stmt = self._parse_statement()
+        if self._curr:
+            self._retreat(index)
+            stmt = self._parse_as_command(self._tokens[index])
+
+        return exp.Anonymous(this="IF", expressions=[cond, stmt])
 
 
 def _create_parser(parser_type: t.Type[exp.Expression], table_keys: t.List[str]) -> t.Callable:
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 90adcabf..615fac88 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -902,10 +902,23 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
             upper_bound_ts = to_timestamp(execution_time)
             end_ts = min(end_ts, upper_bound_ts)
 
-        lookback = self.model.lookback if self.is_model else 0
+        lookback = 0
+        model_end_ts: t.Optional[int] = None
+
+        if self.is_model:
+            lookback = self.model.lookback
+            model_end_ts = (
+                to_timestamp(make_inclusive_end(self.model.end)) if self.model.end else None
+            )
 
         return compute_missing_intervals(
-            interval_unit, tuple(intervals), start_ts, end_ts, upper_bound_ts, lookback
+            interval_unit,
+            tuple(intervals),
+            start_ts,
+            end_ts,
+            upper_bound_ts,
+            lookback,
+            model_end_ts,
         )
 
     def categorize_as(self, category: SnapshotChangeCategory) -> None:
@@ -1641,6 +1654,7 @@ def compute_missing_intervals(
     end_ts: int,
     upper_bound_ts: int,
     lookback: int,
+    model_end_ts: t.Optional[int],
 ) -> Intervals:
     """Computes all missing intervals between start and end given intervals.
 
@@ -1651,6 +1665,7 @@ def compute_missing_intervals(
         end_ts: Exclusive timestamp end.
         upper_bound_ts: The exclusive upper bound timestamp for lookback.
         lookback: A lookback window.
+        model_end_ts: The inclusive end timestamp set on the model (if one is set)
 
     Returns:
         A list of all timestamps in this range.
@@ -1660,20 +1675,17 @@ def compute_missing_intervals(
 
     # get all individual timestamps with the addition of extra lookback timestamps up to the execution date
     # when a model has lookback, we need to check all the intervals between itself and its lookback exist.
+    intervals_beyond_end_ts = 0
     while True:
         ts = to_timestamp(croniter.get_next(estimate=True))
 
         if ts < end_ts:
             timestamps.append(ts)
-        else:
-            croniter.get_prev(estimate=True)
-            break
-
-    for _ in range(lookback):
-        ts = to_timestamp(croniter.get_next(estimate=True))
-        if ts < upper_bound_ts:
+        elif lookback and ts < upper_bound_ts:
             timestamps.append(ts)
+            intervals_beyond_end_ts += 1
         else:
+            croniter.get_prev(estimate=True)
             break
 
     missing = []
@@ -1697,7 +1709,8 @@ def compute_missing_intervals(
             elif current_ts >= low and compare_ts < high:
                 break
         else:
-            missing.append((current_ts, next_ts))
+            if model_end_ts is None or compare_ts < model_end_ts or i > intervals_beyond_end_ts + 1:
+                missing.append((current_ts, next_ts))
 
     return missing
 
diff --git a/sqlmesh/dbt/adapter.py b/sqlmesh/dbt/adapter.py
index 1820e72b..a8349c9a 100644
--- a/sqlmesh/dbt/adapter.py
+++ b/sqlmesh/dbt/adapter.py
@@ -30,11 +30,15 @@ class BaseAdapter(abc.ABC):
         jinja_macros: JinjaMacroRegistry,
         jinja_globals: t.Optional[t.Dict[str, t.Any]] = None,
         project_dialect: t.Optional[str] = None,
+        quote_policy: t.Optional[Policy] = None,
     ):
+        from dbt.adapters.base.relation import Policy
+
         self.jinja_macros = jinja_macros
         self.jinja_globals = jinja_globals.copy() if jinja_globals else {}
         self.jinja_globals["adapter"] = self
         self.project_dialect = project_dialect
+        self.quote_policy = quote_policy or Policy()
 
     @abc.abstractmethod
     def get_relation(self, database: str, schema: str, identifier: str) -> t.Optional[BaseRelation]:
@@ -95,6 +99,10 @@ class BaseAdapter(abc.ABC):
         """Returns a quoted identifier."""
         return exp.to_column(identifier).sql(dialect=self.project_dialect, identify=True)
 
+    def quote_as_configured(self, value: str, component_type: str) -> str:
+        """Returns the value quoted according to the quote policy."""
+        return self.quote(value) if getattr(self.quote_policy, component_type, False) else value
+
     def dispatch(self, name: str, package: t.Optional[str] = None) -> t.Callable:
         """Returns a dialect-specific version of a macro with the given name."""
         target_type = self.jinja_globals["target"]["type"]
@@ -216,12 +224,12 @@ class RuntimeAdapter(BaseAdapter):
     ):
         from dbt.adapters.base import BaseRelation
         from dbt.adapters.base.column import Column
-        from dbt.adapters.base.relation import Policy
 
         super().__init__(
             jinja_macros,
             jinja_globals=jinja_globals,
             project_dialect=project_dialect or engine_adapter.dialect,
+            quote_policy=quote_policy,
         )
 
         table_mapping = table_mapping or {}
@@ -229,7 +237,6 @@ class RuntimeAdapter(BaseAdapter):
         self.engine_adapter = engine_adapter
         self.relation_type = relation_type or BaseRelation
         self.column_type = column_type or Column
-        self.quote_policy = quote_policy or Policy()
         self.table_mapping = {
             **to_table_mapping((snapshots or {}).values(), deployability_index),
             **table_mapping,
diff --git a/sqlmesh/dbt/builtin.py b/sqlmesh/dbt/builtin.py
index b3d55461..d7ebcfd3 100644
--- a/sqlmesh/dbt/builtin.py
+++ b/sqlmesh/dbt/builtin.py
@@ -382,6 +382,7 @@ def create_builtin_globals(
             jinja_macros,
             jinja_globals={**builtin_globals, **jinja_globals},
             project_dialect=project_dialect,
+            quote_policy=api.quote_policy,
         )
 
     sql_execution = SQLExecution(adapter)


Test patch:
diff --git a/tests/core/test_dialect.py b/tests/core/test_dialect.py
index fd188181..6c5a6da2 100644
--- a/tests/core/test_dialect.py
+++ b/tests/core/test_dialect.py
@@ -606,3 +606,6 @@ def test_conditional_statement():
         q.sql("snowflake")
         == "@IF(TRUE, COPY INTO 's3://example/data.csv' FROM EXTRA.EXAMPLE.TABLE STORAGE_INTEGRATION = S3_INTEGRATION FILE_FORMAT = (TYPE=CSV COMPRESSION=NONE NULL_IF=('') FIELD_OPTIONALLY_ENCLOSED_BY='\"') HEADER = TRUE OVERWRITE = TRUE SINGLE = TRUE /* this is a comment */)"
     )
+
+    q = parse_one("@IF(cond, VACUUM ANALYZE);", read="postgres")
+    assert q.sql(dialect="postgres") == "@IF(cond, VACUUM ANALYZE)"
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 050f3662..077b5448 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -370,6 +370,85 @@ def test_missing_intervals_end_bounded_with_ignore_cron(make_snapshot):
     ]
 
 
+def test_missing_intervals_past_end_date_with_lookback(make_snapshot):
+    snapshot: Snapshot = make_snapshot(
+        SqlModel(
+            name="test_model",
+            kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ds"), lookback=2),
+            owner="owner",
+            cron="@daily",
+            query=parse_one("SELECT 1, ds FROM name"),
+            start="2023-01-01",
+            end="2023-01-05",  # inclusive, equivalent to to_timestamp('2023-01-05 23:59:59.999999')
+        )
+    )
+
+    start_time = to_timestamp("2023-01-01")
+    end_time = to_timestamp(
+        "2023-01-06"
+    )  # exclusive because to_timestamp() returns a timestamp and not a date
+    assert snapshot.inclusive_exclusive(snapshot.node.start, snapshot.node.end) == (
+        start_time,
+        end_time,
+    )
+
+    # baseline - all intervals missing
+    assert snapshot.missing_intervals(start_time, end_time, execution_time=end_time) == [
+        (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")),
+        (to_timestamp("2023-01-02"), to_timestamp("2023-01-03")),
+        (to_timestamp("2023-01-03"), to_timestamp("2023-01-04")),
+        (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")),
+        (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")),
+    ]
+
+    # fully backfill model - no intervals missing
+    snapshot.add_interval(start_time, end_time)
+
+    # even though lookback=2, because every interval has been filled,
+    # there should be no missing intervals
+    assert snapshot.missing_intervals(start_time, end_time, execution_time=end_time) == []
+
+    # however, when running for a new interval, this triggers lookback
+    # in this case, we remove the most recent interval (the one for 2023-01-05) to simulate it being new
+    # since lookback=2 days, this triggers missing intervals for 2023-01-03, 2023-01-04, 2023-01-05
+    snapshot.remove_interval(interval=(to_timestamp("2023-01-05"), to_timestamp("2023-01-06")))
+    assert snapshot.missing_intervals(start_time, end_time, execution_time=end_time) == [
+        (to_timestamp("2023-01-03"), to_timestamp("2023-01-04")),
+        (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")),
+        (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")),
+    ]
+
+    # put the interval we just removed back to make the model fully backfilled again
+    snapshot.add_interval(to_timestamp("2023-01-05"), to_timestamp("2023-01-06"))
+    assert snapshot.missing_intervals(start_time, end_time, execution_time=end_time) == []
+
+    # running on the end date + 1 day (2023-01-07)
+    # 2023-01-06 "would" run and since lookback=2 this pulls in 2023-01-04 and 2023-01-05 as well
+    # however, only 2023-01-04 and 2023-01-05 are within the model end date
+    end_time = to_timestamp("2023-01-07")
+    assert snapshot.missing_intervals(start_time, end_time, execution_time=end_time) == [
+        (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")),
+        (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")),
+    ]
+
+    # running on the end date + 2 days (2023-01-08)
+    # 2023-01-07 "would" run and since lookback=2 this pulls in 2023-01-06 and 2023-01-05 as well
+    # however, only 2023-01-05 is within the model end date
+    end_time = to_timestamp("2023-01-08")
+    assert snapshot.missing_intervals(start_time, end_time, execution_time=end_time) == [
+        (to_timestamp("2023-01-05"), to_timestamp("2023-01-06"))
+    ]
+
+    # running on the end date + 3 days (2023-01-09)
+    # no missing intervals because subtracting 2 days for lookback exceeds the models end date
+    end_time = to_timestamp("2023-01-09")
+    assert snapshot.missing_intervals(start_time, end_time, execution_time=end_time) == []
+
+    # running way in the future, no missing intervals because subtracting 2 days for lookback still exceeds the models end date
+    end_time = to_timestamp("2024-01-01")
+    assert snapshot.missing_intervals(start_time, end_time, execution_time=end_time) == []
+
+
 def test_incremental_time_self_reference(make_snapshot):
     snapshot = make_snapshot(
         SqlModel(
diff --git a/tests/dbt/test_adapter.py b/tests/dbt/test_adapter.py
index 320a64be..bfd71545 100644
--- a/tests/dbt/test_adapter.py
+++ b/tests/dbt/test_adapter.py
@@ -14,10 +14,12 @@ from sqlglot import exp, parse_one
 from sqlmesh import Context
 from sqlmesh.core.dialect import schema_
 from sqlmesh.core.snapshot import SnapshotId
+from sqlmesh.dbt.adapter import ParsetimeAdapter
 from sqlmesh.dbt.project import Project
 from sqlmesh.dbt.relation import Policy
 from sqlmesh.dbt.target import SnowflakeConfig
 from sqlmesh.utils.errors import ConfigError
+from sqlmesh.utils.jinja import JinjaMacroRegistry
 
 pytestmark = pytest.mark.dbt
 
@@ -257,3 +259,14 @@ def test_feature_flag_scd_type_2(copy_to_temp_path, caplog):
             "Skipping loading Snapshot (SCD Type 2) models due to the feature flag disabling this feature"
             in caplog.text
         )
+
+
+def test_quote_as_configured():
+    adapter = ParsetimeAdapter(
+        JinjaMacroRegistry(),
+        project_dialect="duckdb",
+        quote_policy=Policy(schema=False, identifier=True),
+    )
+    adapter.quote_as_configured("foo", "identifier") == '"foo"'
+    adapter.quote_as_configured("foo", "schema") == "foo"
+    adapter.quote_as_configured("foo", "database") == "foo"