feat: extend ownership control to physical layer tables

Add `physical_owner` field to `OwnershipConfig` so that SQLMesh__*
physical tables get ownership applied at creation time, not just
virtual-layer views and schemas.

* `OwnershipConfig.physical_owner` - optional plain string, no env-pattern matching needed
* `EngineAdapterBase.alter_table_owner` - no-op default
* `SparkEngineAdapter.alter_table_owner` - ALTER TABLE ... OWNER TO ... with backtick-quoting
* `SnapshotEvaluator.create_snapshot` - calls alter_table_owner after _execute_create (skips ViewKind)
* `SnapshotEvaluator.create` / `create_physical_schemas` - accept and thread owner param
* `BuiltInPlanEvaluator` - resolves physical_owner in both PhysicalLayerSchemaCreation and PhysicalLayerUpdate stages

Signed-off-by: Gabe Pesco <gabe.pesco@milliman.com>
Signed-off-by: Gabe Pesco <PescoG@medinsight.milliman.com>

Author: Gabe Pesco

sqlmesh/core/config/ownership.py

@@ -30,9 +30,11 @@ class OwnershipConfig(BaseConfig):
           environment_owner_mapping:
             "^prod$": "svc_prod_spn"
             ".*": "group:shared-developers"
+          physical_owner: "group:shared-developers"
     """
 
     environment_owner_mapping: OwnershipMapping = {}
+    physical_owner: t.Optional[str] = None
 
     def resolve_owner(self, environment_name: str) -> t.Optional[str]:
         """Return the configured owner for the given environment name, or None."""

sqlmesh/core/engine_adapter/base.py

@@ -1432,6 +1432,13 @@ def alter_view_owner(self, view_name: TableName, owner: str) -> None:
         (e.g. Spark/Databricks Unity Catalog: ALTER VIEW ... OWNER TO ...).
         """
 
+    def alter_table_owner(self, table_name: TableName, owner: str) -> None:
+        """Set the owner of a table.
+
+        No-op by default. Override in dialect-specific adapters that support ownership control
+        (e.g. Spark/Databricks Unity Catalog: ALTER TABLE ... OWNER TO ...).
+        """
+
     def drop_schema(
         self,
         schema_name: SchemaName,

sqlmesh/core/engine_adapter/spark.py

@@ -567,6 +567,13 @@ def alter_view_owner(self, view_name: TableName, owner: str) -> None:
         owner_sql = exp.to_identifier(owner, quoted=True).sql(dialect=self.dialect)
         self.execute(f"ALTER VIEW {view_sql} OWNER TO {owner_sql}")
 
+    def alter_table_owner(self, table_name: TableName, owner: str) -> None:
+        table_sql = exp.to_table(table_name, dialect=self.dialect).sql(
+            dialect=self.dialect, identify=True
+        )
+        owner_sql = exp.to_identifier(owner, quoted=True).sql(dialect=self.dialect)
+        self.execute(f"ALTER TABLE {table_sql} OWNER TO {owner_sql}")
+
     @classmethod
     def _wap_branch_name(cls, wap_id: str) -> str:
         return f"{cls.WAP_PREFIX}{wap_id}"

sqlmesh/core/plan/evaluator.py

@@ -175,6 +175,7 @@ def visit_physical_layer_update_stage(
             self.console.log_success(skip_message)
             return
 
+        physical_owner = self.ownership_config.physical_owner if self.ownership_config else None
         completion_status = None
         progress_stopped = False
         try:
@@ -188,6 +189,7 @@ def visit_physical_layer_update_stage(
                     x, plan.environment, self.default_catalog
                 ),
                 on_complete=self.console.update_creation_progress,
+                owner=physical_owner,
             )
             if completion_status.is_nothing_to_do:
                 self.console.log_success(skip_message)
@@ -212,9 +214,10 @@ def visit_physical_layer_update_stage(
     def visit_physical_layer_schema_creation_stage(
         self, stage: stages.PhysicalLayerSchemaCreationStage, plan: EvaluatablePlan
     ) -> None:
+        physical_owner = self.ownership_config.physical_owner if self.ownership_config else None
         try:
             self.snapshot_evaluator.create_physical_schemas(
-                stage.snapshots, stage.deployability_index
+                stage.snapshots, stage.deployability_index, owner=physical_owner
             )
         except Exception as ex:
             raise PlanError("Plan application failed.") from ex

sqlmesh/core/snapshot/evaluator.py

@@ -368,6 +368,7 @@ def create(
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
         allow_destructive_snapshots: t.Optional[t.Set[str]] = None,
         allow_additive_snapshots: t.Optional[t.Set[str]] = None,
+        owner: t.Optional[str] = None,
     ) -> CompletionStatus:
         """Creates a physical snapshot schema and table for the given collection of snapshots.
 
@@ -379,6 +380,7 @@ def create(
             on_complete: A callback to call on each successfully created snapshot.
             allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
             allow_additive_snapshots: Set of snapshots that are allowed to have additive schema changes.
+            owner: Optional principal to set as table owner after creation.
 
         Returns:
             CompletionStatus: The status of the creation operation (success, failure, nothing to do).
@@ -398,17 +400,22 @@ def create(
             on_complete=on_complete,
             allow_destructive_snapshots=allow_destructive_snapshots or set(),
             allow_additive_snapshots=allow_additive_snapshots or set(),
+            owner=owner,
         )
         return CompletionStatus.SUCCESS
 
     def create_physical_schemas(
-        self, snapshots: t.Iterable[Snapshot], deployability_index: DeployabilityIndex
+        self,
+        snapshots: t.Iterable[Snapshot],
+        deployability_index: DeployabilityIndex,
+        owner: t.Optional[str] = None,
     ) -> None:
         """Creates the physical schemas for the given snapshots.
 
         Args:
             snapshots: Snapshots to create physical schemas for.
             deployability_index: Determines snapshots that are deployable in the context of this creation.
+            owner: Optional principal to set as schema owner after creation.
         """
         tables_by_gateway: t.Dict[t.Optional[str], t.List[str]] = defaultdict(list)
         for snapshot in snapshots:
@@ -420,7 +427,7 @@ def create_physical_schemas(
         gateway_table_pairs = [
             (gateway, table) for gateway, tables in tables_by_gateway.items() for table in tables
         ]
-        self._create_schemas(gateway_table_pairs=gateway_table_pairs)
+        self._create_schemas(gateway_table_pairs=gateway_table_pairs, owner=owner)
 
     def get_snapshots_to_create(
         self, target_snapshots: t.Iterable[Snapshot], deployability_index: DeployabilityIndex
@@ -453,6 +460,7 @@ def _create_snapshots(
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
         allow_destructive_snapshots: t.Set[str],
         allow_additive_snapshots: t.Set[str],
+        owner: t.Optional[str] = None,
     ) -> None:
         """Internal method to create tables in parallel."""
         with self.concurrent_context():
@@ -465,6 +473,7 @@ def _create_snapshots(
                     allow_destructive_snapshots=allow_destructive_snapshots,
                     allow_additive_snapshots=allow_additive_snapshots,
                     on_complete=on_complete,
+                    owner=owner,
                 ),
                 self.ddl_concurrent_tasks,
                 raise_on_error=False,
@@ -870,6 +879,7 @@ def create_snapshot(
         allow_destructive_snapshots: t.Set[str],
         allow_additive_snapshots: t.Set[str],
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
+        owner: t.Optional[str] = None,
     ) -> None:
         """Creates a physical table for the given snapshot.
 
@@ -880,6 +890,7 @@ def create_snapshot(
             on_complete: A callback to call on each successfully created database object.
             allow_destructive_snapshots: Snapshots for which destructive schema changes are allowed.
             allow_additive_snapshots: Snapshots for which additive schema changes are allowed.
+            owner: Optional principal to set as table owner after creation.
         """
         if not snapshot.is_model:
             return
@@ -907,6 +918,9 @@ def create_snapshot(
                 **create_render_kwargs
             )
 
+            is_table_deployable = deployability_index.is_deployable(snapshot)
+            table_name = snapshot.table_name(is_deployable=is_table_deployable)
+
             if self._can_clone(snapshot, deployability_index):
                 self._clone_snapshot_in_dev(
                     snapshot=snapshot,
@@ -919,17 +933,19 @@ def create_snapshot(
                     run_pre_post_statements=True,
                 )
             else:
-                is_table_deployable = deployability_index.is_deployable(snapshot)
                 self._execute_create(
                     snapshot=snapshot,
-                    table_name=snapshot.table_name(is_deployable=is_table_deployable),
+                    table_name=table_name,
                     is_table_deployable=is_table_deployable,
                     deployability_index=deployability_index,
                     create_render_kwargs=create_render_kwargs,
                     rendered_physical_properties=rendered_physical_properties,
                     dry_run=True,
                 )
 
+        if owner and not isinstance(snapshot.model.kind, ViewKind):
+            adapter.alter_table_owner(table_name, owner)
+
         evaluation_strategy.run_post_statements(
             snapshot=snapshot, render_kwargs={**create_render_kwargs, "inside_transaction": False}
         )

tests/core/engine_adapter/test_spark.py

@@ -1125,6 +1125,23 @@ def test_alter_view_owner_special_chars_in_principal(make_mocked_engine_adapter:
     ]
 
 
+def test_alter_table_owner(make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(SparkEngineAdapter)
+    adapter.alter_table_owner("catalog.sqlmesh__sushi.orders__abc123", "svc_prod_spn")
+    assert to_sql_calls(adapter) == [
+        "ALTER TABLE `catalog`.`sqlmesh__sushi`.`orders__abc123` OWNER TO `svc_prod_spn`"
+    ]
+
+
+def test_alter_table_owner_special_chars_in_principal(make_mocked_engine_adapter: t.Callable):
+    # Databricks Unity Catalog principals can contain colons and @ signs.
+    adapter = make_mocked_engine_adapter(SparkEngineAdapter)
+    adapter.alter_table_owner("catalog.sqlmesh__sushi.orders__abc123", "group:data@company.com")
+    assert to_sql_calls(adapter) == [
+        "ALTER TABLE `catalog`.`sqlmesh__sushi`.`orders__abc123` OWNER TO `group:data@company.com`"
+    ]
+
+
 def test_alter_schema_owner_base_noop(make_mocked_engine_adapter: t.Callable):
     # The base EngineAdapter.alter_schema_owner is a no-op: adapters that don't
     # support ownership control silently skip it without emitting any SQL.
@@ -1133,6 +1150,7 @@ def test_alter_schema_owner_base_noop(make_mocked_engine_adapter: t.Callable):
     adapter = make_mocked_engine_adapter(DuckDBEngineAdapter)
     adapter.alter_schema_owner("my_schema", "some_owner")
     adapter.alter_view_owner("my_schema.my_view", "some_owner")
+    adapter.alter_table_owner("my_schema.my_table", "some_owner")
     # No ALTER SQL should have been emitted
     alter_calls = [s for s in to_sql_calls(adapter) if "OWNER" in s.upper()]
     assert alter_calls == []

tests/core/test_config.py

@@ -713,6 +713,28 @@ def test_config_ownership_defaults_to_empty():
     assert attach_config_2.read_only is True
 
 
+def test_ownership_config_physical_owner():
+    # physical_owner is a simple optional string — no pattern matching.
+    config = OwnershipConfig(physical_owner="group:data-platform")
+    assert config.physical_owner == "group:data-platform"
+
+
+def test_ownership_config_physical_owner_default_none():
+    assert OwnershipConfig().physical_owner is None
+
+
+def test_ownership_config_physical_owner_deserialization():
+    config = Config(
+        model_defaults=ModelDefaultsConfig(dialect="duckdb"),
+        ownership={
+            "environment_owner_mapping": {"^prod$": "svc_prod"},
+            "physical_owner": "group:data-platform",
+        },
+    )
+    assert config.ownership.physical_owner == "group:data-platform"
+    assert config.ownership.resolve_owner("prod") == "svc_prod"
+
+
 def test_load_model_defaults_audits(tmp_path):
     config_path = tmp_path / "config_model_defaults_audits.yaml"
     with open(config_path, "w", encoding="utf-8") as fd:

tests/core/test_snapshot_evaluator.py

@@ -504,6 +504,88 @@ def test_promote_owner_applied_per_view(mocker: MockerFixture, adapter_mock, mak
     assert called_owners == {"svc_prod_spn"}
 
 
+def test_create_with_physical_owner(mocker: MockerFixture, adapter_mock, make_snapshot):
+    """alter_table_owner is called for each non-view table when physical owner is set."""
+    adapter_mock.get_data_objects.return_value = []
+    evaluator = SnapshotEvaluator(adapter_mock)
+
+    model = SqlModel(
+        name="test_schema.test_model",
+        kind=IncrementalByTimeRangeKind(time_column="ds"),
+        storage_format="parquet",
+        query=parse_one("SELECT a, ds FROM tbl WHERE ds BETWEEN @start_ds AND @end_ds"),
+    )
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    evaluator.create([snapshot], {}, owner="group:data-platform")
+
+    adapter_mock.alter_table_owner.assert_called_once()
+    call_args = adapter_mock.alter_table_owner.call_args
+    assert call_args.args[1] == "group:data-platform"
+
+
+def test_create_without_physical_owner_skips_alter(
+    mocker: MockerFixture, adapter_mock, make_snapshot
+):
+    """When no physical owner is set, alter_table_owner is never called."""
+    adapter_mock.get_data_objects.return_value = []
+    evaluator = SnapshotEvaluator(adapter_mock)
+
+    model = SqlModel(
+        name="test_schema.test_model",
+        kind=IncrementalByTimeRangeKind(time_column="ds"),
+        storage_format="parquet",
+        query=parse_one("SELECT a, ds FROM tbl WHERE ds BETWEEN @start_ds AND @end_ds"),
+    )
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    evaluator.create([snapshot], {})
+
+    adapter_mock.alter_table_owner.assert_not_called()
+
+
+def test_create_view_kind_skips_physical_owner(
+    mocker: MockerFixture, adapter_mock, make_snapshot
+):
+    """ViewKind snapshots skip alter_table_owner even when physical_owner is set."""
+    adapter_mock.get_data_objects.return_value = []
+    evaluator = SnapshotEvaluator(adapter_mock)
+
+    model = SqlModel(
+        name="test_schema.test_view",
+        kind=ViewKind(),
+        query=parse_one("SELECT 1"),
+    )
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    evaluator.create([snapshot], {}, owner="group:data-platform")
+
+    adapter_mock.alter_table_owner.assert_not_called()
+
+
+def test_create_physical_schemas_with_owner(mocker: MockerFixture, adapter_mock, make_snapshot):
+    """create_physical_schemas passes owner to _create_schemas so alter_schema_owner is called."""
+    evaluator = SnapshotEvaluator(adapter_mock)
+    deployability_index = DeployabilityIndex.all_deployable()
+
+    model = SqlModel(
+        name="test_schema.test_model",
+        kind=IncrementalByTimeRangeKind(time_column="ds"),
+        storage_format="parquet",
+        query=parse_one("SELECT a, ds FROM tbl WHERE ds BETWEEN @start_ds AND @end_ds"),
+    )
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    evaluator.create_physical_schemas([snapshot], deployability_index, owner="svc_prod_spn")
+
+    adapter_mock.alter_schema_owner.assert_called_once()
+    assert adapter_mock.alter_schema_owner.call_args.args[1] == "svc_prod_spn"
+
+
 def test_cleanup(mocker: MockerFixture, adapter_mock, make_snapshot):
     evaluator = SnapshotEvaluator(adapter_mock)