Fix!: Derive the name of a snapshot's physical schema from its model's schema (#885)

Author: izeigerman

docs/reference/configuration.md

@@ -146,7 +146,6 @@ See [Airflow Integration Guide](../integrations/airflow.md) for information on h
 ## SQLMesh-specific configurations
 | Option                    | Description                                                                                                                                                                                                                                                                                        |         Type         | Required |
 |---------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:--------------------:|:--------:|
-| `physical_schema`         | The default schema used to store physical tables for models (Default: `sqlmesh`)                                                                                                                                                                                                                  |        string        |    N     |
 | `snapshot_ttl`            | The period of time that a model snapshot not a part of any environment should exist before being deleted. This is defined as a string with the default `in 1 week`. Other [relative dates](https://dateparser.readthedocs.io/en/latest/) can be used, such as `in 30 days`. (Default: `in 1 week`) |        string        |    N     |
 | `environment_ttl`         | The period of time that a development environment should exist before being deleted. This is defined as a string with the default `in 1 week`. Other [relative dates](https://dateparser.readthedocs.io/en/latest/) can be used, such as `in 30 days`. (Default: `in 1 week`)                      |        string        |    N     |
 | `ignore_patterns`         | Files that match glob patterns specified in this list are ignored when scanning the project folder (Default: `[]`)                                                                                                                                                                                |     list[string]     |    N     |

sqlmesh/core/config/root.py

@@ -28,7 +28,6 @@ class Config(BaseConfig):
         default_gateway: The default gateway.
         notification_targets: The notification targets to use.
         dialect: The default sql dialect of model queries. Default: same as engine dialect.
-        physical_schema: The default schema used to store materialized tables.
         project: The project name of this config. Used for multi-repo setups.
         snapshot_ttl: The period of time that a model snapshot that is not a part of any environment should exist before being deleted.
         environment_ttl: The period of time that a development environment should exist before being deleted.
@@ -47,7 +46,6 @@ class Config(BaseConfig):
     default_scheduler: SchedulerConfig = BuiltInSchedulerConfig()
     default_gateway: str = ""
     notification_targets: t.List[NotificationTarget] = []
-    physical_schema: str = c.SQLMESH
     project: str = ""
     snapshot_ttl: str = c.DEFAULT_SNAPSHOT_TTL
     environment_ttl: t.Optional[str] = c.DEFAULT_ENVIRONMENT_TTL

sqlmesh/core/constants.py

@@ -38,3 +38,6 @@
 TESTS = "tests"
 CACHE = ".cache"
 SCHEMA_YAML = "schema.yaml"
+
+
+DEFAULT_SCHEMA = "default"

sqlmesh/core/context.py

@@ -432,12 +432,10 @@ def snapshots(self) -> t.Dict[str, Snapshot]:
         for model in models.values():
             if model.name in remote_snapshots:
                 snapshot = remote_snapshots[model.name]
-                physical_schema = snapshot.physical_schema
                 ttl = snapshot.ttl
                 project = snapshot.project
             else:
                 config = self.config_for_model(model)
-                physical_schema = config.physical_schema
                 ttl = config.snapshot_ttl
                 project = config.project
 
@@ -446,7 +444,6 @@ def snapshots(self) -> t.Dict[str, Snapshot]:
                 models=models,
                 audits=audits,
                 cache=fingerprint_cache,
-                physical_schema=physical_schema,
                 ttl=ttl,
                 project=project,
             )

sqlmesh/core/snapshot/definition.py

@@ -7,7 +7,7 @@
 from datetime import datetime
 from enum import IntEnum
 
-from pydantic import validator
+from pydantic import Field, validator
 from sqlglot import exp
 from sqlglot.helper import seq_get
 
@@ -132,10 +132,17 @@ class SnapshotDataVersion(PydanticModel, frozen=True):
     version: str
     temp_version: t.Optional[str]
     change_category: t.Optional[SnapshotChangeCategory]
+    physical_schema_: t.Optional[str] = Field(default=None, alias="physical_schema")
 
     def snapshot_id(self, name: str) -> SnapshotId:
         return SnapshotId(name=name, identifier=self.fingerprint.to_identifier())
 
+    @property
+    def physical_schema(self) -> str:
+        # The physical schema here is optional to maintain backwards compatibility with
+        # records stored by previous versions of SQLMesh.
+        return self.physical_schema_ or c.SQLMESH
+
     @property
     def data_version(self) -> SnapshotDataVersion:
         return self
@@ -163,7 +170,7 @@ def for_environment(self, environment: str) -> str:
         )
 
     def schema_for_environment(self, environment: str) -> str:
-        schema = self.schema_name or "default"
+        schema = self.schema_name or c.DEFAULT_SCHEMA
         if environment.lower() != c.PROD:
             schema = f"{schema}__{environment}"
         return schema
@@ -174,7 +181,6 @@ class SnapshotInfoMixin(ModelKindMixin):
     temp_version: t.Optional[str]
     change_category: t.Optional[SnapshotChangeCategory]
     fingerprint: SnapshotFingerprint
-    physical_schema: str
     previous_versions: t.Tuple[SnapshotDataVersion, ...] = ()
 
     def is_temporary_table(self, is_dev: bool) -> bool:
@@ -193,7 +199,7 @@ def snapshot_id(self) -> SnapshotId:
 
     @property
     def qualified_view_name(self) -> QualifiedViewName:
-        (catalog, schema, table) = parse_model_name(self.name)
+        catalog, schema, table = parse_model_name(self.name)
         return QualifiedViewName(catalog=catalog, schema_name=schema, table=table)
 
     @property
@@ -203,6 +209,10 @@ def previous_version(self) -> t.Optional[SnapshotDataVersion]:
             return self.previous_versions[-1]
         return None
 
+    @property
+    def physical_schema(self) -> str:
+        raise NotImplementedError
+
     @property
     def data_version(self) -> SnapshotDataVersion:
         raise NotImplementedError
@@ -262,7 +272,7 @@ class SnapshotTableInfo(PydanticModel, SnapshotInfoMixin, frozen=True):
     fingerprint: SnapshotFingerprint
     version: str
     temp_version: t.Optional[str]
-    physical_schema: str
+    physical_schema_: str = Field(alias="physical_schema")
     parents: t.Tuple[SnapshotId, ...]
     previous_versions: t.Tuple[SnapshotDataVersion, ...] = ()
     change_category: t.Optional[SnapshotChangeCategory]
@@ -277,6 +287,10 @@ def table_name(self, is_dev: bool = False, for_read: bool = False) -> str:
         """
         return self._table_name(self.version, is_dev, for_read)
 
+    @property
+    def physical_schema(self) -> str:
+        return self.physical_schema_
+
     @property
     def table_info(self) -> SnapshotTableInfo:
         """Helper method to return self."""
@@ -289,6 +303,7 @@ def data_version(self) -> SnapshotDataVersion:
             version=self.version,
             temp_version=self.temp_version,
             change_category=self.change_category,
+            physical_schema=self.physical_schema,
         )
 
     @property
@@ -342,7 +357,7 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
 
     name: str
     fingerprint: SnapshotFingerprint
-    physical_schema: str
+    physical_schema_: t.Optional[str] = Field(default=None, alias="physical_schema")
     model: Model
     parents: t.Tuple[SnapshotId, ...]
     audits: t.Tuple[Audit, ...]
@@ -442,7 +457,6 @@ def from_model(
         model: Model,
         *,
         models: t.Dict[str, Model],
-        physical_schema: str = c.SQLMESH,
         ttl: str = c.DEFAULT_SNAPSHOT_TTL,
         project: str = "",
         version: t.Optional[str] = None,
@@ -472,19 +486,16 @@ def from_model(
             name=model.name,
             fingerprint=fingerprint_from_model(
                 model,
-                physical_schema=physical_schema,
                 models=models,
                 audits=audits,
                 cache=cache,
             ),
-            physical_schema=physical_schema,
             model=model,
             parents=tuple(
                 SnapshotId(
                     name=name,
                     identifier=fingerprint_from_model(
                         models[name],
-                        physical_schema=physical_schema,
                         models=models,
                         audits=audits,
                         cache=cache,
@@ -672,6 +683,7 @@ def categorize_as(self, category: SnapshotChangeCategory) -> None:
         )
         if is_forward_only and self.previous_version:
             self.version = self.previous_version.data_version.version
+            self.physical_schema_ = self.previous_version.physical_schema
         else:
             self.version = self.fingerprint.to_version()
 
@@ -716,6 +728,15 @@ def version_get_or_generate(self) -> str:
         """Helper method to get the version or generate it from the fingerprint."""
         return self.version or self.fingerprint.to_version()
 
+    @property
+    def physical_schema(self) -> str:
+        if self.physical_schema_ is not None:
+            return self.physical_schema_
+        _, schema, _ = parse_model_name(self.name)
+        if schema is None:
+            schema = c.DEFAULT_SCHEMA
+        return f"{c.SQLMESH}__{schema}"
+
     @property
     def table_info(self) -> SnapshotTableInfo:
         """Helper method to get the SnapshotTableInfo from the Snapshot."""
@@ -740,6 +761,7 @@ def data_version(self) -> SnapshotDataVersion:
             version=self.version,
             temp_version=self.temp_version,
             change_category=self.change_category,
+            physical_schema=self.physical_schema,
         )
 
     @property
@@ -803,7 +825,6 @@ def fingerprint_from_model(
     model: Model,
     *,
     models: t.Dict[str, Model],
-    physical_schema: str = "",
     audits: t.Optional[t.Dict[str, Audit]] = None,
     cache: t.Optional[t.Dict[str, SnapshotFingerprint]] = None,
 ) -> SnapshotFingerprint:
@@ -815,7 +836,6 @@ def fingerprint_from_model(
 
     Args:
         model: Model to fingerprint.
-        physical_schema: The physical_schema of the snapshot which represents where it is stored.
         models: Dictionary of all models in the graph to make the fingerprint dependent on parent changes.
             If no dictionary is passed in the fingerprint will not be dependent on a model's parents.
         audits: Available audits by name.
@@ -831,7 +851,6 @@ def fingerprint_from_model(
             fingerprint_from_model(
                 models[table],
                 models=models,
-                physical_schema=physical_schema,
                 audits=audits,
                 cache=cache,
             )
@@ -846,7 +865,7 @@ def fingerprint_from_model(
         )
 
         cache[model.name] = SnapshotFingerprint(
-            data_hash=_model_data_hash(model, physical_schema),
+            data_hash=_model_data_hash(model),
             metadata_hash=_model_metadata_hash(model, audits or {}),
             parent_data_hash=parent_data_hash,
             parent_metadata_hash=parent_metadata_hash,
@@ -855,7 +874,7 @@ def fingerprint_from_model(
     return cache[model.name]
 
 
-def _model_data_hash(model: Model, physical_schema: str) -> str:
+def _model_data_hash(model: Model) -> str:
     def serialize_hooks(hooks: t.List[HookCall]) -> t.Iterable[str]:
         serialized = []
         for hook in hooks:
@@ -875,7 +894,6 @@ def serialize_hooks(hooks: t.List[HookCall]) -> t.Iterable[str]:
         model.cron,
         model.storage_format,
         str(model.lookback),
-        physical_schema,
         *(model.partitioned_by or []),
         *(expression.sql(comments=False) for expression in model.expressions or []),
         *serialize_hooks(model.pre),

sqlmesh/core/state_sync/engine_adapter.py

@@ -652,7 +652,6 @@ def _migrate_rows(self) -> None:
 
             new_snapshot.fingerprint = fingerprint_from_model(
                 model,
-                physical_schema=snapshot.physical_schema,
                 models=models,
                 audits=audits,
             )
@@ -661,7 +660,6 @@ def _migrate_rows(self) -> None:
                     name=name,
                     identifier=fingerprint_from_model(
                         models[name],
-                        physical_schema=snapshot.physical_schema,
                         models=models,
                         audits=audits,
                         cache=fingerprint_cache,

tests/conftest.py

@@ -132,7 +132,6 @@ def _make_function(model: Model, version: t.Optional[str] = None, **kwargs) -> S
         return Snapshot.from_model(
             model,
             **{  # type: ignore
-                "physical_schema": "sqlmesh",
                 "models": {},
                 "ttl": "in 1 week",
                 **kwargs,

tests/core/test_config.py

@@ -152,11 +152,11 @@ def test_load_config_from_paths(yaml_config_path: Path, python_config_path: Path
 def test_load_config_multiple_config_files_in_folder(tmp_path):
     config_a_path = tmp_path / "config.yaml"
     with open(config_a_path, "w") as fd:
-        fd.write("physical_schema: schema_a")
+        fd.write("project: project_a")
 
     config_b_path = tmp_path / "config.yml"
     with open(config_b_path, "w") as fd:
-        fd.write("physical_schema: schema_b")
+        fd.write("project: project_b")
 
     with pytest.raises(ConfigError, match=r"^Multiple configuration files found in folder.*"):
         load_config_from_paths(config_a_path, config_b_path)

tests/core/test_context.py

@@ -19,7 +19,7 @@
 def test_global_config():
     context = Context(paths="examples/sushi")
     assert context.config.dialect is None
-    assert context.config.physical_schema == "sqlmesh"
+    assert context.config.time_column_format == "%Y-%m-%d"
 
 
 def test_named_config():
@@ -41,10 +41,10 @@ def test_missing_named_config():
 
 
 def test_config_parameter():
-    config = Config(model_defaults=ModelDefaultsConfig(dialect="presto"), physical_schema="dev")
+    config = Config(model_defaults=ModelDefaultsConfig(dialect="presto"), project="test_project")
     context = Context(paths="examples/sushi", config=config)
     assert context.config.dialect == "presto"
-    assert context.config.physical_schema == "dev"
+    assert context.config.project == "test_project"
 
 
 def test_config_not_found():
@@ -139,10 +139,10 @@ def test_render(sushi_context, assert_exp_eq):
             CAST(o.waiter_id AS INT) AS waiter_id, /* Waiter id */
             CAST(SUM(oi.quantity * i.price) AS DOUBLE) AS revenue, /* Revenue from orders taken by this waiter */
             CAST(o.ds AS TEXT) AS ds /* Date */
-          FROM sqlmesh.sushi__orders__2048186253 AS o
-          LEFT JOIN sqlmesh.sushi__order_items__3982722653 AS oi
+          FROM sqlmesh__sushi.sushi__orders__619968963 AS o
+          LEFT JOIN sqlmesh__sushi.sushi__order_items__3090566586 AS oi
             ON o.ds = oi.ds AND o.id = oi.order_id
-          LEFT JOIN sqlmesh.sushi__items__3758356954 AS i
+          LEFT JOIN sqlmesh__sushi.sushi__items__1837306384 AS i
             ON oi.ds = i.ds AND oi.item_id = i.id
           WHERE
             o.ds <= '2021-01-01' AND o.ds >= '2021-01-01'

tests/core/test_scheduler.py

@@ -124,7 +124,7 @@ def test_run(sushi_context_fixed_date: Context, scheduler: Scheduler):
     assert (
         adapter.fetchone(
             f"""
-        SELECT id, name, price FROM sqlmesh.sushi__items__{snapshot.version} ORDER BY ds LIMIT 1
+        SELECT id, name, price FROM sqlmesh__sushi.sushi__items__{snapshot.version} ORDER BY ds LIMIT 1
     """
         )
         == (0, "Hotate", 5.99)