Repository: TobikoData/sqlmesh
Pull request #2806 (2024-06-21 19:05:43)

Feat: enable dbt transpilation
This PR allows dbt project models and tests to be written in a different SQL dialect than the target engine on which the project is being run.

A project has a default dialect - all models/tests without an explicitly configured dialect are assumed to be written in the default. In a dbt project without an explicit default dialect specified, the target engine dialect is the default. 

Specify a default dialect other than the target engine in the `config.py` file model defaults argument
  - Example: `config=sqlmesh_config(Path(__file__).parent, model_defaults=ModelDefaultsConfig(dialect="bigquery"))

Specify dialects for individual models/tests in the standard way:
- Configuration jinja at top of file: `{{ config(dialect='bigquery') }}`
- YAML configuration files in file directories 
- dbt_project.yml

Limitations
- SQLMesh allows kind-specific parameters like `time_column` to be specified in a different dialect than the model query itself. In dbt, kind-specific parameters are parsed with the model's dialect.


Patch:
diff --git a/setup.py b/setup.py
index 3a21a965..a3d9b48c 100644
--- a/setup.py
+++ b/setup.py
@@ -1,11 +1,8 @@
 import os
-from pathlib import Path
 from os.path import exists
 
 from setuptools import find_packages, setup
 
-custom_materializations_path = Path("examples/custom_materializations").absolute()
-
 description = open("README.md").read() if exists("README.md") else ""
 
 setup(
@@ -95,7 +92,7 @@ setup(
             "types-pytz",
             "types-requests==2.28.8",
             "typing-extensions",
-            f"custom_materializations @ file://localhost/{custom_materializations_path}",
+            "custom-materializations",
         ],
         "cicdtest": [
             "dbt-databricks",
diff --git a/sqlmesh/dbt/adapter.py b/sqlmesh/dbt/adapter.py
index 92322f91..a581956d 100644
--- a/sqlmesh/dbt/adapter.py
+++ b/sqlmesh/dbt/adapter.py
@@ -29,12 +29,12 @@ class BaseAdapter(abc.ABC):
         self,
         jinja_macros: JinjaMacroRegistry,
         jinja_globals: t.Optional[t.Dict[str, t.Any]] = None,
-        dialect: t.Optional[str] = None,
+        project_dialect: t.Optional[str] = None,
     ):
         self.jinja_macros = jinja_macros
         self.jinja_globals = jinja_globals.copy() if jinja_globals else {}
         self.jinja_globals["adapter"] = self
-        self.dialect = dialect
+        self.project_dialect = project_dialect
 
     @abc.abstractmethod
     def get_relation(self, database: str, schema: str, identifier: str) -> t.Optional[BaseRelation]:
@@ -93,7 +93,7 @@ class BaseAdapter(abc.ABC):
 
     def quote(self, identifier: str) -> str:
         """Returns a quoted identifier."""
-        return exp.to_column(identifier).sql(dialect=self.dialect, identify=True)
+        return exp.to_column(identifier).sql(dialect=self.project_dialect, identify=True)
 
     def dispatch(self, name: str, package: t.Optional[str] = None) -> t.Callable:
         """Returns a dialect-specific version of a macro with the given name."""
@@ -189,12 +189,17 @@ class RuntimeAdapter(BaseAdapter):
         snapshots: t.Optional[t.Dict[str, Snapshot]] = None,
         table_mapping: t.Optional[t.Dict[str, str]] = None,
         deployability_index: t.Optional[DeployabilityIndex] = None,
+        project_dialect: t.Optional[str] = None,
     ):
         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, dialect=engine_adapter.dialect)
+        super().__init__(
+            jinja_macros,
+            jinja_globals=jinja_globals,
+            project_dialect=project_dialect or engine_adapter.dialect,
+        )
 
         table_mapping = table_mapping or {}
 
@@ -259,7 +264,9 @@ class RuntimeAdapter(BaseAdapter):
 
         mapped_table = self._map_table_name(self._normalize(self._relation_to_table(relation)))
         return [
-            Column.from_description(name=name, raw_data_type=dtype.sql(dialect=self.dialect))
+            Column.from_description(
+                name=name, raw_data_type=dtype.sql(dialect=self.project_dialect)
+            )
             for name, dtype in self.engine_adapter.columns(table_name=mapped_table).items()
         ]
 
@@ -298,12 +305,12 @@ class RuntimeAdapter(BaseAdapter):
             self.engine_adapter.fetchdf if fetch else self.engine_adapter.execute  # type: ignore
         )
 
-        expression = parse_one(sql, read=self.dialect)
+        expression = parse_one(sql, read=self.project_dialect)
         with normalize_and_quote(
-            expression, t.cast(str, self.dialect), self.engine_adapter.default_catalog
+            expression, t.cast(str, self.project_dialect), self.engine_adapter.default_catalog
         ) as expression:
             expression = exp.replace_tables(
-                expression, self.table_mapping, dialect=self.dialect, copy=False
+                expression, self.table_mapping, dialect=self.project_dialect, copy=False
             )
 
         if auto_begin:
@@ -328,17 +335,17 @@ class RuntimeAdapter(BaseAdapter):
         return identifier if identifier else None
 
     def _map_table_name(self, table: exp.Table) -> exp.Table:
-        name = table.sql()
+        name = table.sql(dialect=self.project_dialect)
         physical_table_name = self.table_mapping.get(name)
         if not physical_table_name:
             return table
 
         logger.debug("Resolved ref '%s' to snapshot table '%s'", name, physical_table_name)
 
-        return exp.to_table(physical_table_name, dialect=self.dialect)
+        return exp.to_table(physical_table_name, dialect=self.project_dialect)
 
     def _relation_to_table(self, relation: BaseRelation) -> exp.Table:
-        return exp.to_table(relation.render(), dialect=self.dialect)
+        return exp.to_table(relation.render(), dialect=self.project_dialect)
 
     def _table_to_relation(self, table: exp.Table) -> BaseRelation:
         return self.relation_type.create(
@@ -358,7 +365,7 @@ class RuntimeAdapter(BaseAdapter):
 
     def _normalize(self, input_table: exp.Table) -> exp.Table:
         normalized_name = normalize_model_name(
-            input_table, self.engine_adapter.default_catalog, self.dialect
+            input_table, self.engine_adapter.default_catalog, self.project_dialect
         )
         normalized_table = exp.to_table(normalized_name)
         if not input_table.this:
diff --git a/sqlmesh/dbt/basemodel.py b/sqlmesh/dbt/basemodel.py
index e6bf24bf..0e670531 100644
--- a/sqlmesh/dbt/basemodel.py
+++ b/sqlmesh/dbt/basemodel.py
@@ -105,6 +105,7 @@ class BaseModelConfig(GeneralConfig):
     path: Path = Path()
     dependencies: Dependencies = Dependencies()
     tests: t.List[TestConfig] = []
+    dialect_: t.Optional[str] = Field(None, alias="dialect")
 
     # DBT configuration fields
     name: str = ""
@@ -187,6 +188,9 @@ class BaseModelConfig(GeneralConfig):
         """
         return f"{self.package_name}.{self.name}"
 
+    def dialect(self, context: DbtContext) -> str:
+        return self.dialect_ or context.default_dialect
+
     def canonical_name(self, context: DbtContext) -> str:
         """
         Get the sqlmesh model name
@@ -296,7 +300,7 @@ class BaseModelConfig(GeneralConfig):
         )
         return {
             "audits": [(test.name, {}) for test in self.tests],
-            "columns": column_types_to_sqlmesh(self.columns, context.dialect) or None,
+            "columns": column_types_to_sqlmesh(self.columns, self.dialect(context)) or None,
             "column_descriptions": column_descriptions_to_sqlmesh(self.columns) or None,
             "depends_on": {
                 model.canonical_name(context) for model in model_context.refs.values()
diff --git a/sqlmesh/dbt/builtin.py b/sqlmesh/dbt/builtin.py
index 1338f14e..23fc8964 100644
--- a/sqlmesh/dbt/builtin.py
+++ b/sqlmesh/dbt/builtin.py
@@ -43,9 +43,9 @@ class Exceptions:
 
 
 class Api:
-    def __init__(self, target: t.Optional[AttributeDict] = None) -> None:
-        if target:
-            config_class = TARGET_TYPE_TO_CONFIG_CLASS[target["type"]]
+    def __init__(self, dialect: t.Optional[str]) -> None:
+        if dialect:
+            config_class = TARGET_TYPE_TO_CONFIG_CLASS[dialect]
             self.Relation = config_class.relation_class
             self.Column = config_class.column_class
             self.quote_policy = config_class.quote_policy
@@ -301,8 +301,10 @@ def create_builtin_globals(
     jinja_globals = jinja_globals.copy()
 
     target: t.Optional[AttributeDict] = jinja_globals.get("target", None)
-    api = Api(target)
-    dialect = target.dialect if target else None  # type: ignore
+    project_dialect = jinja_globals.pop("dialect", None) or (
+        target.get("dialect") if target else None
+    )
+    api = Api(project_dialect)
 
     builtin_globals["api"] = api
 
@@ -349,13 +351,14 @@ def create_builtin_globals(
             snapshots=jinja_globals.get("snapshots", {}),
             table_mapping=jinja_globals.get("table_mapping", {}),
             deployability_index=jinja_globals.get("deployability_index"),
+            project_dialect=project_dialect,
         )
     else:
         builtin_globals["flags"] = Flags(which="parse")
         adapter = ParsetimeAdapter(
             jinja_macros,
             jinja_globals={**builtin_globals, **jinja_globals},
-            dialect=dialect,
+            project_dialect=project_dialect,
         )
 
     sql_execution = SQLExecution(adapter)
diff --git a/sqlmesh/dbt/context.py b/sqlmesh/dbt/context.py
index 855cfaf0..d6a41286 100644
--- a/sqlmesh/dbt/context.py
+++ b/sqlmesh/dbt/context.py
@@ -59,9 +59,13 @@ class DbtContext:
     _manifest: t.Optional[ManifestHelper] = None
 
     @property
-    def dialect(self) -> str:
+    def default_dialect(self) -> str:
+        if self.sqlmesh_config.dialect:
+            return self.sqlmesh_config.dialect
         if not self.target:
-            raise SQLMeshError("Target must be configured before calling the dialect property.")
+            raise SQLMeshError(
+                "Target must be configured before calling the default_dialect property."
+            )
         return self.target.dialect
 
     @property
@@ -229,6 +233,9 @@ class DbtContext:
             output["project_name"] = self.project_name
         if self._target is not None:
             output["target"] = self._target.attribute_dict()
+        # pass user-specified default dialect if we have already loaded the config
+        if self.sqlmesh_config.dialect:
+            output["dialect"] = self.sqlmesh_config.dialect
         return output
 
     def context_for_dependencies(self, dependencies: Dependencies) -> DbtContext:
diff --git a/sqlmesh/dbt/loader.py b/sqlmesh/dbt/loader.py
index 16f1a728..f9669fa7 100644
--- a/sqlmesh/dbt/loader.py
+++ b/sqlmesh/dbt/loader.py
@@ -3,7 +3,6 @@ from __future__ import annotations
 import logging
 import typing as t
 from pathlib import Path
-
 from sqlmesh.core import constants as c
 from sqlmesh.core.audit import Audit
 from sqlmesh.core.config import (
@@ -43,7 +42,8 @@ def sqlmesh_config(
     context = DbtContext(project_root=project_root)
     profile = Profile.load(context, target_name=dbt_target_name)
     model_defaults = kwargs.pop("model_defaults", ModelDefaultsConfig())
-    model_defaults.dialect = profile.target.dialect
+    if model_defaults.dialect is None:
+        model_defaults.dialect = profile.target.dialect
 
     target_to_sqlmesh_args = {}
     if register_comments is not None:
diff --git a/sqlmesh/dbt/model.py b/sqlmesh/dbt/model.py
index ada0f3e4..80fffa58 100644
--- a/sqlmesh/dbt/model.py
+++ b/sqlmesh/dbt/model.py
@@ -201,7 +201,9 @@ class ModelConfig(BaseModelConfig):
         if materialization == Materialization.VIEW:
             return ViewKind()
         if materialization == Materialization.INCREMENTAL:
-            incremental_materialization_kwargs: t.Dict[str, t.Any] = {"dialect": context.dialect}
+            incremental_materialization_kwargs: t.Dict[str, t.Any] = {
+                "dialect": self.dialect(context)
+            }
             for field in ("batch_size", "lookback", "forward_only"):
                 field_val = getattr(self, field, None) or self.meta.get(field, None)
                 if field_val:
@@ -277,7 +279,7 @@ class ModelConfig(BaseModelConfig):
                     f"{self.canonical_name(context)}: SQLMesh snapshot strategy is required for snapshot materialization."
                 )
             shared_kwargs = {
-                "dialect": context.dialect,
+                "dialect": self.dialect(context),
                 "unique_key": self.unique_key,
                 "invalidate_hard_deletes": self.invalidate_hard_deletes,
                 "valid_from_name": "dbt_valid_from",
@@ -358,14 +360,14 @@ class ModelConfig(BaseModelConfig):
 
     def to_sqlmesh(self, context: DbtContext) -> Model:
         """Converts the dbt model into a SQLMesh model."""
-        dialect = context.dialect
+        model_dialect = self.dialect(context)
         query = d.jinja_query(self.sql_no_config)
 
         optional_kwargs: t.Dict[str, t.Any] = {}
 
         if self.partition_by:
             optional_kwargs["partitioned_by"] = (
-                [exp.to_column(val) for val in self.partition_by]
+                [exp.to_column(val, dialect=model_dialect) for val in self.partition_by]
                 if isinstance(self.partition_by, list)
                 else self._big_query_partition_by_expr(context)
             )
@@ -374,7 +376,7 @@ class ModelConfig(BaseModelConfig):
             clustered_by = []
             for c in self.cluster_by:
                 try:
-                    clustered_by.append(d.parse_one(c, dialect=dialect).name)
+                    clustered_by.append(d.parse_one(c, dialect=model_dialect).name)
                 except SqlglotError as e:
                     raise ConfigError(f"Failed to parse cluster_by field '{c}': {e}") from e
             optional_kwargs["clustered_by"] = clustered_by
@@ -403,7 +405,7 @@ class ModelConfig(BaseModelConfig):
         return create_sql_model(
             self.canonical_name(context),
             query,
-            dialect=dialect,
+            dialect=model_dialect,
             kind=self.model_kind(context),
             start=self.start,
             **optional_kwargs,
diff --git a/sqlmesh/dbt/seed.py b/sqlmesh/dbt/seed.py
index 3ae58e51..50a1ec9d 100644
--- a/sqlmesh/dbt/seed.py
+++ b/sqlmesh/dbt/seed.py
@@ -50,7 +50,7 @@ class SeedConfig(BaseModelConfig):
         return create_seed_model(
             self.canonical_name(context),
             SeedKind(path=seed_path),
-            dialect=context.dialect,
+            dialect=self.dialect(context),
             **kwargs,
         )
 
diff --git a/sqlmesh/dbt/target.py b/sqlmesh/dbt/target.py
index 70c54407..69bc5eb3 100644
--- a/sqlmesh/dbt/target.py
+++ b/sqlmesh/dbt/target.py
@@ -852,5 +852,6 @@ TARGET_TYPE_TO_CONFIG_CLASS: t.Dict[str, t.Type[TargetConfig]] = {
     "snowflake": SnowflakeConfig,
     "bigquery": BigQueryConfig,
     "sqlserver": MSSQLConfig,
+    "tsql": MSSQLConfig,
     "trino": TrinoConfig,
 }


Test patch:
diff --git a/sqlmesh/dbt/test.py b/sqlmesh/dbt/test.py
index 38755f91..5f005ebc 100644
--- a/sqlmesh/dbt/test.py
+++ b/sqlmesh/dbt/test.py
@@ -6,7 +6,6 @@ from enum import Enum
 from pathlib import Path
 
 from pydantic import Field
-
 import sqlmesh.core.dialect as d
 from sqlmesh.core.audit import Audit, ModelAudit, StandaloneAudit
 from sqlmesh.dbt.common import (
@@ -48,6 +47,7 @@ class TestConfig(GeneralConfig):
         interval_unit: The duration of an interval for the audit. By default, it is computed from the cron expression.
         column_name: The name of the column under test.
         dependencies: The macros, refs, and sources the test depends upon.
+        dialect: SQL dialect of the test query.
         package_name: Name of the package that defines the test.
         alias: The alias for the materialized table where failures are stored (Not supported).
         schema: The schema for the materialized table where the failures are stored (Not supported).
@@ -73,6 +73,7 @@ class TestConfig(GeneralConfig):
     interval_unit: t.Optional[str] = None
     column_name: t.Optional[str] = None
     dependencies: Dependencies = Dependencies()
+    dialect_: t.Optional[str] = Field(None, alias="dialect")
 
     # dbt fields
     package_name: str = ""
@@ -110,6 +111,9 @@ class TestConfig(GeneralConfig):
     def sqlmesh_config_fields(self) -> t.Set[str]:
         return {"description", "owner", "stamp", "cron", "interval_unit"}
 
+    def dialect(self, context: DbtContext) -> str:
+        return self.dialect_ or context.default_dialect
+
     def to_sqlmesh(self, context: DbtContext) -> Audit:
         """Convert dbt Test to SQLMesh Audit
 
@@ -142,7 +146,7 @@ class TestConfig(GeneralConfig):
             jinja_macros.add_globals({"this": self.relation_info})
             audit = StandaloneAudit(
                 name=self.name,
-                dialect=context.dialect,
+                dialect=self.dialect(context),
                 skip=skip,
                 query=query,
                 jinja_macros=jinja_macros,
@@ -158,7 +162,7 @@ class TestConfig(GeneralConfig):
         else:
             audit = ModelAudit(
                 name=self.name,
-                dialect=context.dialect,
+                dialect=self.dialect(context),
                 skip=skip,
                 blocking=blocking,
                 query=query,
diff --git a/tests/dbt/test_adapter.py b/tests/dbt/test_adapter.py
index a68668f7..f8dfe4cd 100644
--- a/tests/dbt/test_adapter.py
+++ b/tests/dbt/test_adapter.py
@@ -73,10 +73,23 @@ def test_normalization(
     context = sushi_test_project.context
     assert context.target
 
+    # bla and bob will be normalized to lowercase since the target is duckdb
     adapter_mock = mocker.MagicMock()
-    adapter_mock.dialect = "snowflake"
     adapter_mock.default_catalog = "test"
+    adapter_mock.dialect = "duckdb"
+
+    duckdb_renderer = runtime_renderer(context, engine_adapter=adapter_mock)
+
+    schema_bla = schema_("bla", "test", quoted=True)
+    relation_bla_bob = exp.table_("bob", db="bla", catalog="test", quoted=True)
 
+    duckdb_renderer("{{ adapter.get_relation(database=None, schema='bla', identifier='bob') }}")
+    adapter_mock.table_exists.assert_has_calls([call(relation_bla_bob)])
+
+    # bla and bob will be normalized to uppercase since the target is Snowflake, even though the default dialect is duckdb
+    adapter_mock = mocker.MagicMock()
+    adapter_mock.default_catalog = "test"
+    adapter_mock.dialect = "snowflake"
     context.target = SnowflakeConfig(
         account="test",
         user="test",
@@ -85,16 +98,19 @@ def test_normalization(
         database="test",
         schema="test",
     )
-
     renderer = runtime_renderer(context, engine_adapter=adapter_mock)
 
-    # bla and bob will be normalized to uppercase since we're dealing with Snowflake
-    schema_bla = schema_("BLA", "TEST", quoted=True)
-    relation_bla_bob = exp.table_("BOB", db="BLA", catalog="TEST", quoted=True)
+    schema_bla = schema_("bla", "test", quoted=True)
+    relation_bla_bob = exp.table_("bob", db="bla", catalog="test", quoted=True)
 
     renderer("{{ adapter.get_relation(database=None, schema='bla', identifier='bob') }}")
     adapter_mock.table_exists.assert_has_calls([call(relation_bla_bob)])
 
+    renderer("{{ adapter.get_relation(database='custom_db', schema='bla', identifier='bob') }}")
+    adapter_mock.table_exists.assert_has_calls(
+        [call(exp.table_("bob", db="bla", catalog="custom_db", quoted=True))]
+    )
+
     renderer(
         "{%- set relation = api.Relation.create(schema='bla') -%}"
         "{{ adapter.create_schema(relation) }}"
@@ -114,7 +130,7 @@ def test_normalization(
     adapter_mock.drop_table.assert_has_calls([call(relation_bla_bob)])
 
     expected_star_query: exp.Select = exp.maybe_parse(
-        'SELECT * FROM "T" as "T"', dialect="snowflake"
+        'SELECT * FROM "t" as "t"', dialect="snowflake"
     )
 
     # The following call to run_query won't return dataframes and so we're expected to
diff --git a/tests/dbt/test_config.py b/tests/dbt/test_config.py
index 4be59e38..bbd4712b 100644
--- a/tests/dbt/test_config.py
+++ b/tests/dbt/test_config.py
@@ -2,12 +2,11 @@ import base64
 import typing as t
 from pathlib import Path
 from shutil import copytree
-from unittest.mock import PropertyMock
 
 import pytest
 from dbt.adapters.base import BaseRelation, Column
 from pytest_mock import MockerFixture
-
+from sqlmesh.core.config import Config, ModelDefaultsConfig
 from sqlmesh.core.dialect import jinja_query
 from sqlmesh.core.model import SqlModel
 from sqlmesh.core.model.kind import OnDestructiveChange
@@ -124,6 +123,17 @@ def test_model_to_sqlmesh_fields():
     assert kind.lookback == 3
     assert kind.on_destructive_change == OnDestructiveChange.ALLOW
 
+    model = model_config.update_with({"dialect": "snowflake"}).to_sqlmesh(context)
+    assert model.dialect == "snowflake"
+
+    bq_default_context = DbtContext(
+        sqlmesh_config=Config(model_defaults=ModelDefaultsConfig(dialect="bigquery"))
+    )
+    bq_default_context.project_name = "Foo"
+    bq_default_context.target = DuckDbConfig(name="target", schema="foo")
+    model = model_config.to_sqlmesh(bq_default_context)
+    assert model.dialect == "bigquery"
+
 
 def test_test_to_sqlmesh_fields():
     sql = "SELECT * FROM FOO WHERE cost > 100"
@@ -165,6 +175,20 @@ def test_test_to_sqlmesh_fields():
     assert not audit.blocking
     assert sql in audit.query.sql()
 
+    test_config = TestConfig(
+        name="foo_null_test",
+        sql=sql,
+        model_name="Foo",
+        column_name="id",
+        severity="WARN",
+        enabled=False,
+        dialect="bigquery",
+    )
+
+    audit = test_config.to_sqlmesh(context)
+
+    assert audit.dialect == "bigquery"
+
 
 def test_singular_test_to_standalone_audit():
     sql = "SELECT * FROM FOO.BAR WHERE cost > 100"
@@ -201,6 +225,10 @@ def test_singular_test_to_standalone_audit():
     assert standalone_audit.query == jinja_query(sql)
     assert standalone_audit.depends_on == {'"memory"."foo"."bar"'}
 
+    test_config.dialect_ = "bigquery"
+    standalone_audit = test_config.to_sqlmesh(context)
+    assert standalone_audit.dialect == "bigquery"
+
 
 def test_model_config_sql_no_config():
     assert (
@@ -347,8 +375,8 @@ def test_seed_config(sushi_test_project: Project, mocker: MockerFixture):
     context = sushi_test_project.context
     assert raw_items_seed.canonical_name(context) == "sushi.waiter_names"
     assert raw_items_seed.to_sqlmesh(context).name == "sushi.waiter_names"
-    mock_dialect = PropertyMock(return_value="snowflake")
-    mocker.patch("sqlmesh.dbt.context.DbtContext.dialect", mock_dialect)
+
+    raw_items_seed.dialect_ = "snowflake"
     assert raw_items_seed.to_sqlmesh(sushi_test_project.context).name == "sushi.waiter_names"
     assert (
         raw_items_seed.to_sqlmesh(sushi_test_project.context).fqn
diff --git a/tests/dbt/test_manifest.py b/tests/dbt/test_manifest.py
index afd8b0ea..bb81d193 100644
--- a/tests/dbt/test_manifest.py
+++ b/tests/dbt/test_manifest.py
@@ -31,6 +31,7 @@ def test_manifest_helper(caplog):
         macros=[MacroReference(name="ref"), MacroReference(name="var")],
     )
     assert helper.models()["top_waiters"].materialized == "view"
+    assert helper.models()["top_waiters"].dialect_ == "postgres"
 
     assert helper.models()["waiters"].dependencies == Dependencies(
         macros={MacroReference(name="incremental_by_time"), MacroReference(name="source")},
@@ -81,6 +82,7 @@ def test_manifest_helper(caplog):
     assert waiter_revenue_by_day_config.incremental_strategy == "delete+insert"
     assert waiter_revenue_by_day_config.cluster_by == ["ds"]
     assert waiter_revenue_by_day_config.time_column == "ds"
+    assert waiter_revenue_by_day_config.dialect_ == "bigquery"
 
     assert helper.models("customers")["customers"].dependencies == Dependencies(
         sources={"raw.orders"},
diff --git a/tests/dbt/test_transformation.py b/tests/dbt/test_transformation.py
index 1fb31a82..1c9e947c 100644
--- a/tests/dbt/test_transformation.py
+++ b/tests/dbt/test_transformation.py
@@ -10,7 +10,6 @@ from dbt.exceptions import CompilationError
 from freezegun import freeze_time
 from pytest_mock.plugin import MockerFixture
 from sqlglot import exp, parse_one
-
 from sqlmesh.core import dialect as d
 from sqlmesh.core.audit import StandaloneAudit
 from sqlmesh.core.context import Context
@@ -100,6 +99,20 @@ def test_model_kind():
         execution_time_as_valid_from=True,
         dialect="duckdb",
     )
+    assert ModelConfig(
+        materialized=Materialization.SNAPSHOT,
+        unique_key=["id"],
+        strategy="check",
+        check_cols=["foo"],
+        dialect="bigquery",
+    ).model_kind(context) == SCDType2ByColumnKind(
+        unique_key=["id"],
+        valid_from_name="dbt_valid_from",
+        valid_to_name="dbt_valid_to",
+        columns=["foo"],
+        execution_time_as_valid_from=True,
+        dialect="bigquery",
+    )
 
     assert ModelConfig(materialized=Materialization.INCREMENTAL, time_column="foo").model_kind(
         context
@@ -118,9 +131,12 @@ def test_model_kind():
         time_column="foo", dialect="duckdb", forward_only=True
     )
     assert ModelConfig(
-        materialized=Materialization.INCREMENTAL, time_column="foo", unique_key=["bar"]
+        materialized=Materialization.INCREMENTAL,
+        time_column="foo",
+        unique_key=["bar"],
+        dialect="bigquery",
     ).model_kind(context) == IncrementalByTimeRangeKind(
-        time_column="foo", dialect="duckdb", forward_only=True
+        time_column="foo", dialect="bigquery", forward_only=True
     )
 
     assert ModelConfig(
@@ -286,6 +302,24 @@ def test_model_kind_snapshot_bigquery():
         dialect="bigquery",
     )
 
+    # time_data_type is bigquery version even though model dialect is DuckDB
+    # because model target is BigQuery
+    assert ModelConfig(
+        materialized=Materialization.SNAPSHOT,
+        unique_key=["id"],
+        updated_at="updated_at",
+        strategy="timestamp",
+        dialect="duckdb",
+    ).model_kind(context) == SCDType2ByTimeKind(
+        unique_key=["id"],
+        valid_from_name="dbt_valid_from",
+        valid_to_name="dbt_valid_to",
+        updated_at_as_valid_from=True,
+        updated_at_name="updated_at",
+        time_data_type=exp.DataType.build("TIMESTAMPTZ"),  # bigquery version
+        dialect="duckdb",
+    )
+
 
 def test_model_columns():
     model = ModelConfig(
@@ -389,6 +423,35 @@ def test_seed_column_inference(tmp_path):
     }
 
 
+@pytest.mark.xdist_group("dbt_manifest")
+def test_model_dialect(sushi_test_project: Project, assert_exp_eq):
+    model_config = ModelConfig(
+        name="model",
+        package_name="package",
+        schema="sushi",
+        alias="table",
+        sql="SELECT 1 AS `one` FROM {{ schema }}",
+    )
+    context = sushi_test_project.context
+
+    # cannot parse model sql without specifying bigquery dialect
+    with pytest.raises(ConfigError):
+        model_config.to_sqlmesh(context).render_query_or_raise().sql()
+
+    model_config = ModelConfig(
+        name="model",
+        package_name="package",
+        schema="sushi",
+        alias="table",
+        sql="SELECT 1 AS `one` FROM {{ schema }}",
+        dialect="bigquery",
+    )
+    assert_exp_eq(
+        model_config.to_sqlmesh(context).render_query_or_raise().sql(),
+        'SELECT 1 AS "one" FROM "sushi" AS "sushi"',
+    )
+
+
 @pytest.mark.xdist_group("dbt_manifest")
 @pytest.mark.parametrize(
     "model_fqn", ['"memory"."sushi"."waiters"', '"memory"."sushi"."waiter_names"']
@@ -500,6 +563,31 @@ def test_test_this(assert_exp_eq, sushi_test_project: Project):
     )
 
 
+@pytest.mark.xdist_group("dbt_manifest")
+def test_test_dialect(assert_exp_eq, sushi_test_project: Project):
+    test_config = TestConfig(
+        name="test",
+        alias="alias",
+        database="database",
+        schema_="schema",
+        standalone=True,
+        sql="SELECT 1 AS `one` FROM {{ this.identifier }}",
+    )
+    context = sushi_test_project.context
+
+    # can't parse test sql without specifying bigquery as default dialect
+    with pytest.raises(ConfigError):
+        audit = t.cast(StandaloneAudit, test_config.to_sqlmesh(context))
+        audit.render_query(audit).sql()
+
+    test_config.dialect_ = "bigquery"
+    audit = t.cast(StandaloneAudit, test_config.to_sqlmesh(context))
+    assert_exp_eq(
+        audit.render_query(audit).sql(),
+        'SELECT 1 AS "one" FROM "test" AS "test"',
+    )
+
+
 @pytest.mark.xdist_group("dbt_manifest")
 def test_statement(sushi_test_project: Project, runtime_renderer: t.Callable):
     context = sushi_test_project.context
diff --git a/tests/fixtures/dbt/sushi_test/config.py b/tests/fixtures/dbt/sushi_test/config.py
index f08c22cc..6f7f9453 100644
--- a/tests/fixtures/dbt/sushi_test/config.py
+++ b/tests/fixtures/dbt/sushi_test/config.py
@@ -1,12 +1,14 @@
 from pathlib import Path
 
-from sqlmesh.core.config import AirflowSchedulerConfig
+from sqlmesh.core.config import AirflowSchedulerConfig, ModelDefaultsConfig
 from sqlmesh.dbt.loader import sqlmesh_config
 
 variables = {"start": "Jan 1 2022"}
 
 
-config = sqlmesh_config(Path(__file__).parent, variables=variables)
+config = sqlmesh_config(
+    Path(__file__).parent, variables=variables, model_defaults=ModelDefaultsConfig(dialect="duckdb")
+)
 
 
 test_config = config
diff --git a/tests/fixtures/dbt/sushi_test/models/schema.yml b/tests/fixtures/dbt/sushi_test/models/schema.yml
index ae99f55e..d6119c53 100644
--- a/tests/fixtures/dbt/sushi_test/models/schema.yml
+++ b/tests/fixtures/dbt/sushi_test/models/schema.yml
@@ -2,6 +2,8 @@ version: 2
 
 models:
   - name: top_waiters
+    config:
+      dialect: postgres
   - name: waiters
   - name: waiter_as_customer_by_day
   - name: waiter_revenue_by_day
diff --git a/tests/fixtures/dbt/sushi_test/models/waiter_revenue_by_day.sql b/tests/fixtures/dbt/sushi_test/models/waiter_revenue_by_day.sql
index a0021e33..335e7ab7 100644
--- a/tests/fixtures/dbt/sushi_test/models/waiter_revenue_by_day.sql
+++ b/tests/fixtures/dbt/sushi_test/models/waiter_revenue_by_day.sql
@@ -4,6 +4,7 @@
         incremental_strategy='delete+insert',
         cluster_by=['ds'],
         time_column='ds',
+        dialect="bigquery"
     )
 }}