diff --git a/NEXT_CHANGELOG.md b/NEXT_CHANGELOG.md
index 85a50d00b9..7088cf7aa1 100644
--- a/NEXT_CHANGELOG.md
+++ b/NEXT_CHANGELOG.md
@@ -11,6 +11,7 @@
 * engine/direct: Fix 400 error when deploying grants with ALL_PRIVILEGES ([#4801](https://github.com/databricks/cli/pull/4801))
 * Deduplicate grant entries with duplicate principals or privileges during initialization ([#4801](https://github.com/databricks/cli/pull/4801))
 * engine/direct: Fix unwanted recreation of secret scopes when scope_backend_type is not set ([#4834](https://github.com/databricks/cli/pull/4834))
+* engine/direct: Fix deploying removed principals ([#4824](https://github.com/databricks/cli/pull/4824))
 
 ### Dependency updates
 
diff --git a/acceptance/bin/gron.py b/acceptance/bin/gron.py
index 9498f27f19..df7e36bd6e 100755
--- a/acceptance/bin/gron.py
+++ b/acceptance/bin/gron.py
@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+import argparse
 import json
 import sys
 from pathlib import Path
@@ -7,7 +8,7 @@
 from print_requests import read_json_many
 
 
-def gron(obj, path="json"):
+def gron(obj, path="json", noindex=False):
     """Flatten JSON into greppable assignments.
 
     The path parameter defaults to "json" to match the original gron tool,
@@ -26,6 +27,10 @@ def gron(obj, path="json"):
     json.items[0] = "apple";
     json.items[1] = "banana";
 
+    >>> gron({"items": ["apple", "banana"]}, noindex=True)
+    json.items[] = "apple";
+    json.items[] = "banana";
+
     >>> gron({"tasks": [{"libraries": [{"whl": "file.whl"}]}]})
     json.tasks[0].libraries[0].whl = "file.whl";
 
@@ -38,31 +43,34 @@ def gron(obj, path="json"):
             print(f"{path} = {{}};")
         else:
             for key in obj:
-                gron(obj[key], f"{path}.{key}")
+                gron(obj[key], f"{path}.{key}", noindex=noindex)
     elif isinstance(obj, list):
         if not obj:
             print(f"{path} = [];")
         else:
             for i, item in enumerate(obj):
-                gron(item, f"{path}[{i}]")
+                index = "[]" if noindex else f"[{i}]"
+                gron(item, f"{path}{index}", noindex=noindex)
     else:
         print(f"{path} = {json.dumps(obj)};")
 
 
 def main():
-    if len(sys.argv) > 1:
-        with open(sys.argv[1]) as f:
-            content = f.read()
-        data = read_json_many(content)
-        if len(data) == 1:
-            data = data[0]
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--noindex", action="store_true")
+    parser.add_argument("file", nargs="?")
+    args = parser.parse_args()
+
+    if args.file:
+        content = Path(args.file).read_text()
     else:
         content = sys.stdin.read()
-        data = read_json_many(content)
-        if len(data) == 1:
-            data = data[0]
 
-    gron(data)
+    data = read_json_many(content)
+    if len(data) == 1:
+        data = data[0]
+
+    gron(data, noindex=args.noindex)
 
 
 if __name__ == "__main__":
diff --git a/acceptance/bundle/migrate/grants/out.original_state.json b/acceptance/bundle/migrate/grants/out.original_state.json
index d5f8508854..742398134b 100644
--- a/acceptance/bundle/migrate/grants/out.original_state.json
+++ b/acceptance/bundle/migrate/grants/out.original_state.json
@@ -180,15 +180,15 @@
           "attributes": {
             "catalog_name": "main",
             "comment": null,
-            "enable_predictive_optimization": null,
+            "enable_predictive_optimization": "INHERIT",
             "force_destroy": true,
             "id": "main.schema_grants",
-            "metastore_id": null,
+            "metastore_id": "[UUID]",
             "name": "schema_grants",
             "owner": "[USERNAME]",
             "properties": null,
             "provider_config": [],
-            "schema_id": "",
+            "schema_id": "[UUID]",
             "storage_root": null
           },
           "sensitive_attributes": [],
diff --git a/acceptance/bundle/resources/grants/schemas/duplicate_principals/out.plan.direct.json b/acceptance/bundle/resources/grants/schemas/duplicate_principals/out.plan.direct.json
index ce43b2a4d3..a9a5727032 100644
--- a/acceptance/bundle/resources/grants/schemas/duplicate_principals/out.plan.direct.json
+++ b/acceptance/bundle/resources/grants/schemas/duplicate_principals/out.plan.direct.json
@@ -12,10 +12,18 @@
         "catalog_type": "MANAGED_CATALOG",
         "created_at": [UNIX_TIME_MILLIS][0],
         "created_by": "[USERNAME]",
+        "effective_predictive_optimization_flag": {
+          "inherited_from_name": "deco-uc-prod-isolated-aws-us-east-1",
+          "inherited_from_type": "METASTORE",
+          "value": "ENABLE"
+        },
+        "enable_predictive_optimization": "INHERIT",
         "full_name": "main.schema_dup_grants_[UNIQUE_NAME]",
+        "metastore_id": "[UUID]",
         "name": "schema_dup_grants_[UNIQUE_NAME]",
         "owner": "[USERNAME]",
-        "updated_at": [UNIX_TIME_MILLIS][0],
+        "schema_id": "[UUID]",
+        "updated_at": [UNIX_TIME_MILLIS][1],
         "updated_by": "[USERNAME]"
       }
     },
diff --git a/acceptance/bundle/resources/grants/schemas/out_of_band_principal/databricks.yml.tmpl b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/databricks.yml.tmpl
new file mode 100644
index 0000000000..7ab71883d7
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/databricks.yml.tmpl
@@ -0,0 +1,12 @@
+bundle:
+  name: schema-grants-out-of-band-principal-$UNIQUE_NAME
+
+resources:
+  schemas:
+    grants_schema:
+      name: schema_out_of_band_principal_$UNIQUE_NAME
+      catalog_name: main
+      grants:
+        - principal: $CURRENT_USER_NAME
+          privileges:
+            - CREATE_TABLE
diff --git a/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan.direct.json b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan.direct.json
new file mode 100644
index 0000000000..12bfde2659
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan.direct.json
@@ -0,0 +1,83 @@
+{
+  "plan_version": 2,
+  "cli_version": "[DEV_VERSION]",
+  "lineage": "[UUID]",
+  "serial": 1,
+  "plan": {
+    "resources.schemas.grants_schema": {
+      "action": "skip",
+      "remote_state": {
+        "browse_only": false,
+        "catalog_name": "main",
+        "catalog_type": "MANAGED_CATALOG",
+        "created_at": [UNIX_TIME_MILLIS][0],
+        "created_by": "[USERNAME]",
+        "effective_predictive_optimization_flag": {
+          "inherited_from_name": "deco-uc-prod-isolated-aws-us-east-1",
+          "inherited_from_type": "METASTORE",
+          "value": "ENABLE"
+        },
+        "enable_predictive_optimization": "INHERIT",
+        "full_name": "main.schema_out_of_band_principal_[UNIQUE_NAME]",
+        "metastore_id": "[UUID]",
+        "name": "schema_out_of_band_principal_[UNIQUE_NAME]",
+        "owner": "[USERNAME]",
+        "schema_id": "[UUID]",
+        "updated_at": [UNIX_TIME_MILLIS][1],
+        "updated_by": "[USERNAME]"
+      }
+    },
+    "resources.schemas.grants_schema.grants": {
+      "depends_on": [
+        {
+          "node": "resources.schemas.grants_schema",
+          "label": "${resources.schemas.grants_schema.id}"
+        }
+      ],
+      "action": "update",
+      "new_state": {
+        "value": {
+          "securable_type": "schema",
+          "full_name": "main.schema_out_of_band_principal_[UNIQUE_NAME]",
+          "__embed__": [
+            {
+              "principal": "[USERNAME]",
+              "privileges": [
+                "CREATE_TABLE"
+              ]
+            }
+          ]
+        }
+      },
+      "remote_state": {
+        "securable_type": "schema",
+        "full_name": "main.schema_out_of_band_principal_[UNIQUE_NAME]",
+        "__embed__": [
+          {
+            "principal": "[USERNAME]",
+            "privileges": [
+              "CREATE_TABLE"
+            ]
+          },
+          {
+            "principal": "deco-test-user@databricks.com",
+            "privileges": [
+              "USE_SCHEMA"
+            ]
+          }
+        ]
+      },
+      "changes": {
+        "[principal='deco-test-user@databricks.com']": {
+          "action": "update",
+          "remote": {
+            "principal": "deco-test-user@databricks.com",
+            "privileges": [
+              "USE_SCHEMA"
+            ]
+          }
+        }
+      }
+    }
+  }
+}
diff --git a/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan.terraform.json b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan.terraform.json
new file mode 100644
index 0000000000..ea73a95ede
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan.terraform.json
@@ -0,0 +1,11 @@
+{
+  "cli_version": "[DEV_VERSION]",
+  "plan": {
+    "resources.schemas.grants_schema": {
+      "action": "skip"
+    },
+    "resources.schemas.grants_schema.grants": {
+      "action": "update"
+    }
+  }
+}
diff --git a/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan2.direct.json b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan2.direct.json
new file mode 100644
index 0000000000..38758d490a
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan2.direct.json
@@ -0,0 +1,52 @@
+{
+  "plan_version": 2,
+  "cli_version": "[DEV_VERSION]",
+  "lineage": "[UUID]",
+  "serial": 2,
+  "plan": {
+    "resources.schemas.grants_schema": {
+      "action": "skip",
+      "remote_state": {
+        "browse_only": false,
+        "catalog_name": "main",
+        "catalog_type": "MANAGED_CATALOG",
+        "created_at": [UNIX_TIME_MILLIS][0],
+        "created_by": "[USERNAME]",
+        "effective_predictive_optimization_flag": {
+          "inherited_from_name": "deco-uc-prod-isolated-aws-us-east-1",
+          "inherited_from_type": "METASTORE",
+          "value": "ENABLE"
+        },
+        "enable_predictive_optimization": "INHERIT",
+        "full_name": "main.schema_out_of_band_principal_[UNIQUE_NAME]",
+        "metastore_id": "[UUID]",
+        "name": "schema_out_of_band_principal_[UNIQUE_NAME]",
+        "owner": "[USERNAME]",
+        "schema_id": "[UUID]",
+        "updated_at": [UNIX_TIME_MILLIS][1],
+        "updated_by": "[USERNAME]"
+      }
+    },
+    "resources.schemas.grants_schema.grants": {
+      "depends_on": [
+        {
+          "node": "resources.schemas.grants_schema",
+          "label": "${resources.schemas.grants_schema.id}"
+        }
+      ],
+      "action": "skip",
+      "remote_state": {
+        "securable_type": "schema",
+        "full_name": "main.schema_out_of_band_principal_[UNIQUE_NAME]",
+        "__embed__": [
+          {
+            "principal": "[USERNAME]",
+            "privileges": [
+              "CREATE_TABLE"
+            ]
+          }
+        ]
+      }
+    }
+  }
+}
diff --git a/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan2.terraform.json b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan2.terraform.json
new file mode 100644
index 0000000000..9f3c906c0f
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.plan2.terraform.json
@@ -0,0 +1,11 @@
+{
+  "cli_version": "[DEV_VERSION]",
+  "plan": {
+    "resources.schemas.grants_schema": {
+      "action": "skip"
+    },
+    "resources.schemas.grants_schema.grants": {
+      "action": "skip"
+    }
+  }
+}
diff --git a/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.test.toml b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.test.toml
new file mode 100644
index 0000000000..d61c11e25c
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/out.test.toml
@@ -0,0 +1,6 @@
+Local = true
+Cloud = true
+RequiresUnityCatalog = true
+
+[EnvMatrix]
+  DATABRICKS_BUNDLE_ENGINE = ["terraform", "direct"]
diff --git a/acceptance/bundle/resources/grants/schemas/out_of_band_principal/output.txt b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/output.txt
new file mode 100644
index 0000000000..269907e2fd
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/output.txt
@@ -0,0 +1,30 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/schema-grants-out-of-band-principal-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] grants update schema main.schema_out_of_band_principal_[UNIQUE_NAME] --json @update.json
+
+>>> [CLI] bundle plan -o json
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/schema-grants-out-of-band-principal-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle plan -o json
+
+>>> errcode [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.schemas.grants_schema
+
+This action will result in the deletion of the following UC schemas. Any underlying data may be lost:
+  delete resources.schemas.grants_schema
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/schema-grants-out-of-band-principal-[UNIQUE_NAME]/default
+
+Deleting files...
+Destroy complete!
diff --git a/acceptance/bundle/resources/grants/schemas/out_of_band_principal/script b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/script
new file mode 100644
index 0000000000..d3b2d17ed3
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/script
@@ -0,0 +1,17 @@
+SCHEMA_NAME=schema_out_of_band_principal_$UNIQUE_NAME
+SCHEMA_FULL_NAME=main.$SCHEMA_NAME
+
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace errcode $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+trace $CLI grants update schema "$SCHEMA_FULL_NAME" --json @update.json > /dev/null
+$CLI grants get schema "$SCHEMA_FULL_NAME" | gron.py --noindex | sort | contains.py "$CURRENT_USER_NAME" 'deco-test-user@databricks.com' > /dev/null
+trace $CLI bundle plan -o json > out.plan.$DATABRICKS_BUNDLE_ENGINE.json
+trace $CLI bundle deploy
+trace $CLI bundle plan -o json > out.plan2.$DATABRICKS_BUNDLE_ENGINE.json
diff --git a/acceptance/bundle/resources/grants/schemas/out_of_band_principal/update.json b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/update.json
new file mode 100644
index 0000000000..3694568664
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/out_of_band_principal/update.json
@@ -0,0 +1,8 @@
+{
+  "changes": [
+    {
+      "principal": "deco-test-user@databricks.com",
+      "add": ["USE_SCHEMA"]
+    }
+  ]
+}
diff --git a/acceptance/bundle/resources/grants/schemas/remove_principal/databricks.yml.tmpl b/acceptance/bundle/resources/grants/schemas/remove_principal/databricks.yml.tmpl
new file mode 100644
index 0000000000..c24bcd4bef
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/remove_principal/databricks.yml.tmpl
@@ -0,0 +1,13 @@
+bundle:
+  name: schema-grants-remove-principal-$UNIQUE_NAME
+
+resources:
+  schemas:
+    grants_schema:
+      name: schema_remove_principal_$UNIQUE_NAME
+      catalog_name: main
+      grants:
+        - principal: $CURRENT_USER_NAME
+          privileges:
+            - CREATE_TABLE
+        - { principal: deco-test-user@databricks.com, privileges: [USE_SCHEMA] } # TO_REMOVE
diff --git a/acceptance/bundle/resources/grants/schemas/remove_principal/out.grants.direct.txt b/acceptance/bundle/resources/grants/schemas/remove_principal/out.grants.direct.txt
new file mode 100644
index 0000000000..0ef2f8e8b7
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/remove_principal/out.grants.direct.txt
@@ -0,0 +1,2 @@
+json.privilege_assignments[].principal = "[USERNAME]";
+json.privilege_assignments[].privileges[] = "CREATE_TABLE";
diff --git a/acceptance/bundle/resources/grants/schemas/remove_principal/out.grants.terraform.txt b/acceptance/bundle/resources/grants/schemas/remove_principal/out.grants.terraform.txt
new file mode 100644
index 0000000000..0ef2f8e8b7
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/remove_principal/out.grants.terraform.txt
@@ -0,0 +1,2 @@
+json.privilege_assignments[].principal = "[USERNAME]";
+json.privilege_assignments[].privileges[] = "CREATE_TABLE";
diff --git a/acceptance/bundle/resources/grants/schemas/remove_principal/out.test.toml b/acceptance/bundle/resources/grants/schemas/remove_principal/out.test.toml
new file mode 100644
index 0000000000..d61c11e25c
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/remove_principal/out.test.toml
@@ -0,0 +1,6 @@
+Local = true
+Cloud = true
+RequiresUnityCatalog = true
+
+[EnvMatrix]
+  DATABRICKS_BUNDLE_ENGINE = ["terraform", "direct"]
diff --git a/acceptance/bundle/resources/grants/schemas/remove_principal/output.txt b/acceptance/bundle/resources/grants/schemas/remove_principal/output.txt
new file mode 100644
index 0000000000..30e139d4fe
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/remove_principal/output.txt
@@ -0,0 +1,44 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/schema-grants-remove-principal-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] grants get schema main.schema_remove_principal_[UNIQUE_NAME]
+json.privilege_assignments[].principal = "deco-test-user@databricks.com";
+json.privilege_assignments[].principal = "[USERNAME]";
+json.privilege_assignments[].privileges[] = "CREATE_TABLE";
+json.privilege_assignments[].privileges[] = "USE_SCHEMA";
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/schema-grants-remove-principal-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] grants get schema main.schema_remove_principal_[UNIQUE_NAME]
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/schema-grants-remove-principal-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+>>> errcode [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.schemas.grants_schema
+
+This action will result in the deletion of the following UC schemas. Any underlying data may be lost:
+  delete resources.schemas.grants_schema
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/schema-grants-remove-principal-[UNIQUE_NAME]/default
+
+Deleting files...
+Destroy complete!
diff --git a/acceptance/bundle/resources/grants/schemas/remove_principal/script b/acceptance/bundle/resources/grants/schemas/remove_principal/script
new file mode 100644
index 0000000000..9196653c47
--- /dev/null
+++ b/acceptance/bundle/resources/grants/schemas/remove_principal/script
@@ -0,0 +1,18 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace errcode $CLI bundle destroy --auto-approve
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+trace $CLI bundle deploy
+trace $CLI grants get schema main.schema_remove_principal_$UNIQUE_NAME | gron.py --noindex | sort | contains.py "$CURRENT_USER_NAME" 'deco-test-user@databricks.com'
+
+grep -v 'TO_REMOVE' databricks.yml > tmp.yml && mv tmp.yml databricks.yml
+
+trace $CLI bundle deploy
+trace $CLI grants get schema main.schema_remove_principal_$UNIQUE_NAME | gron.py --noindex | sort | contains.py "$CURRENT_USER_NAME" '!deco-test-user@databricks.com' &> out.grants.$DATABRICKS_BUNDLE_ENGINE.txt
+trace $CLI bundle plan
+trace $CLI bundle deploy
+trace $CLI bundle plan
diff --git a/acceptance/bundle/resources/schemas/recreate/output.txt b/acceptance/bundle/resources/schemas/recreate/output.txt
index 571d750e4a..135db4876c 100644
--- a/acceptance/bundle/resources/schemas/recreate/output.txt
+++ b/acceptance/bundle/resources/schemas/recreate/output.txt
@@ -73,9 +73,17 @@ Error: Resource catalog.SchemaInfo not found: main.myschema
   "comment":"COMMENT1",
   "created_at":[UNIX_TIME_MILLIS][0],
   "created_by":"[USERNAME]",
+  "effective_predictive_optimization_flag": {
+    "inherited_from_name":"deco-uc-prod-isolated-aws-us-east-1",
+    "inherited_from_type":"METASTORE",
+    "value":"ENABLE"
+  },
+  "enable_predictive_optimization":"INHERIT",
   "full_name":"newmain.myschema",
+  "metastore_id":"[UUID]",
   "name":"myschema",
   "owner":"[USERNAME]",
+  "schema_id":"[UUID]",
   "updated_at":[UNIX_TIME_MILLIS][0],
   "updated_by":"[USERNAME]"
 }
diff --git a/acceptance/bundle/resources/schemas/update/output.txt b/acceptance/bundle/resources/schemas/update/output.txt
index 84850072ba..d8d35b6bf4 100644
--- a/acceptance/bundle/resources/schemas/update/output.txt
+++ b/acceptance/bundle/resources/schemas/update/output.txt
@@ -5,7 +5,7 @@ Deploying resources...
 Updating deployment state...
 Deployment complete!
 
->>> print_requests
+>>> print_requests.py //unity
 {
   "method": "POST",
   "path": "/api/2.1/unity-catalog/schemas",
@@ -26,14 +26,10 @@ Deploying resources...
 Updating deployment state...
 Deployment complete!
 
->>> print_requests
-{
-  "method": "PATCH",
-  "path": "/api/2.1/unity-catalog/schemas/main.myschema",
-  "body": {
-    "comment": "COMMENT2"
-  }
-}
+>>> print_requests.py //unity
+json.method = "PATCH";
+json.path = "/api/2.1/unity-catalog/schemas/main.myschema";
+json.body.comment = "COMMENT2";
 schemas schema1 id='main.myschema' name='myschema' catalog_name='main' comment='COMMENT2'
 
 === Restore comment to original value and re-deploy
@@ -45,12 +41,8 @@ Deploying resources...
 Updating deployment state...
 Deployment complete!
 
->>> print_requests
-{
-  "method": "PATCH",
-  "path": "/api/2.1/unity-catalog/schemas/main.myschema",
-  "body": {
-    "comment": "COMMENT1"
-  }
-}
+>>> print_requests.py //unity
+json.method = "PATCH";
+json.path = "/api/2.1/unity-catalog/schemas/main.myschema";
+json.body.comment = "COMMENT1";
 schemas schema1 id='main.myschema' name='myschema' catalog_name='main' comment='COMMENT1'
diff --git a/acceptance/bundle/resources/schemas/update/script b/acceptance/bundle/resources/schemas/update/script
index 09091d283e..733ed7eb9d 100644
--- a/acceptance/bundle/resources/schemas/update/script
+++ b/acceptance/bundle/resources/schemas/update/script
@@ -1,22 +1,21 @@
 echo "*" > .gitignore
 trace $CLI bundle deploy
 
-print_requests() {
-    jq 'select(.method != "GET" and (.path | contains("/unity")))' < out.requests.txt
-    rm out.requests.txt
-}
-
-trace print_requests
+trace print_requests.py //unity
 read_state.py schemas schema1 id name catalog_name comment
 
 title "Update comment and re-deploy"
 trace update_file.py databricks.yml COMMENT1 COMMENT2
 trace $CLI bundle deploy
-trace print_requests
+# Why the first time request match for direct & terraform, the requests from second deploy no longer match:
+# Terraform also sends "enable_predictive_optimization": "INHERIT" which is remote value that it stored in the state.
+trace print_requests.py //unity | gron.py | grep -v enable_predictive_optimization
 read_state.py schemas schema1 id name catalog_name comment
 
 title "Restore comment to original value and re-deploy"
 trace update_file.py databricks.yml COMMENT2 COMMENT1
 trace $CLI bundle deploy
-trace print_requests
+trace print_requests.py //unity | gron.py | grep -v enable_predictive_optimization
 read_state.py schemas schema1 id name catalog_name comment
+
+rm -f out.requests.txt
diff --git a/acceptance/bundle/user_agent/output.txt b/acceptance/bundle/user_agent/output.txt
index 3589f99e06..664f8ded44 100644
--- a/acceptance/bundle/user_agent/output.txt
+++ b/acceptance/bundle/user_agent/output.txt
@@ -64,6 +64,7 @@ OK  	destroy.terraform	/api/2.0/workspace-files/import-file/Workspace/Users/[USE
 OK  	destroy.terraform	/api/2.0/workspace/delete	engine/terraform
 MISS	destroy.terraform	/.well-known/databricks-config	'cli/[DEV_VERSION] databricks-sdk-go/[SDK_VERSION] go/[GO_VERSION] os/[OS]'
 MISS	destroy.terraform	/api/2.1/unity-catalog/schemas/mycatalog.myschema	'databricks-tf-provider/1.111.0 databricks-sdk-go/[SDK_VERSION] go/1.24.0 os/[OS] cli/[DEV_VERSION] terraform/1.5.5 sdk/sdkv2 resource/schema auth/pat'
+MISS	destroy.terraform	/api/2.1/unity-catalog/current-metastore-assignment	'databricks-tf-provider/1.111.0 databricks-sdk-go/[SDK_VERSION] go/1.24.0 os/[OS] cli/[DEV_VERSION] terraform/1.5.5 sdk/sdkv2 resource/schema auth/pat'
 MISS	destroy.terraform	/api/2.1/unity-catalog/schemas/mycatalog.myschema	'databricks-tf-provider/1.111.0 databricks-sdk-go/[SDK_VERSION] go/1.24.0 os/[OS] cli/[DEV_VERSION] terraform/1.5.5 sdk/sdkv2 resource/schema auth/pat'
 MISS	plan.direct	/api/2.0/preview/scim/v2/Me	'cli/[DEV_VERSION] databricks-sdk-go/[SDK_VERSION] go/[GO_VERSION] os/[OS] cmd/bundle_plan cmd-exec-id/[UUID] interactive/none auth/pat'
 MISS	plan.direct	/api/2.0/workspace/get-status	'cli/[DEV_VERSION] databricks-sdk-go/[SDK_VERSION] go/[GO_VERSION] os/[OS] cmd/bundle_plan cmd-exec-id/[UUID] interactive/none auth/pat'
diff --git a/acceptance/bundle/user_agent/simple/out.requests.deploy.terraform.json b/acceptance/bundle/user_agent/simple/out.requests.deploy.terraform.json
index 4b2699450a..e0981c7f29 100644
--- a/acceptance/bundle/user_agent/simple/out.requests.deploy.terraform.json
+++ b/acceptance/bundle/user_agent/simple/out.requests.deploy.terraform.json
@@ -239,15 +239,15 @@
             "attributes": {
               "catalog_name": "mycatalog",
               "comment": null,
-              "enable_predictive_optimization": null,
+              "enable_predictive_optimization": "INHERIT",
               "force_destroy": true,
               "id": "mycatalog.myschema",
-              "metastore_id": null,
+              "metastore_id": "[UUID]",
               "name": "myschema",
               "owner": "[USERNAME]",
               "properties": null,
               "provider_config": [],
-              "schema_id": "",
+              "schema_id": "[UUID]",
               "storage_root": null
             },
             "sensitive_attributes": [],
diff --git a/acceptance/bundle/user_agent/simple/out.requests.destroy.terraform.json b/acceptance/bundle/user_agent/simple/out.requests.destroy.terraform.json
index 152263ba11..82065d0ca4 100644
--- a/acceptance/bundle/user_agent/simple/out.requests.destroy.terraform.json
+++ b/acceptance/bundle/user_agent/simple/out.requests.destroy.terraform.json
@@ -145,6 +145,15 @@
     "force": "true"
   }
 }
+{
+  "headers": {
+    "User-Agent": [
+      "databricks-tf-provider/1.111.0 databricks-sdk-go/[SDK_VERSION] go/1.24.0 os/[OS] cli/[DEV_VERSION] terraform/1.5.5 sdk/sdkv2 resource/schema auth/pat"
+    ]
+  },
+  "method": "GET",
+  "path": "/api/2.1/unity-catalog/current-metastore-assignment"
+}
 {
   "headers": {
     "User-Agent": [
diff --git a/bundle/direct/apply.go b/bundle/direct/apply.go
index 04bb7054a2..b10df8ceb4 100644
--- a/bundle/direct/apply.go
+++ b/bundle/direct/apply.go
@@ -27,7 +27,7 @@ func (d *DeploymentUnit) Destroy(ctx context.Context, db *dstate.DeploymentState
 	return d.Delete(ctx, db, entry.ID)
 }
 
-func (d *DeploymentUnit) Deploy(ctx context.Context, db *dstate.DeploymentState, newState any, actionType deployplan.ActionType, changes deployplan.Changes) error {
+func (d *DeploymentUnit) Deploy(ctx context.Context, db *dstate.DeploymentState, newState any, actionType deployplan.ActionType, planEntry *deployplan.PlanEntry) error {
 	if actionType == deployplan.Create {
 		return d.Create(ctx, db, newState)
 	}
@@ -46,7 +46,7 @@ func (d *DeploymentUnit) Deploy(ctx context.Context, db *dstate.DeploymentState,
 	case deployplan.Recreate:
 		return d.Recreate(ctx, db, oldID, newState)
 	case deployplan.Update:
-		return d.Update(ctx, db, oldID, newState, changes)
+		return d.Update(ctx, db, oldID, newState, planEntry)
 	case deployplan.UpdateWithID:
 		return d.UpdateWithID(ctx, db, oldID, newState)
 	case deployplan.Resize:
@@ -103,12 +103,12 @@ func (d *DeploymentUnit) Recreate(ctx context.Context, db *dstate.DeploymentStat
 	return d.Create(ctx, db, newState)
 }
 
-func (d *DeploymentUnit) Update(ctx context.Context, db *dstate.DeploymentState, id string, newState any, changes deployplan.Changes) error {
+func (d *DeploymentUnit) Update(ctx context.Context, db *dstate.DeploymentState, id string, newState any, planEntry *deployplan.PlanEntry) error {
 	if !d.Adapter.HasDoUpdate() {
 		return fmt.Errorf("internal error: DoUpdate not implemented for resource %s", d.ResourceKey)
 	}
 
-	remoteState, err := d.Adapter.DoUpdate(ctx, id, newState, changes)
+	remoteState, err := d.Adapter.DoUpdate(ctx, id, newState, planEntry)
 	if err != nil {
 		return fmt.Errorf("updating id=%s: %w", id, err)
 	}
diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go
index 77d98a8a1f..ea3f615f7f 100644
--- a/bundle/direct/bundle_apply.go
+++ b/bundle/direct/bundle_apply.go
@@ -121,7 +121,7 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa
 				err = b.StateDB.SaveState(resourceKey, dbentry.ID, sv.Value, entry.DependsOn)
 			} else {
 				// TODO: redo calcDiff to downgrade planned action if possible (?)
-				err = d.Deploy(ctx, &b.StateDB, sv.Value, action, entry.Changes)
+				err = d.Deploy(ctx, &b.StateDB, sv.Value, action, entry)
 			}
 
 			if err != nil {
diff --git a/bundle/direct/dresources/adapter.go b/bundle/direct/dresources/adapter.go
index 2b54daeb08..bd4e7c750a 100644
--- a/bundle/direct/dresources/adapter.go
+++ b/bundle/direct/dresources/adapter.go
@@ -15,6 +15,7 @@ import (
 type (
 	Changes    = deployplan.Changes
 	ChangeDesc = deployplan.ChangeDesc
+	PlanEntry  = deployplan.PlanEntry
 )
 
 // IResource describes core methods for the resource implementation.
@@ -55,8 +56,8 @@ type IResource interface {
 
 	// [Optional] DoUpdate updates the resource. ID must not change as a result of this operation. Returns optionally remote state.
 	// If remote state is available as part of the operation, return it; otherwise return nil.
-	// Example: func (r *ResourceSchema) DoUpdate(ctx context.Context, id string, newState *catalog.CreateSchema, changes Changes) (*catalog.SchemaInfo, error)
-	DoUpdate(ctx context.Context, id string, newState any, changes Changes) (remoteState any, e error)
+	// Example: func (r *ResourceSchema) DoUpdate(ctx context.Context, id string, newState *catalog.CreateSchema, entry *PlanEntry) (*catalog.SchemaInfo, error)
+	DoUpdate(ctx context.Context, id string, newState any, entry *PlanEntry) (remoteState any, e error)
 
 	// [Optional] DoUpdateWithID performs an update that may result in resource having a new ID. Returns new id and optionally remote state.
 	DoUpdateWithID(ctx context.Context, id string, newState any) (newID string, remoteState any, e error)
@@ -435,14 +436,14 @@ func (a *Adapter) HasDoUpdate() bool {
 	return a.doUpdate != nil
 }
 
-// DoUpdate updates the resource with information about changes computed during plan.
+// DoUpdate updates the resource with the plan entry computed during plan.
 // Returns remote state if available, otherwise nil.
-func (a *Adapter) DoUpdate(ctx context.Context, id string, newState any, changes Changes) (any, error) {
+func (a *Adapter) DoUpdate(ctx context.Context, id string, newState any, entry *PlanEntry) (any, error) {
 	if a.doUpdate == nil {
 		return nil, errors.New("internal error: DoUpdate not found")
 	}
 
-	outs, err := a.doUpdate.Call(ctx, id, newState, changes)
+	outs, err := a.doUpdate.Call(ctx, id, newState, entry)
 	if err != nil {
 		return nil, err
 	}
diff --git a/bundle/direct/dresources/alert.go b/bundle/direct/dresources/alert.go
index 2300f8f56e..4fa4e410a1 100644
--- a/bundle/direct/dresources/alert.go
+++ b/bundle/direct/dresources/alert.go
@@ -49,7 +49,7 @@ func (r *ResourceAlert) DoCreate(ctx context.Context, config *sql.AlertV2) (stri
 }
 
 // DoUpdate updates the alert in place.
-func (r *ResourceAlert) DoUpdate(ctx context.Context, id string, config *sql.AlertV2, _ Changes) (*sql.AlertV2, error) {
+func (r *ResourceAlert) DoUpdate(ctx context.Context, id string, config *sql.AlertV2, _ *PlanEntry) (*sql.AlertV2, error) {
 	request := sql.UpdateAlertV2Request{
 		Id:         id,
 		Alert:      *config,
diff --git a/bundle/direct/dresources/all_test.go b/bundle/direct/dresources/all_test.go
index da59fedbe0..63caa5cfed 100644
--- a/bundle/direct/dresources/all_test.go
+++ b/bundle/direct/dresources/all_test.go
@@ -778,7 +778,7 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 	}
 
 	if adapter.HasDoUpdate() {
-		remoteStateFromUpdate, err := adapter.DoUpdate(ctx, createdID, newState, nil)
+		remoteStateFromUpdate, err := adapter.DoUpdate(ctx, createdID, newState, &deployplan.PlanEntry{})
 		require.NoError(t, err, "DoUpdate failed")
 		if remoteStateFromUpdate != nil {
 			remappedStateFromUpdate, err := adapter.RemapState(remoteStateFromUpdate)
diff --git a/bundle/direct/dresources/app.go b/bundle/direct/dresources/app.go
index e753079dac..1a5188c335 100644
--- a/bundle/direct/dresources/app.go
+++ b/bundle/direct/dresources/app.go
@@ -68,8 +68,8 @@ func (r *ResourceApp) DoCreate(ctx context.Context, config *apps.App) (string, *
 	return app.Name, nil, nil
 }
 
-func (r *ResourceApp) DoUpdate(ctx context.Context, id string, config *apps.App, changes Changes) (*apps.App, error) {
-	updateMask := strings.Join(collectUpdatePathsWithPrefix(changes, ""), ",")
+func (r *ResourceApp) DoUpdate(ctx context.Context, id string, config *apps.App, entry *PlanEntry) (*apps.App, error) {
+	updateMask := strings.Join(collectUpdatePathsWithPrefix(entry.Changes, ""), ",")
 
 	request := apps.AsyncUpdateAppRequest{
 		App:        config,
diff --git a/bundle/direct/dresources/catalog.go b/bundle/direct/dresources/catalog.go
index bb1fb82490..a9afa71cbf 100644
--- a/bundle/direct/dresources/catalog.go
+++ b/bundle/direct/dresources/catalog.go
@@ -48,7 +48,7 @@ func (r *ResourceCatalog) DoCreate(ctx context.Context, config *catalog.CreateCa
 }
 
 // DoUpdate updates the catalog in place and returns remote state.
-func (r *ResourceCatalog) DoUpdate(ctx context.Context, id string, config *catalog.CreateCatalog, _ Changes) (*catalog.CatalogInfo, error) {
+func (r *ResourceCatalog) DoUpdate(ctx context.Context, id string, config *catalog.CreateCatalog, _ *PlanEntry) (*catalog.CatalogInfo, error) {
 	updateRequest := catalog.UpdateCatalog{
 		Comment:                      config.Comment,
 		EnablePredictiveOptimization: "", // Not supported by DABs
diff --git a/bundle/direct/dresources/cluster.go b/bundle/direct/dresources/cluster.go
index a8f78d12f9..46148a2655 100644
--- a/bundle/direct/dresources/cluster.go
+++ b/bundle/direct/dresources/cluster.go
@@ -86,7 +86,7 @@ func (r *ResourceCluster) DoCreate(ctx context.Context, config *compute.ClusterS
 	return wait.ClusterId, nil, nil
 }
 
-func (r *ResourceCluster) DoUpdate(ctx context.Context, id string, config *compute.ClusterSpec, _ Changes) (*compute.ClusterDetails, error) {
+func (r *ResourceCluster) DoUpdate(ctx context.Context, id string, config *compute.ClusterSpec, _ *PlanEntry) (*compute.ClusterDetails, error) {
 	// Same retry as in TF provider logic
 	// https://github.com/databricks/terraform-provider-databricks/blob/3eecd0f90cf99d7777e79a3d03c41f9b2aafb004/clusters/resource_cluster.go#L624
 	timeout := 15 * time.Minute
diff --git a/bundle/direct/dresources/dashboard.go b/bundle/direct/dresources/dashboard.go
index 02f573bb81..f33d7946e8 100644
--- a/bundle/direct/dresources/dashboard.go
+++ b/bundle/direct/dresources/dashboard.go
@@ -321,7 +321,7 @@ func (r *ResourceDashboard) DoCreate(ctx context.Context, config *DashboardState
 	return createResp.DashboardId, responseToState(createResp, publishResp, dashboard.SerializedDashboard, config.Published), nil
 }
 
-func (r *ResourceDashboard) DoUpdate(ctx context.Context, id string, config *DashboardState, _ Changes) (*DashboardState, error) {
+func (r *ResourceDashboard) DoUpdate(ctx context.Context, id string, config *DashboardState, _ *PlanEntry) (*DashboardState, error) {
 	dashboard, err := prepareDashboardRequest(config)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/database_catalog.go b/bundle/direct/dresources/database_catalog.go
index fdf09fbac9..ea97574d4b 100644
--- a/bundle/direct/dresources/database_catalog.go
+++ b/bundle/direct/dresources/database_catalog.go
@@ -34,7 +34,7 @@ func (r *ResourceDatabaseCatalog) DoCreate(ctx context.Context, config *database
 	return result.Name, nil, nil
 }
 
-func (r *ResourceDatabaseCatalog) DoUpdate(ctx context.Context, id string, config *database.DatabaseCatalog, _ Changes) (*database.DatabaseCatalog, error) {
+func (r *ResourceDatabaseCatalog) DoUpdate(ctx context.Context, id string, config *database.DatabaseCatalog, _ *PlanEntry) (*database.DatabaseCatalog, error) {
 	request := database.UpdateDatabaseCatalogRequest{
 		DatabaseCatalog: *config,
 		Name:            id,
diff --git a/bundle/direct/dresources/database_instance.go b/bundle/direct/dresources/database_instance.go
index 39f74caa04..e7a5e8d824 100644
--- a/bundle/direct/dresources/database_instance.go
+++ b/bundle/direct/dresources/database_instance.go
@@ -35,7 +35,7 @@ func (d *ResourceDatabaseInstance) DoCreate(ctx context.Context, config *databas
 	return waiter.Response.Name, nil, nil
 }
 
-func (d *ResourceDatabaseInstance) DoUpdate(ctx context.Context, id string, config *database.DatabaseInstance, _ Changes) (*database.DatabaseInstance, error) {
+func (d *ResourceDatabaseInstance) DoUpdate(ctx context.Context, id string, config *database.DatabaseInstance, _ *PlanEntry) (*database.DatabaseInstance, error) {
 	request := database.UpdateDatabaseInstanceRequest{
 		DatabaseInstance: *config,
 		Name:             config.Name,
diff --git a/bundle/direct/dresources/experiment.go b/bundle/direct/dresources/experiment.go
index b2fc45af64..de62720fe8 100644
--- a/bundle/direct/dresources/experiment.go
+++ b/bundle/direct/dresources/experiment.go
@@ -58,7 +58,7 @@ func (r *ResourceExperiment) DoCreate(ctx context.Context, config *ml.CreateExpe
 	return result.ExperimentId, nil, nil
 }
 
-func (r *ResourceExperiment) DoUpdate(ctx context.Context, id string, config *ml.CreateExperiment, _ Changes) (*ml.Experiment, error) {
+func (r *ResourceExperiment) DoUpdate(ctx context.Context, id string, config *ml.CreateExperiment, _ *PlanEntry) (*ml.Experiment, error) {
 	updateReq := ml.UpdateExperiment{
 		ExperimentId:    id,
 		NewName:         config.Name,
diff --git a/bundle/direct/dresources/external_location.go b/bundle/direct/dresources/external_location.go
index ca6821611c..f9416567cb 100644
--- a/bundle/direct/dresources/external_location.go
+++ b/bundle/direct/dresources/external_location.go
@@ -52,7 +52,7 @@ func (r *ResourceExternalLocation) DoCreate(ctx context.Context, config *catalog
 }
 
 // DoUpdate updates the external location in place and returns remote state.
-func (r *ResourceExternalLocation) DoUpdate(ctx context.Context, id string, config *catalog.CreateExternalLocation, _ Changes) (*catalog.ExternalLocationInfo, error) {
+func (r *ResourceExternalLocation) DoUpdate(ctx context.Context, id string, config *catalog.CreateExternalLocation, _ *PlanEntry) (*catalog.ExternalLocationInfo, error) {
 	updateRequest := catalog.UpdateExternalLocation{
 		Comment:        config.Comment,
 		CredentialName: config.CredentialName,
diff --git a/bundle/direct/dresources/grants.go b/bundle/direct/dresources/grants.go
index 3172ce55e3..8bb1906122 100644
--- a/bundle/direct/dresources/grants.go
+++ b/bundle/direct/dresources/grants.go
@@ -49,7 +49,7 @@ func PrepareGrantsInputConfig(inputConfig any, node string) (*structvar.StructVa
 
 	// Backend sorts privileges, so we sort here as well.
 	for i := range *grantsPtr {
-		sortPriviliges((*grantsPtr)[i].Privileges)
+		slices.Sort((*grantsPtr)[i].Privileges)
 	}
 
 	return &structvar.StructVar{
@@ -107,16 +107,25 @@ func (r *ResourceGrants) DoRead(ctx context.Context, id string) (*GrantsState, e
 }
 
 func (r *ResourceGrants) DoCreate(ctx context.Context, state *GrantsState) (string, *GrantsState, error) {
-	err := r.applyGrants(ctx, state)
+	_, err := r.DoUpdate(ctx, "", state, nil)
 	if err != nil {
 		return "", nil, err
 	}
 
-	return makeGrantsID(state.SecurableType, state.FullName), nil, nil
+	return state.SecurableType + "/" + state.FullName, nil, nil
 }
 
-func (r *ResourceGrants) DoUpdate(ctx context.Context, _ string, state *GrantsState, _ Changes) (*GrantsState, error) {
-	return nil, r.applyGrants(ctx, state)
+func (r *ResourceGrants) DoUpdate(ctx context.Context, _ string, state *GrantsState, entry *PlanEntry) (*GrantsState, error) {
+	if state.FullName == "" {
+		return nil, errors.New("internal error: grants full_name must be resolved before deployment")
+	}
+	removedPrincipals := removedGrantPrincipals(state.EmbeddedSlice, entry)
+	_, err := r.client.Grants.Update(ctx, catalog.UpdatePermissions{
+		SecurableType: state.SecurableType,
+		FullName:      state.FullName,
+		Changes:       buildGrantChanges(state.EmbeddedSlice, removedPrincipals),
+	})
+	return nil, err
 }
 
 func (r *ResourceGrants) DoDelete(ctx context.Context, id string) error {
@@ -125,13 +134,9 @@ func (r *ResourceGrants) DoDelete(ctx context.Context, id string) error {
 	return nil
 }
 
-func (r *ResourceGrants) applyGrants(ctx context.Context, state *GrantsState) error {
-	if state.FullName == "" {
-		return errors.New("internal error: grants full_name must be resolved before deployment")
-	}
-
-	var changes []catalog.PermissionsChange
-	for _, ga := range state.EmbeddedSlice {
+func buildGrantChanges(desiredAssignments []catalog.PrivilegeAssignment, removedPrincipals []string) []catalog.PermissionsChange {
+	changes := make([]catalog.PermissionsChange, 0, len(desiredAssignments)+len(removedPrincipals))
+	for _, ga := range desiredAssignments {
 		change := catalog.PermissionsChange{
 			Principal:       ga.Principal,
 			Add:             ga.Privileges,
@@ -145,13 +150,42 @@ func (r *ResourceGrants) applyGrants(ctx context.Context, state *GrantsState) er
 		}
 		changes = append(changes, change)
 	}
+	for _, principal := range removedPrincipals {
+		changes = append(changes, catalog.PermissionsChange{
+			Principal:       principal,
+			Add:             nil,
+			Remove:          []catalog.Privilege{catalog.PrivilegeAllPrivileges},
+			ForceSendFields: nil,
+		})
+	}
+	return changes
+}
 
-	_, err := r.client.Grants.Update(ctx, catalog.UpdatePermissions{
-		SecurableType: state.SecurableType,
-		FullName:      state.FullName,
-		Changes:       changes,
-	})
-	return err
+// removedGrantPrincipals returns principals present in the remote state but absent from the desired assignments.
+func removedGrantPrincipals(desiredAssignments []catalog.PrivilegeAssignment, entry *PlanEntry) []string {
+	if entry == nil {
+		return nil
+	}
+	remote, ok := entry.RemoteState.(*GrantsState)
+	if !ok || remote == nil {
+		return nil
+	}
+
+	desired := make(map[string]struct{}, len(desiredAssignments))
+	for _, a := range desiredAssignments {
+		if a.Principal != "" {
+			desired[a.Principal] = struct{}{}
+		}
+	}
+
+	var result []string
+	for _, a := range remote.EmbeddedSlice {
+		if _, ok := desired[a.Principal]; !ok {
+			result = append(result, a.Principal)
+		}
+	}
+	slices.Sort(result)
+	return result
 }
 
 func (r *ResourceGrants) listGrants(ctx context.Context, securableType, fullName string) ([]catalog.PrivilegeAssignment, error) {
@@ -173,11 +207,9 @@ func (r *ResourceGrants) listGrants(ctx context.Context, securableType, fullName
 			if assignment.Principal == "" {
 				continue
 			}
-			privs := make([]catalog.Privilege, len(assignment.Privileges))
-			copy(privs, assignment.Privileges)
 			assignments = append(assignments, catalog.PrivilegeAssignment{
 				Principal:       assignment.Principal,
-				Privileges:      privs,
+				Privileges:      assignment.Privileges,
 				ForceSendFields: nil,
 			})
 		}
@@ -189,10 +221,6 @@ func (r *ResourceGrants) listGrants(ctx context.Context, securableType, fullName
 	return assignments, nil
 }
 
-func sortPriviliges(privileges []catalog.Privilege) {
-	slices.Sort(privileges)
-}
-
 func extractGrantResourceType(node string) (string, error) {
 	rest, ok := strings.CutPrefix(node, "resources.")
 	if !ok {
@@ -215,7 +243,3 @@ func parseGrantsID(id string) (string, string, error) {
 	}
 	return parts[0], parts[1], nil
 }
-
-func makeGrantsID(securableType, fullName string) string {
-	return securableType + "/" + fullName
-}
diff --git a/bundle/direct/dresources/grants_test.go b/bundle/direct/dresources/grants_test.go
new file mode 100644
index 0000000000..1724eed40d
--- /dev/null
+++ b/bundle/direct/dresources/grants_test.go
@@ -0,0 +1,76 @@
+package dresources
+
+import (
+	"testing"
+
+	"github.com/databricks/databricks-sdk-go/service/catalog"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBuildGrantChanges(t *testing.T) {
+	tests := []struct {
+		name     string
+		desired  []catalog.PrivilegeAssignment
+		removed  []string
+		expected []catalog.PermissionsChange
+	}{
+		{
+			name: "removes all other privileges for desired principal",
+			desired: []catalog.PrivilegeAssignment{
+				{
+					Principal: "alice",
+					Privileges: []catalog.Privilege{
+						catalog.PrivilegeApplyTag,
+						catalog.PrivilegeCreateTable,
+					},
+				},
+			},
+			expected: []catalog.PermissionsChange{
+				{
+					Principal: "alice",
+					Add: []catalog.Privilege{
+						catalog.PrivilegeApplyTag,
+						catalog.PrivilegeCreateTable,
+					},
+					Remove: []catalog.Privilege{
+						catalog.PrivilegeAllPrivileges,
+					},
+				},
+			},
+		},
+		{
+			name: "skips ALL_PRIVILEGES removal when granting ALL_PRIVILEGES",
+			desired: []catalog.PrivilegeAssignment{
+				{
+					Principal: "alice",
+					Privileges: []catalog.Privilege{
+						catalog.PrivilegeAllPrivileges,
+					},
+				},
+			},
+			removed: []string{
+				"bob",
+			},
+			expected: []catalog.PermissionsChange{
+				{
+					Principal: "alice",
+					Add: []catalog.Privilege{
+						catalog.PrivilegeAllPrivileges,
+					},
+				},
+				{
+					Principal: "bob",
+					Remove: []catalog.Privilege{
+						catalog.PrivilegeAllPrivileges,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.expected, buildGrantChanges(tt.desired, tt.removed))
+		})
+	}
+}
diff --git a/bundle/direct/dresources/job.go b/bundle/direct/dresources/job.go
index ba3e932b34..8ea33e1f7f 100644
--- a/bundle/direct/dresources/job.go
+++ b/bundle/direct/dresources/job.go
@@ -112,7 +112,7 @@ func (r *ResourceJob) DoCreate(ctx context.Context, config *jobs.JobSettings) (s
 	return strconv.FormatInt(response.JobId, 10), nil, nil
 }
 
-func (r *ResourceJob) DoUpdate(ctx context.Context, id string, config *jobs.JobSettings, _ Changes) (*JobRemote, error) {
+func (r *ResourceJob) DoUpdate(ctx context.Context, id string, config *jobs.JobSettings, _ *PlanEntry) (*JobRemote, error) {
 	request, err := makeResetJob(*config, id)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/model.go b/bundle/direct/dresources/model.go
index cb9a3c9b74..9373de99ba 100644
--- a/bundle/direct/dresources/model.go
+++ b/bundle/direct/dresources/model.go
@@ -67,7 +67,7 @@ func (r *ResourceMlflowModel) DoCreate(ctx context.Context, config *ml.CreateMod
 	return response.RegisteredModel.Name, modelDatabricks, nil
 }
 
-func (r *ResourceMlflowModel) DoUpdate(ctx context.Context, id string, config *ml.CreateModelRequest, _ Changes) (*ml.ModelDatabricks, error) {
+func (r *ResourceMlflowModel) DoUpdate(ctx context.Context, id string, config *ml.CreateModelRequest, _ *PlanEntry) (*ml.ModelDatabricks, error) {
 	updateRequest := ml.UpdateModelRequest{
 		Name:            id,
 		Description:     config.Description,
diff --git a/bundle/direct/dresources/model_serving_endpoint.go b/bundle/direct/dresources/model_serving_endpoint.go
index 8a8aa126b2..3822897ff3 100644
--- a/bundle/direct/dresources/model_serving_endpoint.go
+++ b/bundle/direct/dresources/model_serving_endpoint.go
@@ -285,34 +285,34 @@ func (r *ResourceModelServingEndpoint) updateTags(ctx context.Context, id string
 	return nil
 }
 
-func (r *ResourceModelServingEndpoint) DoUpdate(ctx context.Context, id string, config *serving.CreateServingEndpoint, changes Changes) (*RefreshOutput, error) {
+func (r *ResourceModelServingEndpoint) DoUpdate(ctx context.Context, id string, config *serving.CreateServingEndpoint, entry *PlanEntry) (*RefreshOutput, error) {
 	var err error
 
 	// Terraform makes these API calls sequentially. We do the same here.
 	// It's an unknown as of 1st Dec 2025 if these APIs are safe to make in parallel. (we did not check)
 	// https://github.com/databricks/terraform-provider-databricks/blob/c61a32300445f84efb2bb6827dee35e6e523f4ff/serving/resource_model_serving.go#L373
-	if changes.HasChange(pathTags) {
+	if entry.Changes.HasChange(pathTags) {
 		err = r.updateTags(ctx, id, config.Tags)
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	if changes.HasChange(pathAiGateway) {
+	if entry.Changes.HasChange(pathAiGateway) {
 		err = r.updateAiGateway(ctx, id, config.AiGateway)
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	if changes.HasChange(pathConfig) {
+	if entry.Changes.HasChange(pathConfig) {
 		err = r.updateConfig(ctx, id, config.Config)
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	if changes.HasChange(pathEmailNotifications) {
+	if entry.Changes.HasChange(pathEmailNotifications) {
 		err = r.updateNotifications(ctx, id, config.EmailNotifications)
 		if err != nil {
 			return nil, err
diff --git a/bundle/direct/dresources/permissions.go b/bundle/direct/dresources/permissions.go
index 7a64f53d4b..ba8e3ccfb2 100644
--- a/bundle/direct/dresources/permissions.go
+++ b/bundle/direct/dresources/permissions.go
@@ -223,7 +223,7 @@ func (r *ResourcePermissions) DoCreate(ctx context.Context, newState *Permission
 }
 
 // DoUpdate calls https://docs.databricks.com/api/workspace/jobs/setjobpermissions.
-func (r *ResourcePermissions) DoUpdate(ctx context.Context, _ string, newState *PermissionsState, _ Changes) (*PermissionsState, error) {
+func (r *ResourcePermissions) DoUpdate(ctx context.Context, _ string, newState *PermissionsState, _ *PlanEntry) (*PermissionsState, error) {
 	extractedType, extractedID, err := parsePermissionsID(newState.ObjectID)
 	if err != nil {
 		return nil, err
diff --git a/bundle/direct/dresources/pipeline.go b/bundle/direct/dresources/pipeline.go
index 30d2ab5313..9a59ab4e40 100644
--- a/bundle/direct/dresources/pipeline.go
+++ b/bundle/direct/dresources/pipeline.go
@@ -130,7 +130,7 @@ func (r *ResourcePipeline) DoCreate(ctx context.Context, config *pipelines.Creat
 	return response.PipelineId, nil, nil
 }
 
-func (r *ResourcePipeline) DoUpdate(ctx context.Context, id string, config *pipelines.CreatePipeline, _ Changes) (*PipelineRemote, error) {
+func (r *ResourcePipeline) DoUpdate(ctx context.Context, id string, config *pipelines.CreatePipeline, _ *PlanEntry) (*PipelineRemote, error) {
 	request := pipelines.EditPipeline{
 		AllowDuplicateNames:  config.AllowDuplicateNames,
 		BudgetPolicyId:       config.BudgetPolicyId,
diff --git a/bundle/direct/dresources/postgres_branch.go b/bundle/direct/dresources/postgres_branch.go
index 2e79eb5dda..cb042699de 100644
--- a/bundle/direct/dresources/postgres_branch.go
+++ b/bundle/direct/dresources/postgres_branch.go
@@ -86,12 +86,12 @@ func (r *ResourcePostgresBranch) DoCreate(ctx context.Context, config *PostgresB
 	return result.Name, result, nil
 }
 
-func (r *ResourcePostgresBranch) DoUpdate(ctx context.Context, id string, config *PostgresBranchState, changes Changes) (*postgres.Branch, error) {
+func (r *ResourcePostgresBranch) DoUpdate(ctx context.Context, id string, config *PostgresBranchState, entry *PlanEntry) (*postgres.Branch, error) {
 	// Build update mask from fields that have action="update" in the changes map.
 	// This excludes immutable fields and fields that haven't changed.
 	// Prefix with "spec." because the API expects paths relative to the Branch object,
 	// not relative to our flattened state type.
-	fieldPaths := collectUpdatePathsWithPrefix(changes, "spec.")
+	fieldPaths := collectUpdatePathsWithPrefix(entry.Changes, "spec.")
 
 	waiter, err := r.client.Postgres.UpdateBranch(ctx, postgres.UpdateBranchRequest{
 		Branch: postgres.Branch{
diff --git a/bundle/direct/dresources/postgres_endpoint.go b/bundle/direct/dresources/postgres_endpoint.go
index 3d3f763ee1..aa2c06c82a 100644
--- a/bundle/direct/dresources/postgres_endpoint.go
+++ b/bundle/direct/dresources/postgres_endpoint.go
@@ -131,12 +131,12 @@ func (r *ResourcePostgresEndpoint) DoCreate(ctx context.Context, config *Postgre
 	return result.Name, result, nil
 }
 
-func (r *ResourcePostgresEndpoint) DoUpdate(ctx context.Context, id string, config *PostgresEndpointState, changes Changes) (*postgres.Endpoint, error) {
+func (r *ResourcePostgresEndpoint) DoUpdate(ctx context.Context, id string, config *PostgresEndpointState, entry *PlanEntry) (*postgres.Endpoint, error) {
 	// Build update mask from fields that have action="update" in the changes map.
 	// This excludes immutable fields and fields that haven't changed.
 	// Prefix with "spec." because the API expects paths relative to the Endpoint object,
 	// not relative to our flattened state type.
-	fieldPaths := collectUpdatePathsWithPrefix(changes, "spec.")
+	fieldPaths := collectUpdatePathsWithPrefix(entry.Changes, "spec.")
 
 	waiter, err := r.client.Postgres.UpdateEndpoint(ctx, postgres.UpdateEndpointRequest{
 		Endpoint: postgres.Endpoint{
diff --git a/bundle/direct/dresources/postgres_project.go b/bundle/direct/dresources/postgres_project.go
index 85c20b9cbe..222d201d8f 100644
--- a/bundle/direct/dresources/postgres_project.go
+++ b/bundle/direct/dresources/postgres_project.go
@@ -83,12 +83,12 @@ func (r *ResourcePostgresProject) DoCreate(ctx context.Context, config *Postgres
 	return result.Name, result, nil
 }
 
-func (r *ResourcePostgresProject) DoUpdate(ctx context.Context, id string, config *PostgresProjectState, changes Changes) (*postgres.Project, error) {
+func (r *ResourcePostgresProject) DoUpdate(ctx context.Context, id string, config *PostgresProjectState, entry *PlanEntry) (*postgres.Project, error) {
 	// Build update mask from fields that have action="update" in the changes map.
 	// This excludes immutable fields and fields that haven't changed.
 	// Prefix with "spec." because the API expects paths relative to the Project object,
 	// not relative to our flattened state type.
-	fieldPaths := collectUpdatePathsWithPrefix(changes, "spec.")
+	fieldPaths := collectUpdatePathsWithPrefix(entry.Changes, "spec.")
 
 	waiter, err := r.client.Postgres.UpdateProject(ctx, postgres.UpdateProjectRequest{
 		Project: postgres.Project{
diff --git a/bundle/direct/dresources/quality_monitor.go b/bundle/direct/dresources/quality_monitor.go
index dcc48fb95d..3866779618 100644
--- a/bundle/direct/dresources/quality_monitor.go
+++ b/bundle/direct/dresources/quality_monitor.go
@@ -83,7 +83,7 @@ func (r *ResourceQualityMonitor) DoCreate(ctx context.Context, config *QualityMo
 	return response.TableName, response, nil
 }
 
-func (r *ResourceQualityMonitor) DoUpdate(ctx context.Context, id string, config *QualityMonitorState, _ Changes) (*catalog.MonitorInfo, error) {
+func (r *ResourceQualityMonitor) DoUpdate(ctx context.Context, id string, config *QualityMonitorState, _ *PlanEntry) (*catalog.MonitorInfo, error) {
 	updateRequest := catalog.UpdateMonitor{
 		TableName:                id,
 		BaselineTableName:        config.BaselineTableName,
diff --git a/bundle/direct/dresources/registered_model.go b/bundle/direct/dresources/registered_model.go
index 666716b48b..e191f52b8b 100644
--- a/bundle/direct/dresources/registered_model.go
+++ b/bundle/direct/dresources/registered_model.go
@@ -64,7 +64,7 @@ func (r *ResourceRegisteredModel) DoCreate(ctx context.Context, config *catalog.
 	return response.FullName, response, nil
 }
 
-func (r *ResourceRegisteredModel) DoUpdate(ctx context.Context, id string, config *catalog.CreateRegisteredModelRequest, _ Changes) (*catalog.RegisteredModelInfo, error) {
+func (r *ResourceRegisteredModel) DoUpdate(ctx context.Context, id string, config *catalog.CreateRegisteredModelRequest, _ *PlanEntry) (*catalog.RegisteredModelInfo, error) {
 	updateRequest := catalog.UpdateRegisteredModelRequest{
 		FullName:        id,
 		Comment:         config.Comment,
diff --git a/bundle/direct/dresources/schema.go b/bundle/direct/dresources/schema.go
index e0e05ac0d9..783b87caf9 100644
--- a/bundle/direct/dresources/schema.go
+++ b/bundle/direct/dresources/schema.go
@@ -46,7 +46,7 @@ func (r *ResourceSchema) DoCreate(ctx context.Context, config *catalog.CreateSch
 }
 
 // DoUpdate updates the schema in place and returns remote state.
-func (r *ResourceSchema) DoUpdate(ctx context.Context, id string, config *catalog.CreateSchema, _ Changes) (*catalog.SchemaInfo, error) {
+func (r *ResourceSchema) DoUpdate(ctx context.Context, id string, config *catalog.CreateSchema, _ *PlanEntry) (*catalog.SchemaInfo, error) {
 	updateRequest := catalog.UpdateSchema{
 		Comment:                      config.Comment,
 		EnablePredictiveOptimization: "", // Not supported by DABs
diff --git a/bundle/direct/dresources/schema_test.go b/bundle/direct/dresources/schema_test.go
index f22e3d87ce..d013610e05 100644
--- a/bundle/direct/dresources/schema_test.go
+++ b/bundle/direct/dresources/schema_test.go
@@ -37,7 +37,7 @@ func TestResourceSchema_DoUpdate_WithUnsupportedForceSendFields(t *testing.T) {
 		"Owner",                        // Unsupported - should be filtered out
 	}
 
-	_, err = adapter.DoUpdate(ctx, id, config, nil)
+	_, err = adapter.DoUpdate(ctx, id, config, &PlanEntry{})
 	require.NoError(t, err)
 
 	result, err := adapter.DoRead(ctx, id)
@@ -45,6 +45,7 @@ func TestResourceSchema_DoUpdate_WithUnsupportedForceSendFields(t *testing.T) {
 
 	result.CreatedAt = 0
 	result.UpdatedAt = 0
+	result.SchemaId = ""
 
 	resultJSON, err := json.Marshal(result)
 	require.NoError(t, err)
@@ -54,12 +55,20 @@ func TestResourceSchema_DoUpdate_WithUnsupportedForceSendFields(t *testing.T) {
 		"created_at": 0,
 		"created_by": "tester@databricks.com",
 		"comment": "updated comment",
+		"effective_predictive_optimization_flag": {
+			"inherited_from_name": "deco-uc-prod-isolated-aws-us-east-1",
+			"inherited_from_type": "METASTORE",
+			"value": "ENABLE"
+		},
+		"enable_predictive_optimization": "INHERIT",
 		"properties": {"key": "updated_value"},
 		"full_name": "main.test_schema",
+		"metastore_id": "120efa64-9b68-46ba-be38-f319458430d2",
 		"name": "test_schema",
 		"owner": "tester@databricks.com",
 		"updated_at": 0,
-		"updated_by": "tester@databricks.com"
+		"updated_by": "tester@databricks.com",
+		"schema_id": ""
 	}`
 	assert.JSONEq(t, expected, string(resultJSON))
 }
diff --git a/bundle/direct/dresources/secret_scope_acls.go b/bundle/direct/dresources/secret_scope_acls.go
index 034f6c3952..dc03544e9c 100644
--- a/bundle/direct/dresources/secret_scope_acls.go
+++ b/bundle/direct/dresources/secret_scope_acls.go
@@ -109,7 +109,7 @@ func (r *ResourceSecretScopeAcls) DoUpdateWithID(ctx context.Context, id string,
 	return state.ScopeName, nil, nil
 }
 
-func (r *ResourceSecretScopeAcls) DoUpdate(ctx context.Context, id string, state *SecretScopeAclsState, changes Changes) (*SecretScopeAclsState, error) {
+func (r *ResourceSecretScopeAcls) DoUpdate(ctx context.Context, id string, state *SecretScopeAclsState, _ *PlanEntry) (*SecretScopeAclsState, error) {
 	_, _, err := r.DoUpdateWithID(ctx, id, state)
 	return nil, err
 }
diff --git a/bundle/direct/dresources/sql_warehouse.go b/bundle/direct/dresources/sql_warehouse.go
index 5d9d7793b7..2e8caa4a4a 100644
--- a/bundle/direct/dresources/sql_warehouse.go
+++ b/bundle/direct/dresources/sql_warehouse.go
@@ -58,7 +58,7 @@ func (r *ResourceSqlWarehouse) DoCreate(ctx context.Context, config *sql.CreateW
 }
 
 // DoUpdate updates the warehouse in place.
-func (r *ResourceSqlWarehouse) DoUpdate(ctx context.Context, id string, config *sql.CreateWarehouseRequest, _ Changes) (*sql.GetWarehouseResponse, error) {
+func (r *ResourceSqlWarehouse) DoUpdate(ctx context.Context, id string, config *sql.CreateWarehouseRequest, _ *PlanEntry) (*sql.GetWarehouseResponse, error) {
 	request := sql.EditWarehouseRequest{
 		AutoStopMins:            config.AutoStopMins,
 		Channel:                 config.Channel,
diff --git a/bundle/direct/dresources/synced_database_table.go b/bundle/direct/dresources/synced_database_table.go
index d0cb4987b9..94182b729f 100644
--- a/bundle/direct/dresources/synced_database_table.go
+++ b/bundle/direct/dresources/synced_database_table.go
@@ -34,7 +34,7 @@ func (r *ResourceSyncedDatabaseTable) DoCreate(ctx context.Context, config *data
 	return result.Name, nil, nil
 }
 
-func (r *ResourceSyncedDatabaseTable) DoUpdate(ctx context.Context, id string, config *database.SyncedDatabaseTable, _ Changes) (*database.SyncedDatabaseTable, error) {
+func (r *ResourceSyncedDatabaseTable) DoUpdate(ctx context.Context, id string, config *database.SyncedDatabaseTable, _ *PlanEntry) (*database.SyncedDatabaseTable, error) {
 	request := database.UpdateSyncedDatabaseTableRequest{
 		SyncedTable: *config,
 		Name:        id,
diff --git a/bundle/direct/dresources/volume.go b/bundle/direct/dresources/volume.go
index 6d33a49e27..62a08987ee 100644
--- a/bundle/direct/dresources/volume.go
+++ b/bundle/direct/dresources/volume.go
@@ -48,7 +48,7 @@ func (r *ResourceVolume) DoCreate(ctx context.Context, config *catalog.CreateVol
 	return response.FullName, response, nil
 }
 
-func (r *ResourceVolume) DoUpdate(ctx context.Context, id string, config *catalog.CreateVolumeRequestContent, _ Changes) (*catalog.VolumeInfo, error) {
+func (r *ResourceVolume) DoUpdate(ctx context.Context, id string, config *catalog.CreateVolumeRequestContent, _ *PlanEntry) (*catalog.VolumeInfo, error) {
 	updateRequest := catalog.UpdateVolumeRequestContent{
 		Comment: config.Comment,
 		Name:    id,
diff --git a/libs/testserver/grants.go b/libs/testserver/grants.go
index 2070a1b0e1..4710a49d29 100644
--- a/libs/testserver/grants.go
+++ b/libs/testserver/grants.go
@@ -107,6 +107,15 @@ func (s *FakeWorkspace) GrantsUpdate(req Request, securableType, fullName string
 
 	s.Grants[key] = assignments
 
+	if securableType == "schema" {
+		schema, ok := s.Schemas[fullName]
+		if ok {
+			schema.UpdatedAt = nowMilli()
+			schema.UpdatedBy = s.CurrentUser().UserName
+			s.Schemas[fullName] = schema
+		}
+	}
+
 	return Response{Body: catalog.UpdatePermissionsResponse{PrivilegeAssignments: assignments}}
 }
 
diff --git a/libs/testserver/schemas.go b/libs/testserver/schemas.go
index 1d10c4b19f..1d4dc79e7a 100644
--- a/libs/testserver/schemas.go
+++ b/libs/testserver/schemas.go
@@ -9,6 +9,8 @@ import (
 	"github.com/databricks/databricks-sdk-go/service/catalog"
 )
 
+const testMetastoreName = "deco-uc-prod-isolated-aws-us-east-1"
+
 func (s *FakeWorkspace) SchemasCreate(req Request) Response {
 	defer s.LockUnlock()()
 
@@ -28,7 +30,15 @@ func (s *FakeWorkspace) SchemasCreate(req Request) Response {
 	schema.UpdatedAt = schema.CreatedAt
 	schema.CreatedBy = s.CurrentUser().UserName
 	schema.UpdatedBy = s.CurrentUser().UserName
+	schema.EffectivePredictiveOptimizationFlag = &catalog.EffectivePredictiveOptimizationFlag{
+		InheritedFromName: testMetastoreName,
+		InheritedFromType: catalog.EffectivePredictiveOptimizationFlagInheritedFromType("METASTORE"),
+		Value:             catalog.EnablePredictiveOptimizationEnable,
+	}
+	schema.EnablePredictiveOptimization = catalog.EnablePredictiveOptimizationInherit
+	schema.MetastoreId = TestMetastore.MetastoreId
 	schema.Owner = s.CurrentUser().UserName
+	schema.SchemaId = nextUUID()
 	s.Schemas[schema.FullName] = schema
 
 	return Response{