Feat!: Fully Qualify Models and Tables with Connection Catalog (#1574)

* feat!: use connection as source for catalog

* fmt

* remove integration test

* update docs

Author: eakmanrq

.pre-commit-config.yaml

@@ -36,6 +36,7 @@ repos:
         "types": [python]
         files: *files
         require_serial: true
+        exclude: ^(tests/fixtures)
   - repo: https://github.com/pre-commit/mirrors-prettier
     rev: "fc26039"
     hooks:

docs/concepts/models/python_models.md

@@ -202,7 +202,7 @@ from sqlmesh import ExecutionContext, model
 def execute(
     context: ExecutionContext,
     start: datetime,
-    end: datetime
+    end: datetime,
     execution_time: datetime,
     **kwargs: t.Any,
 ) -> DataFrame:

docs/reference/model_configuration.md

@@ -46,18 +46,6 @@ The SQLMesh project-level `model_defaults` key supports the following options, d
 
 The project-level `model_defaults` key also supports two keys for specifying a default catalog or schema, described [below](#default-schema--catalog).
 
-#### Default Schema / Catalog
-SQLMesh model names may contain one to three levels of nesting, where the naming hierarchy is `catalog.schema.model`.
-
-SQLMesh requires all model names and references to have the same level of nesting. For example, if you decide to specify a catalog in any model name, all model names and references must include a catalog.
-
-You can specify default schema and catalog names in the `model_defaults` key if you want to omit them from some model names.
-
-| Option    | Description                                                        | Type | Required |
-| --------- | ------------------------------------------------------------------ | :--: | :------: |
-| `catalog` | The default catalog name for models that do not specify a catalog. | str  |    N     |
-| `schema`  | The default schema name for models that do not specify a schema.   | str  |    N     |
-
 ## Model kind properties
 
 Configuration options for kind-specific SQLMesh model properties.

examples/sushi/models/waiters.py

@@ -36,7 +36,17 @@ def entrypoint(evaluator: MacroEvaluator) -> exp.Select:
 
         parent_snapshots = snapshot.parents
         assert len(parent_snapshots) == 1
-        assert parent_snapshots[0].name.lower() == "sushi.orders"
+        name = '"sushi"."orders"'
+        # There are tests which force not having a default catalog so we check here if one is defined
+        # and add it to the name if it is
+        default_catalog = evaluator.default_catalog
+        if default_catalog:
+            # make sure we don't double quote the default catalog
+            default_catalog = default_catalog.strip('"')
+            name = ".".join([f'"{default_catalog}"', name])
+        assert (
+            parent_snapshots[0].name == name
+        ), f"Snapshot Name: {parent_snapshots[0].name}, Name: {name}"
 
     excluded = {"id", "customer_id", "start_ts", "end_ts"}
     projections = []

setup.py

@@ -64,6 +64,9 @@
             "google-cloud-bigquery",
             "google-cloud-bigquery-storage",
             "black==22.6.0",
+            "dbt-bigquery; python_version >= '3.8'",
+            # Remove once we drop support for Python 3.7
+            "dbt-bigquery==1.5.2; python_version < '3.8'",
             "dbt-core",
             "dbt-duckdb>=1.4.2",
             "dbt-snowflake",

sqlmesh/core/audit/definition.py

@@ -12,7 +12,12 @@
 from sqlmesh.core import constants as c
 from sqlmesh.core import dialect as d
 from sqlmesh.core.macros import MacroRegistry, macro
-from sqlmesh.core.model.common import bool_validator, expression_validator
+from sqlmesh.core.model.common import (
+    bool_validator,
+    default_catalog_validator,
+    depends_on_validator,
+    expression_validator,
+)
 from sqlmesh.core.model.definition import _Model, _python_env, _single_value_or_tuple
 from sqlmesh.core.node import _Node
 from sqlmesh.core.renderer import QueryRenderer
@@ -195,7 +200,7 @@ def render_query(
 
         node = snapshot_or_node if isinstance(snapshot_or_node, _Node) else snapshot_or_node.node
         this_model = kwargs.pop("this_model", None) or (
-            node.name
+            node.fqn
             if isinstance(snapshot_or_node, _Node)
             else t.cast(Snapshot, snapshot_or_node).table_name(
                 deployability_index.is_deployable(snapshot_or_node)
@@ -248,6 +253,7 @@ def _create_query_renderer(self, node: _Node) -> QueryRenderer:
             jinja_macro_registry=self.jinja_macros,
             python_env=model.python_env,
             only_execution_time=model.kind.only_execution_time,
+            default_catalog=model.default_catalog,
         )
 
     def __str__(self) -> str:
@@ -272,6 +278,7 @@ class StandaloneAudit(_Node, AuditMixin):
     defaults: t.Dict[str, exp.Expression] = {}
     expressions_: t.Optional[t.List[exp.Expression]] = Field(default=None, alias="expressions")
     jinja_macros: JinjaMacroRegistry = JinjaMacroRegistry()
+    default_catalog: t.Optional[str] = None
     depends_on_: t.Optional[t.Set[str]] = Field(default=None, alias="depends_on")
     hash_raw_query: bool = False
     python_env_: t.Optional[t.Dict[str, Executable]] = Field(default=None, alias="python_env")
@@ -285,6 +292,8 @@ class StandaloneAudit(_Node, AuditMixin):
     _bool_validator = bool_validator
     _string_validator = audit_string_validator
     _map_validator = audit_map_validator
+    _default_catalog_validator = default_catalog_validator
+    _depends_on_validator = depends_on_validator
 
     @model_validator(mode="after")
     @model_validator_v1_args
@@ -301,7 +310,9 @@ def depends_on(self) -> t.Set[str]:
 
             query = self.render_query(self)
             if query is not None:
-                self._depends_on |= d.find_tables(query, dialect=self.dialect)
+                self._depends_on |= d.find_tables(
+                    query, default_catalog=self.default_catalog, dialect=self.dialect
+                )
 
             self._depends_on -= {self.name}
         return self._depends_on
@@ -427,6 +438,7 @@ def _create_query_renderer(self, node: _Node) -> QueryRenderer:
             path=self._path or Path(),
             jinja_macro_registry=self.jinja_macros,
             python_env=self.python_env,
+            default_catalog=self.default_catalog,
         )
 
 
@@ -453,6 +465,7 @@ def load_audit(
     macros: t.Optional[MacroRegistry] = None,
     jinja_macros: t.Optional[JinjaMacroRegistry] = None,
     dialect: t.Optional[str] = None,
+    default_catalog: t.Optional[str] = None,
 ) -> Audit:
     """Load an audit from a parsed SQLMesh audit file.
 
@@ -523,11 +536,16 @@ def load_audit(
         extra_kwargs["jinja_macros"] = (jinja_macros or JinjaMacroRegistry()).trim(
             jinja_macro_refrences
         )
+        extra_kwargs["default_catalog"] = default_catalog
 
     dialect = meta_fields.pop("dialect", dialect)
     try:
         audit = audit_class(
-            query=query, expressions=statements, dialect=dialect, **meta_fields, **extra_kwargs
+            query=query,
+            expressions=statements,
+            dialect=dialect,
+            **extra_kwargs,
+            **meta_fields,
         )
     except Exception as ex:
         _raise_config_error(str(ex), path)
@@ -544,6 +562,7 @@ def load_multiple_audits(
     macros: t.Optional[MacroRegistry] = None,
     jinja_macros: t.Optional[JinjaMacroRegistry] = None,
     dialect: t.Optional[str] = None,
+    default_catalog: t.Optional[str] = None,
 ) -> t.Generator[Audit, None, None]:
     audit_block: t.List[exp.Expression] = []
     for expression in expressions:
@@ -556,13 +575,15 @@ def load_multiple_audits(
                     macros=macros,
                     jinja_macros=jinja_macros,
                     dialect=dialect,
+                    default_catalog=default_catalog,
                 )
                 audit_block.clear()
         audit_block.append(expression)
     yield load_audit(
         expressions=audit_block,
         path=path,
         dialect=dialect,
+        default_catalog=default_catalog,
     )
 
 
@@ -586,4 +607,5 @@ def _maybe_parse_arg_pair(e: exp.Expression) -> t.Tuple[str, exp.Expression]:
     "depends_on_": lambda value: exp.Tuple(expressions=sorted(value)),
     "tags": _single_value_or_tuple,
     "hash_raw_query": exp.convert,
+    "default_catalog": exp.to_identifier,
 }

sqlmesh/core/config/connection.py

@@ -542,6 +542,9 @@ def _connection_factory(self) -> t.Callable:
 
         return connect
 
+    def get_catalog(self) -> t.Optional[str]:
+        return self.project
+
 
 class GCPPostgresConnectionConfig(ConnectionConfig):
     """

sqlmesh/core/config/model.py

@@ -2,8 +2,6 @@
 
 import typing as t
 
-from pydantic import Field
-
 from sqlmesh.core.config.base import BaseConfig
 from sqlmesh.core.model.kind import ModelKind, model_kind_validator
 from sqlmesh.utils.date import TimeLike
@@ -26,8 +24,6 @@ class ModelDefaultsConfig(BaseConfig):
             will be chunked such that each individual job will only contain jobs with max `batch_size` intervals.
         storage_format: The storage format used to store the physical table, only applicable in certain engines.
             (eg. 'parquet')
-        catalog: The default catalog to use if one is not provided (catalog.schema.table).
-        schema: The default schema to use if one is not provided. (catalog.schema.table).
     """
 
     kind: t.Optional[ModelKind] = None
@@ -37,7 +33,5 @@ class ModelDefaultsConfig(BaseConfig):
     start: t.Optional[TimeLike] = None
     batch_size: t.Optional[int] = None
     storage_format: t.Optional[str] = None
-    catalog: t.Optional[str] = None
-    schema_: t.Optional[str] = Field(default=None, alias="schema")
 
     _model_kind_validator = model_kind_validator

sqlmesh/core/config/scheduler.py

@@ -54,6 +54,14 @@ def create_state_sync(self, context: Context) -> StateSync:
             The StateSync instance.
         """
 
+    @abc.abstractmethod
+    def get_default_catalog(self, context: Context) -> t.Optional[str]:
+        """Returns the default catalog for the Scheduler.
+
+        Args:
+            context: The SQLMesh Context.
+        """
+
 
 class _EngineAdapterStateSyncSchedulerConfig(_SchedulerConfig):
     def create_state_sync(self, context: Context) -> StateSync:
@@ -69,6 +77,9 @@ def create_state_sync(self, context: Context) -> StateSync:
         schema = context.config.get_state_schema(context.gateway)
         return EngineAdapterStateSync(engine_adapter, schema=schema, console=context.console)
 
+    def get_default_catalog(self, context: Context) -> t.Optional[str]:
+        return context.engine_adapter.default_catalog
+
 
 class BuiltInSchedulerConfig(_EngineAdapterStateSyncSchedulerConfig, BaseConfig):
     """The Built-In Scheduler configuration."""
@@ -79,6 +90,7 @@ def create_plan_evaluator(self, context: Context) -> PlanEvaluator:
         return BuiltInPlanEvaluator(
             state_sync=context.state_sync,
             snapshot_evaluator=context.snapshot_evaluator,
+            default_catalog=self.get_default_catalog(context),
             backfill_concurrent_tasks=context.concurrent_tasks,
             console=context.console,
             notification_target_manager=context.notification_target_manager,
@@ -125,6 +137,9 @@ def create_plan_evaluator(self, context: Context) -> PlanEvaluator:
             state_sync=context.state_sync if self.use_state_connection else None,
         )
 
+    def get_default_catalog(self, context: Context) -> t.Optional[str]:
+        return None
+
 
 class AirflowSchedulerConfig(_BaseAirflowSchedulerConfig, BaseConfig):
     """The Airflow Scheduler configuration.
@@ -280,6 +295,9 @@ def create_plan_evaluator(self, context: Context) -> PlanEvaluator:
             users=context.users,
         )
 
+    def get_default_catalog(self, context: Context) -> t.Optional[str]:
+        return None
+
 
 SchedulerConfig = Annotated[
     t.Union[