Feat(experimental): DBT project conversion (#4495)

Author: erindru

sqlmesh/cli/main.py

@@ -33,6 +33,7 @@
     "rollback",
     "run",
     "table_name",
+    "dbt",
 )
 SKIP_CONTEXT_COMMANDS = ("init", "ui")
 
@@ -1219,3 +1220,39 @@ def state_import(obj: Context, input_file: Path, replace: bool, no_confirm: bool
     """Import a state export file back into the state database"""
     confirm = not no_confirm
     obj.import_state(input_file=input_file, clear=replace, confirm=confirm)
+
+
+@cli.group(no_args_is_help=True, hidden=True)
+def dbt() -> None:
+    """Commands for doing dbt-specific things"""
+    pass
+
+
+@dbt.command("convert")
+@click.option(
+    "-i",
+    "--input-dir",
+    help="Path to the DBT project",
+    required=True,
+    type=click.Path(exists=True, dir_okay=True, file_okay=False, readable=True, path_type=Path),
+)
+@click.option(
+    "-o",
+    "--output-dir",
+    required=True,
+    help="Path to write out the converted SQLMesh project",
+    type=click.Path(exists=False, dir_okay=True, file_okay=False, readable=True, path_type=Path),
+)
+@click.option("--no-prompts", is_flag=True, help="Disable interactive prompts", default=False)
+@click.pass_obj
+@error_handler
+@cli_analytics
+def dbt_convert(obj: Context, input_dir: Path, output_dir: Path, no_prompts: bool) -> None:
+    """Convert a DBT project to a SQLMesh project"""
+    from sqlmesh.dbt.converter.convert import convert_project_files
+
+    convert_project_files(
+        input_dir.absolute(),
+        output_dir.absolute(),
+        no_prompts=no_prompts,
+    )

sqlmesh/core/config/root.py

@@ -39,7 +39,7 @@
     scheduler_config_validator,
 )
 from sqlmesh.core.config.ui import UIConfig
-from sqlmesh.core.loader import Loader, SqlMeshLoader
+from sqlmesh.core.loader import Loader, SqlMeshLoader, MigratedDbtProjectLoader
 from sqlmesh.core.notification_target import NotificationTarget
 from sqlmesh.core.user import User
 from sqlmesh.utils.date import to_timestamp, now
@@ -219,6 +219,13 @@ def _normalize_and_validate_fields(cls, data: t.Any) -> t.Any:
                 f"^{k}$": v for k, v in physical_schema_override.items()
             }
 
+        if (
+            (variables := data.get("variables", ""))
+            and isinstance(variables, dict)
+            and c.MIGRATED_DBT_PROJECT_NAME in variables
+        ):
+            data["loader"] = MigratedDbtProjectLoader
+
         return data
 
     @model_validator(mode="after")

sqlmesh/core/constants.py

@@ -31,6 +31,9 @@
 MAX_MODEL_DEFINITION_SIZE = 10000
 """Maximum number of characters in a model definition"""
 
+MIGRATED_DBT_PROJECT_NAME = "__dbt_project_name__"
+MIGRATED_DBT_PACKAGES = "__dbt_packages__"
+
 
 # The maximum number of fork processes, used for loading projects
 # None means default to process pool, 1 means don't fork, :N is number of processes

sqlmesh/core/loader.py

@@ -38,7 +38,11 @@
 from sqlmesh.core.test import ModelTestMetadata, filter_tests_by_patterns
 from sqlmesh.utils import UniqueKeyDict, sys_path
 from sqlmesh.utils.errors import ConfigError
-from sqlmesh.utils.jinja import JinjaMacroRegistry, MacroExtractor
+from sqlmesh.utils.jinja import (
+    JinjaMacroRegistry,
+    MacroExtractor,
+    SQLMESH_DBT_COMPATIBILITY_PACKAGE,
+)
 from sqlmesh.utils.metaprogramming import import_python_file
 from sqlmesh.utils.pydantic import validation_error_message
 from sqlmesh.utils.process import create_process_pool_executor
@@ -548,6 +552,7 @@ def _load_sql_models(
         signals: UniqueKeyDict[str, signal],
         cache: CacheBase,
         gateway: t.Optional[str],
+        loading_default_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
     ) -> UniqueKeyDict[str, Model]:
         """Loads the sql models into a Dict"""
         models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
@@ -590,6 +595,7 @@ def _load_sql_models(
                 infer_names=self.config.model_naming.infer_names,
                 signal_definitions=signals,
                 default_catalog_per_gateway=self.context.default_catalog_per_gateway,
+                **loading_default_kwargs or {},
             )
 
             with create_process_pool_executor(
@@ -942,3 +948,104 @@ def _model_cache_entry_id(self, model_path: Path) -> str:
                     self._loader.context.gateway or self._loader.config.default_gateway_name,
                 ]
             )
+
+
+class MigratedDbtProjectLoader(SqlMeshLoader):
+    @property
+    def migrated_dbt_project_name(self) -> str:
+        return self.config.variables[c.MIGRATED_DBT_PROJECT_NAME]
+
+    def _load_scripts(self) -> t.Tuple[MacroRegistry, JinjaMacroRegistry]:
+        from sqlmesh.dbt.converter.common import infer_dbt_package_from_path
+        from sqlmesh.dbt.target import TARGET_TYPE_TO_CONFIG_CLASS
+
+        # Store a copy of the macro registry
+        standard_macros = macro.get_registry()
+
+        jinja_macros = JinjaMacroRegistry(
+            create_builtins_module=SQLMESH_DBT_COMPATIBILITY_PACKAGE,
+            top_level_packages=["dbt", self.migrated_dbt_project_name],
+        )
+        extractor = MacroExtractor()
+
+        macros_max_mtime: t.Optional[float] = None
+
+        for path in self._glob_paths(
+            self.config_path / c.MACROS,
+            ignore_patterns=self.config.ignore_patterns,
+            extension=".py",
+        ):
+            if import_python_file(path, self.config_path):
+                self._track_file(path)
+                macro_file_mtime = self._path_mtimes[path]
+                macros_max_mtime = (
+                    max(macros_max_mtime, macro_file_mtime)
+                    if macros_max_mtime
+                    else macro_file_mtime
+                )
+
+        for path in self._glob_paths(
+            self.config_path / c.MACROS,
+            ignore_patterns=self.config.ignore_patterns,
+            extension=".sql",
+        ):
+            self._track_file(path)
+            macro_file_mtime = self._path_mtimes[path]
+            macros_max_mtime = (
+                max(macros_max_mtime, macro_file_mtime) if macros_max_mtime else macro_file_mtime
+            )
+
+            with open(path, "r", encoding="utf-8") as file:
+                try:
+                    package = infer_dbt_package_from_path(path) or self.migrated_dbt_project_name
+
+                    jinja_macros.add_macros(
+                        extractor.extract(file.read(), dialect=self.config.model_defaults.dialect),
+                        package=package,
+                    )
+                except Exception as e:
+                    raise ConfigError(f"Failed to load macro file: {path}", e)
+
+        self._macros_max_mtime = macros_max_mtime
+
+        macros = macro.get_registry()
+        macro.set_registry(standard_macros)
+
+        connection_config = self.context.connection_config
+        # this triggers the DBT create_builtins_module to have a `target` property which is required for a bunch of DBT macros to work
+        if dbt_config_type := TARGET_TYPE_TO_CONFIG_CLASS.get(connection_config.type_):
+            try:
+                jinja_macros.add_globals(
+                    {
+                        "target": dbt_config_type.from_sqlmesh(
+                            connection_config,
+                            name=self.config.default_gateway_name,
+                        ).attribute_dict()
+                    }
+                )
+            except NotImplementedError:
+                raise ConfigError(f"Unsupported dbt target type: {connection_config.type_}")
+
+        return macros, jinja_macros
+
+    def _load_sql_models(
+        self,
+        macros: MacroRegistry,
+        jinja_macros: JinjaMacroRegistry,
+        audits: UniqueKeyDict[str, ModelAudit],
+        signals: UniqueKeyDict[str, signal],
+        cache: CacheBase,
+        gateway: t.Optional[str],
+        loading_default_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
+    ) -> UniqueKeyDict[str, Model]:
+        return super()._load_sql_models(
+            macros=macros,
+            jinja_macros=jinja_macros,
+            audits=audits,
+            signals=signals,
+            cache=cache,
+            gateway=gateway,
+            loading_default_kwargs=dict(
+                migrated_dbt_project_name=self.migrated_dbt_project_name,
+            ),
+        )

sqlmesh/core/model/definition.py

@@ -2017,6 +2017,7 @@ def load_sql_based_model(
     variables: t.Optional[t.Dict[str, t.Any]] = None,
     infer_names: t.Optional[bool] = False,
     blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
+    migrated_dbt_project_name: t.Optional[str] = None,
     **kwargs: t.Any,
 ) -> Model:
     """Load a model from a parsed SQLMesh model SQL file.
@@ -2193,6 +2194,7 @@ def load_sql_based_model(
             query_or_seed_insert,
             kind=kind,
             time_column_format=time_column_format,
+            migrated_dbt_project_name=migrated_dbt_project_name,
             **common_kwargs,
         )
 
@@ -2400,6 +2402,7 @@ def _create_model(
     signal_definitions: t.Optional[SignalRegistry] = None,
     variables: t.Optional[t.Dict[str, t.Any]] = None,
     blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
+    migrated_dbt_project_name: t.Optional[str] = None,
     **kwargs: t.Any,
 ) -> Model:
     validate_extra_and_required_fields(
@@ -2455,13 +2458,28 @@ def _create_model(
 
     if jinja_macros:
         jinja_macros = (
-            jinja_macros if jinja_macros.trimmed else jinja_macros.trim(jinja_macro_references)
+            jinja_macros
+            if jinja_macros.trimmed
+            else jinja_macros.trim(jinja_macro_references, package=migrated_dbt_project_name)
         )
     else:
         jinja_macros = JinjaMacroRegistry()
 
-    for jinja_macro in jinja_macros.root_macros.values():
-        used_variables.update(extract_macro_references_and_variables(jinja_macro.definition)[1])
+    if migrated_dbt_project_name:
+        # extract {{ var() }} references used in all jinja macro dependencies to check for any variables specific
+        # to a migrated DBT package and resolve them accordingly
+        # vars are added into __sqlmesh_vars__ in the Python env so that the native SQLMesh var() function can resolve them
+        variables = variables or {}
+
+        nested_macro_used_variables, flattened_package_variables = (
+            _extract_migrated_dbt_variable_references(jinja_macros, variables)
+        )
+
+        used_variables.update(nested_macro_used_variables)
+        variables.update(flattened_package_variables)
+    else:
+        for jinja_macro in jinja_macros.root_macros.values():
+            used_variables.update(extract_macro_references_and_variables(jinja_macro.definition)[1])
 
     model = klass(
         name=name,
@@ -2844,7 +2862,7 @@ def render_expression(
     "cron_tz": lambda value: exp.Literal.string(value),
     "partitioned_by_": _single_expr_or_tuple,
     "clustered_by": _single_expr_or_tuple,
-    "depends_on_": lambda value: exp.Tuple(expressions=sorted(value)),
+    "depends_on_": lambda value: exp.Tuple(expressions=sorted(value)) if value else "()",
     "pre": _list_of_calls_to_exp,
     "post": _list_of_calls_to_exp,
     "audits": _list_of_calls_to_exp,
@@ -2915,4 +2933,37 @@ def clickhouse_partition_func(
     )
 
 
+def _extract_migrated_dbt_variable_references(
+    jinja_macros: JinjaMacroRegistry, project_variables: t.Dict[str, t.Any]
+) -> t.Tuple[t.Set[str], t.Dict[str, t.Any]]:
+    if not jinja_macros.trimmed:
+        raise ValueError("Expecting a trimmed JinjaMacroRegistry")
+
+    used_variables = set()
+    # note: JinjaMacroRegistry is trimmed here so "all_macros" should be just be all the macros used by this model
+    for _, _, jinja_macro in jinja_macros.all_macros:
+        _, extracted_variable_names = extract_macro_references_and_variables(jinja_macro.definition)
+        used_variables.update(extracted_variable_names)
+
+    flattened = {}
+    if (dbt_package_variables := project_variables.get(c.MIGRATED_DBT_PACKAGES)) and isinstance(
+        dbt_package_variables, dict
+    ):
+        # flatten the nested dict structure from the migrated dbt package variables in the SQLmesh config into __dbt_packages.<package>.<variable>
+        # to match what extract_macro_references_and_variables() returns. This allows the usage checks in create_python_env() to work
+        def _flatten(prefix: str, root: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
+            acc = {}
+            for k, v in root.items():
+                key_with_prefix = f"{prefix}.{k}"
+                if isinstance(v, dict):
+                    acc.update(_flatten(key_with_prefix, v))
+                else:
+                    acc[key_with_prefix] = v
+            return acc
+
+        flattened = _flatten(c.MIGRATED_DBT_PACKAGES, dbt_package_variables)
+
+    return used_variables, flattened
+
+
 TIME_COL_PARTITION_FUNC = {"clickhouse": clickhouse_partition_func}

sqlmesh/core/model/kind.py

@@ -4,7 +4,7 @@
 from enum import Enum
 from typing_extensions import Self
 
-from pydantic import Field
+from pydantic import Field, BeforeValidator
 from sqlglot import exp
 from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 from sqlglot.optimizer.qualify_columns import quote_identifiers
@@ -33,6 +33,7 @@
     field_validator,
     get_dialect,
     validate_string,
+    positive_int_validator,
 )
 
 
@@ -455,7 +456,7 @@ class IncrementalByUniqueKeyKind(_IncrementalBy):
     unique_key: SQLGlotListOfFields
     when_matched: t.Optional[exp.Whens] = None
     merge_filter: t.Optional[exp.Expression] = None
-    batch_concurrency: t.Literal[1] = 1
+    batch_concurrency: t.Annotated[t.Literal[1], BeforeValidator(positive_int_validator)] = 1
 
     @field_validator("when_matched", mode="before")
     def _when_matched_validator(

sqlmesh/core/renderer.py

@@ -179,6 +179,7 @@ def _resolve_table(table: str | exp.Table) -> str:
         )
 
         render_kwargs = {
+            "dialect": self._dialect,
             **date_dict(
                 to_datetime(execution_time or c.EPOCH),
                 start_time,

sqlmesh/dbt/adapter.py

@@ -38,6 +38,9 @@ def __init__(
         self.jinja_globals = jinja_globals.copy() if jinja_globals else {}
         self.jinja_globals["adapter"] = self
         self.project_dialect = project_dialect
+        self.jinja_globals["dialect"] = (
+            project_dialect  # so the dialect is available in the jinja env created by self.dispatch()
+        )
         self.quote_policy = quote_policy or Policy()
 
     @abc.abstractmethod

sqlmesh/dbt/builtin.py

@@ -156,7 +156,7 @@ class Var:
     def __init__(self, variables: t.Dict[str, t.Any]) -> None:
         self.variables = variables
 
-    def __call__(self, name: str, default: t.Optional[t.Any] = None) -> t.Any:
+    def __call__(self, name: str, default: t.Optional[t.Any] = None, **kwargs: t.Any) -> t.Any:
         return self.variables.get(name, default)
 
     def has_var(self, name: str) -> bool: