SQLMesh
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
@@ -2075,12 +2075,13 @@ def _create_model(
 
     physical_schema_mapping = physical_schema_mapping or {}
     model_schema_name = exp.to_table(name, dialect=dialect).db
-    physical_schema_override: t.Optional[str] = None
+    physical_schema_override: t.Optional[str] = kwargs.pop("physical_schema_override", None)
 
-    for re_pattern, override_schema in physical_schema_mapping.items():
-        if re.match(re_pattern, model_schema_name):
-            physical_schema_override = override_schema
-            break
+    if not physical_schema_override:
+        for re_pattern, override_schema in physical_schema_mapping.items():
+            if re.match(re_pattern, model_schema_name):
+                physical_schema_override = override_schema
+                break
 
     raw_kind = kwargs.pop("kind", None)
     if raw_kind:

diff --git a/sqlmesh/core/model/meta.py b/sqlmesh/core/model/meta.py
@@ -166,8 +166,8 @@ def _dialect_validator(cls, v: t.Any) -> t.Optional[str]:
         dialect = str_or_exp_to_str(v)
         return dialect and dialect.lower()
 
-    @field_validator("physical_version", mode="before")
-    def _physical_version_validator(cls, v: t.Any) -> t.Optional[str]:
+    @field_validator("physical_version", "physical_schema_override", mode="before")
+    def _physical_version_and_schema_validator(cls, v: t.Any) -> t.Optional[str]:
         if v is None:
             return v
         return str_or_exp_to_str(v)

diff --git a/tests/core/test_context.py b/tests/core/test_context.py
@@ -755,6 +755,13 @@ def test_physical_schema_mapping(tmp_path: pathlib.Path) -> None:
         "MODEL(name untouched.model_c); SELECT 1;",
     )
 
+    # a physical_schema_override at a model level takes precedence
+    create_temp_file(
+        tmp_path,
+        pathlib.Path(pathlib.Path("models"), "d.sql"),
+        "MODEL(name testone.model_d, physical_schema_override testing_critical); SELECT 1;",
+    )
+
     ctx = Context(
         config=Config(
             model_defaults=ModelDefaultsConfig(dialect="duckdb"),
@@ -776,9 +783,14 @@ def test_physical_schema_mapping(tmp_path: pathlib.Path) -> None:
         snapshot.qualified_view_name.schema_name for snapshot in sorted(ctx.snapshots.values())
     ]
 
-    assert len(physical_schemas) == len(view_schemas) == 3
-    assert physical_schemas == ["overridden_staging", "testing", "sqlmesh__untouched"]
-    assert view_schemas == ["foo_staging", "testone", "untouched"]
+    assert len(view_schemas) == len(physical_schemas) == 4
+    assert physical_schemas == [
+        "overridden_staging",
+        "testing",
+        "testing_critical",
+        "sqlmesh__untouched",
+    ]
+    assert view_schemas == ["foo_staging", "testone", "testone", "untouched"]
 
 
 @pytest.mark.slow

diff --git a/tests/core/test_model.py b/tests/core/test_model.py
@@ -6303,6 +6303,27 @@ def test_physical_version():
         ).validate_definition()
 
 
+def test_physical_schema():
+    # in conjunction with physical_version, you can have sqlmesh manage an existing table without moving it
+    # or using a dedicated schema just for the physical_schema_mapping
+    expressions = d.parse(
+        """
+        MODEL (
+            name db.table,
+            kind INCREMENTAL_BY_TIME_RANGE(
+                time_column a
+            ),
+            physical_schema_override db
+        );
+
+        SELECT a, b
+        """
+    )
+
+    model = load_sql_based_model(expressions)
+    assert model.physical_schema == "db"
+
+
 def test_trailing_comments():
     expressions = d.parse(
         """