feat(sdk): expose table overrides for remote tables

Expose table-level maintenance override read, update, and reset APIs for remote
Rust and Python tables. Local tables continue to return `NotSupported`.

Updates recursively merge the caller-provided JSON object into the current
override document so unrelated and unknown service fields are preserved. Reset
uses the explicit reset endpoint.

The integrated change keeps current main's remote database identity, namespace,
branch, SQL, WAL, and threaded mock-server behavior.

Validated with focused Rust and Python override tests and the complete applicable
CI run for 43bccbadc5.

python/python/lancedb/_lancedb.pyi

@@ -593,6 +593,9 @@ class Table:
     async def enable_rte(self, spec: RTEWriteSpec) -> None: ...
     async def disable_rte(self) -> None: ...
     async def get_rte_spec(self) -> Optional[RTEWriteSpec]: ...
+    async def get_table_overrides(self) -> str: ...
+    async def update_table_overrides(self, overrides_json: str) -> str: ...
+    async def reset_table_overrides(self) -> str: ...
     # WAL-PK-FUSION: delete both.
     def hybrid_pk_fusion_learned(self) -> bool: ...
     def note_hybrid_pk_fusion(self) -> None: ...

python/python/lancedb/remote/table.py

@@ -1153,6 +1153,18 @@ class RemoteTable(Table):
         """Read the installed RTEWriteSpec, or ``None``."""
         return LOOP.run(self._table.get_rte_spec())
 
+    def get_table_overrides(self) -> dict[str, Any]:
+        """Read Enterprise table overrides for this table."""
+        return LOOP.run(self._table.get_table_overrides())
+
+    def update_table_overrides(self, **overrides: Any) -> dict[str, Any]:
+        """Merge Enterprise table overrides into the current table overrides."""
+        return LOOP.run(self._table.update_table_overrides(**overrides))
+
+    def reset_table_overrides(self) -> dict[str, Any]:
+        """Clear Enterprise table overrides for this table."""
+        return LOOP.run(self._table.reset_table_overrides())
+
     # WAL-PK-FUSION: delete both hooks.
     def _hybrid_pk_fusion_learned(self) -> bool:
         # `self._table` is the async table, which owns the Rust handle.

python/python/lancedb/table.py

@@ -1931,6 +1931,27 @@ class Table(ABC):
             "create_index_async is not supported for this table type"
         )
 
+    def get_table_overrides(self) -> dict[str, Any]:
+        """Read Enterprise table overrides for this table.
+
+        Table overrides configure Enterprise-managed background behavior such as
+        automatic compaction, cleanup, and reindex execution. This method is
+        supported only on remote Enterprise tables.
+        """
+        raise NotImplementedError
+
+    def update_table_overrides(self, **overrides: Any) -> dict[str, Any]:
+        """Merge Enterprise table overrides into the current table overrides.
+
+        Only fields supplied in ``overrides`` are changed; unrelated existing
+        overrides are preserved by the remote service update flow.
+        """
+        raise NotImplementedError
+
+    def reset_table_overrides(self) -> dict[str, Any]:
+        """Clear Enterprise table overrides for this table."""
+        raise NotImplementedError
+
     def drop_index(self, name: str) -> None:
         """
         Drop an index from the table.
@@ -5193,6 +5214,36 @@ class LanceTable(Table):
         [`AsyncTable.get_rte_spec`][lancedb.AsyncTable.get_rte_spec]."""
         return LOOP.run(self._table.get_rte_spec())
 
+    def get_table_overrides(self) -> dict[str, Any]:
+        """Read Enterprise table overrides for this table.
+
+        Table overrides configure Enterprise-managed background behavior such as
+        automatic compaction, cleanup, and reindex execution. This method is
+        supported only on remote Enterprise tables.
+        """
+        return LOOP.run(self._table.get_table_overrides())
+
+    def update_table_overrides(self, **overrides: Any) -> dict[str, Any]:
+        """Merge Enterprise table overrides into the current table overrides.
+
+        Only fields supplied in ``overrides`` are changed; unrelated overrides
+        from the current read snapshot are preserved.
+
+        Examples
+        --------
+        >>> table.update_table_overrides(  # doctest: +SKIP
+        ...     job_types={
+        ...         "compaction": {"enabled": True},
+        ...         "cleanup": {"enabled": False},
+        ...     }
+        ... )
+        """
+        return LOOP.run(self._table.update_table_overrides(**overrides))
+
+    def reset_table_overrides(self) -> dict[str, Any]:
+        """Clear Enterprise table overrides for this table."""
+        return LOOP.run(self._table.reset_table_overrides())
+
     def checkpoint_rte(self) -> None:
         """Synchronous version of
         [`AsyncTable.checkpoint_rte`][lancedb.AsyncTable.checkpoint_rte]."""
@@ -5942,6 +5993,37 @@ class AsyncTable:
         """
         return await self._inner.get_rte_spec()
 
+    async def get_table_overrides(self) -> dict[str, Any]:
+        """Read Enterprise table overrides for this table.
+
+        Table overrides configure Enterprise-managed background behavior such as
+        automatic compaction, cleanup, and reindex execution. This method is
+        supported only on remote Enterprise tables.
+        """
+        return json.loads(await self._inner.get_table_overrides())
+
+    async def update_table_overrides(self, **overrides: Any) -> dict[str, Any]:
+        """Merge Enterprise table overrides into the current table overrides.
+
+        Only fields supplied in ``overrides`` are changed; unrelated overrides
+        from the current read snapshot are preserved.
+
+        Examples
+        --------
+        >>> await table.update_table_overrides(  # doctest: +SKIP
+        ...     job_types={
+        ...         "compaction": {"enabled": True},
+        ...         "cleanup": {"enabled": False},
+        ...     }
+        ... )
+        """
+        encoded = json.dumps(overrides)
+        return json.loads(await self._inner.update_table_overrides(encoded))
+
+    async def reset_table_overrides(self) -> dict[str, Any]:
+        """Clear Enterprise table overrides for this table."""
+        return json.loads(await self._inner.reset_table_overrides())
+
     # WAL-PK-FUSION: delete both hooks.
     def _hybrid_pk_fusion_learned(self) -> bool:
         """See [`Table._hybrid_pk_fusion_learned`][lancedb.table.Table]."""

python/python/tests/test_remote_db.py

@@ -17,9 +17,9 @@ import sys
 import textwrap
 import threading
 import time
+from urllib.parse import parse_qs, urlparse
 from unittest.mock import MagicMock, patch
 import uuid
-from urllib.parse import parse_qs, urlparse
 from packaging.version import Version
 
 import lancedb
@@ -55,6 +55,9 @@ def make_mock_http_handler(handler):
         def do_POST(self):
             handler(self)
 
+        def do_PUT(self):
+            handler(self)
+
     return MockLanceDBHandler
 
 
@@ -2346,6 +2349,118 @@ def test_checkpoint_rte_sync():
         assert called == ["flush_rte", "get_rte_stats"]
 
 
+@contextlib.contextmanager
+def table_overrides_test_table(overrides_handler):
+    def handler(request):
+        parsed = urlparse(request.path)
+        if parsed.path in (
+            "/admin/table/overrides",
+            "/admin/table/overrides/reset",
+        ):
+            overrides_handler(request, parsed)
+        elif parsed.path == "/v1/table/test/describe/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(b'{"version": 1, "schema": {"fields": []}}')
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        yield db.open_table("test")
+
+
+def assert_table_overrides_query(parsed):
+    query = parse_qs(parsed.query)
+    assert query == {"db": ["dev"], "table": ["test"]}
+
+
+def test_remote_table_exposes_table_overrides_methods():
+    from lancedb.remote.table import RemoteTable
+
+    for name in (
+        "get_table_overrides",
+        "update_table_overrides",
+        "reset_table_overrides",
+    ):
+        assert hasattr(RemoteTable, name)
+
+
+def test_get_table_overrides_sync_remote_table():
+    def overrides_handler(request, parsed):
+        assert request.command == "GET"
+        assert parsed.path == "/admin/table/overrides"
+        assert_table_overrides_query(parsed)
+        send_json(request, {"job_types": {"cleanup": {"enabled": False}}})
+
+    with table_overrides_test_table(overrides_handler) as table:
+        assert table.get_table_overrides() == {
+            "job_types": {"cleanup": {"enabled": False}}
+        }
+
+
+def test_update_table_overrides_sync_remote_table_merges_existing_fields():
+    calls = []
+
+    def overrides_handler(request, parsed):
+        calls.append(request.command)
+        assert parsed.path == "/admin/table/overrides"
+        if request.command == "GET":
+            assert_table_overrides_query(parsed)
+            send_json(
+                request,
+                {
+                    "job_types": {
+                        "compaction": {"enabled": True},
+                        "reindex_vector": {"table": {"num_workers": 3}},
+                    },
+                    "unknown": {"keep": True},
+                },
+            )
+        elif request.command == "PUT":
+            body = read_json_body(request)
+            assert body["db"] == "dev"
+            assert body["namespace"] == []
+            assert body["table"] == "test"
+            assert body["overrides"] == {
+                "job_types": {
+                    "compaction": {"enabled": True},
+                    "cleanup": {"enabled": False},
+                    "reindex_vector": {"table": {"num_workers": 3}},
+                },
+                "unknown": {"keep": True},
+            }
+            send_json(request, body["overrides"])
+        else:
+            raise AssertionError(f"unexpected method: {request.command}")
+
+    with table_overrides_test_table(overrides_handler) as table:
+        assert table.update_table_overrides(
+            job_types={"cleanup": {"enabled": False}}
+        ) == {
+            "job_types": {
+                "compaction": {"enabled": True},
+                "cleanup": {"enabled": False},
+                "reindex_vector": {"table": {"num_workers": 3}},
+            },
+            "unknown": {"keep": True},
+        }
+
+    assert calls == ["GET", "PUT"]
+
+
+def test_reset_table_overrides_sync_remote_table():
+    def overrides_handler(request, parsed):
+        assert request.command == "POST"
+        assert parsed.path == "/admin/table/overrides/reset"
+        assert_table_overrides_query(parsed)
+        send_json(request, {})
+
+    with table_overrides_test_table(overrides_handler) as table:
+        assert table.reset_table_overrides() == {}
+
+
 @contextlib.contextmanager
 def query_test_table(
     query_handler,

python/python/tests/test_table.py

@@ -1933,6 +1933,66 @@ def test_create_index_method(mock_create_index, mem_db: DBConnection):
     )
 
 
+@patch("lancedb.table.AsyncTable.get_table_overrides")
+def test_get_table_overrides_method(mock_get_table_overrides, mem_db: DBConnection):
+    expected = {"job_types": {"cleanup": {"enabled": False}}}
+    mock_get_table_overrides.return_value = expected
+    table = mem_db.create_table("test", data=[{"id": 1}])
+
+    assert table.get_table_overrides() == expected
+    mock_get_table_overrides.assert_called_once_with()
+
+
+@patch("lancedb.table.AsyncTable.update_table_overrides")
+def test_update_table_overrides_method(
+    mock_update_table_overrides, mem_db: DBConnection
+):
+    overrides = {
+        "job_types": {
+            "cleanup": {"enabled": False},
+            "index": {
+                "scalar": {
+                    "reindex": {
+                        "row_threshold": 1000,
+                        "ttl_millis": 300000,
+                    }
+                }
+            },
+        }
+    }
+    mock_update_table_overrides.return_value = overrides
+    table = mem_db.create_table("test", data=[{"id": 1}])
+
+    assert table.update_table_overrides(**overrides) == overrides
+    mock_update_table_overrides.assert_called_once_with(**overrides)
+
+
+@patch("lancedb.table.AsyncTable.reset_table_overrides")
+def test_reset_table_overrides_method(mock_reset_table_overrides, mem_db: DBConnection):
+    expected = {}
+    mock_reset_table_overrides.return_value = expected
+    table = mem_db.create_table("test", data=[{"id": 1}])
+
+    assert table.reset_table_overrides() == expected
+    mock_reset_table_overrides.assert_called_once_with()
+
+
+@pytest.mark.asyncio
+async def test_async_update_table_overrides_encodes_json():
+    class MockInner:
+        async def update_table_overrides(self, encoded_overrides):
+            assert json.loads(encoded_overrides) == {
+                "job_types": {"cleanup": {"enabled": False}}
+            }
+            return json.dumps({"job_types": {"cleanup": {"enabled": False}}})
+
+    table = lancedb.table.AsyncTable(MockInner())
+
+    assert await table.update_table_overrides(
+        job_types={"cleanup": {"enabled": False}}
+    ) == {"job_types": {"cleanup": {"enabled": False}}}
+
+
 @pytest.mark.parametrize("api", ["legacy", "unified"])
 @pytest.mark.parametrize("cuda_failure", ["unavailable", "init_error"])
 def test_create_index_cuda_failure_suggests_cpu(

python/src/table.rs

@@ -1849,6 +1849,45 @@ impl Table {
         })
     }
 
+    pub fn get_table_overrides(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let overrides = inner.get_table_overrides().await.infer_error()?;
+            serde_json::to_string(&overrides).map_err(|err| {
+                PyRuntimeError::new_err(format!("Failed to serialize table overrides: {err}"))
+            })
+        })
+    }
+
+    pub fn update_table_overrides<'a>(
+        self_: PyRef<'a, Self>,
+        overrides_json: String,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let overrides = serde_json::from_str(&overrides_json).map_err(|err| {
+                PyValueError::new_err(format!("Invalid table overrides JSON: {err}"))
+            })?;
+            let overrides = inner
+                .update_table_overrides(overrides)
+                .await
+                .infer_error()?;
+            serde_json::to_string(&overrides).map_err(|err| {
+                PyRuntimeError::new_err(format!("Failed to serialize table overrides: {err}"))
+            })
+        })
+    }
+
+    pub fn reset_table_overrides(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            let overrides = inner.reset_table_overrides().await.infer_error()?;
+            serde_json::to_string(&overrides).map_err(|err| {
+                PyRuntimeError::new_err(format!("Failed to serialize table overrides: {err}"))
+            })
+        })
+    }
+
     /// Whether a hybrid query on this table has already been refused `_rowid`.
     /// Learned from a previous refusal, never probed, so this is free.
     ///

rust/lancedb/src/remote/client.rs

@@ -824,6 +824,11 @@ impl<S: HttpSend> RestfulLanceDbClient<S> {
         self.client.post(full_uri)
     }
 
+    pub fn put(&self, uri: &str) -> RequestBuilder {
+        let full_uri = format!("{}{}", self.host, uri);
+        self.client.put(full_uri)
+    }
+
     /// Apply dynamic headers from the header provider if configured
     pub(crate) async fn apply_dynamic_headers(&self, mut request: Request) -> Result<Request> {
         if let Some(ref provider) = self.header_provider {

rust/lancedb/src/remote/db.rs

@@ -223,6 +223,7 @@ pub struct RemoteDatabase<S: HttpSend = Sender> {
     pub(super) client: RestfulLanceDbClient<S>,
     // Cache existence and server capabilities, not mutable per-handle table state.
     table_cache: Cache<String, ServerVersion>,
+    db_name: String,
     uri: String,
     /// Headers to pass to the namespace client for authentication
     namespace_headers: HashMap<String, String>,
@@ -435,6 +436,7 @@ impl RemoteDatabase {
         Ok(Self {
             client,
             table_cache,
+            db_name: parsed.db_name,
             uri: uri.to_owned(),
             namespace_headers,
             namespace_context_provider,
@@ -565,6 +567,7 @@ mod test_utils {
             Self {
                 client,
                 table_cache: Cache::new(0),
+                db_name: "default".to_string(),
                 uri: "http://localhost".to_string(),
                 namespace_headers: HashMap::new(),
                 namespace_context_provider: None,
@@ -594,6 +597,7 @@ mod test_utils {
             Self {
                 client,
                 table_cache: Cache::new(0),
+                db_name: "default".to_string(),
                 uri: "http://localhost".to_string(),
                 namespace_headers: config.extra_headers.clone(),
                 namespace_context_provider,
@@ -1444,6 +1448,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         let cache_key = build_cache_key(&request.name, &request.namespace_path);
         let table = Arc::new(RemoteTable::new_with_sql_client(
             self.client.clone(),
+            self.db_name.clone(),
             request.name.clone(),
             request.namespace_path.clone(),
             table_identifier,
@@ -1487,6 +1492,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         let cache_key = build_cache_key(&request.target_table_name, &request.target_namespace_path);
         let table = Arc::new(RemoteTable::new_with_sql_client(
             self.client.clone(),
+            self.db_name.clone(),
             request.target_table_name.clone(),
             request.target_namespace_path.clone(),
             table_identifier,
@@ -1536,6 +1542,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
 
             Ok(Arc::new(RemoteTable::new_with_sql_client(
                 self.client.clone(),
+                self.db_name.clone(),
                 request.name,
                 request.namespace_path,
                 identifier,
@@ -1555,6 +1562,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
             let describe_body = rsp.text().await.ok();
             let table = Arc::new(RemoteTable::new_with_sql_client(
                 self.client.clone(),
+                self.db_name.clone(),
                 request.name.clone(),
                 request.namespace_path.clone(),
                 identifier,

rust/lancedb/src/remote/sql_test.rs

@@ -1098,6 +1098,7 @@ async fn remote_refresh_omits_unknown_metrics_and_forwards_full() {
     });
     let table = RemoteTable::new_with_sql_client(
         client,
+        "analytics".into(),
         "view".into(),
         vec!["ns".into()],
         "ns$view".into(),
@@ -1204,6 +1205,7 @@ async fn cancelling_remote_refresh_cancels_sql_query() {
     });
     let table = RemoteTable::new_with_sql_client(
         client,
+        "analytics".into(),
         "cancel_view".into(),
         vec![],
         "cancel_view".into(),
@@ -2157,6 +2159,7 @@ async fn remote_describe_materialized_view_preserves_the_sql_definition() {
             client_with_handler(|request| -> http::Response<String> {
                 panic!("unexpected REST request: {}", request.url().path());
             }),
+            "analytics".into(),
             "daily_sales".into(),
             vec!["analytics".into()],
             "analytics$daily_sales".into(),

rust/lancedb/src/remote/table.rs

@@ -491,6 +491,7 @@ impl<S: HttpSend + 'static> Tags for RemoteTags<'_, S> {
 
 pub struct RemoteTable<S: HttpSend = Sender> {
     client: RestfulLanceDbClient<S>,
+    db_name: String,
     name: String,
     namespace: Vec<String>,
     identifier: String,
@@ -519,6 +520,48 @@ impl<S: HttpSend> std::fmt::Debug for RemoteTable<S> {
 }
 
 impl<S: HttpSend> RemoteTable<S> {
+    fn table_overrides_query(&self) -> Vec<(&str, &str)> {
+        let mut query = Vec::with_capacity(3 + self.namespace.len());
+        query.push(("db", self.db_name.as_str()));
+        query.push(("table", self.name.as_str()));
+        for namespace in &self.namespace {
+            query.push(("namespace", namespace.as_str()));
+        }
+        if let Some(branch) = self.branch.as_deref() {
+            query.push(("branch", branch));
+        }
+        query
+    }
+
+    fn table_overrides_set_body(&self, overrides: serde_json::Value) -> serde_json::Value {
+        let mut body = serde_json::json!({
+            "db": &self.db_name,
+            "namespace": &self.namespace,
+            "table": &self.name,
+            "overrides": overrides,
+        });
+        if let Some(branch) = &self.branch {
+            body["branch"] = serde_json::Value::String(branch.clone());
+        }
+        body
+    }
+
+    fn merge_table_overrides(
+        target: &mut serde_json::Map<String, serde_json::Value>,
+        patch: serde_json::Map<String, serde_json::Value>,
+    ) {
+        for (key, value) in patch {
+            match (target.get_mut(&key), value) {
+                (Some(serde_json::Value::Object(existing)), serde_json::Value::Object(patch)) => {
+                    Self::merge_table_overrides(existing, patch);
+                }
+                (_, value) => {
+                    target.insert(key, value);
+                }
+            }
+        }
+    }
+
     async fn submit_create_index(&self, mut index: IndexBuilder) -> Result<Option<String>> {
         self.check_mutable().await?;
         let request = self
@@ -686,6 +729,7 @@ impl<S: HttpSend> RemoteTable<S> {
 
     pub(super) fn new_with_sql_client(
         client: RestfulLanceDbClient<S>,
+        db_name: String,
         name: String,
         namespace: Vec<String>,
         identifier: String,
@@ -694,6 +738,7 @@ impl<S: HttpSend> RemoteTable<S> {
     ) -> Self {
         Self {
             client,
+            db_name,
             name,
             namespace,
             identifier,
@@ -729,6 +774,7 @@ impl<S: HttpSend> RemoteTable<S> {
     fn with_branch(&self, branch: Option<String>) -> Self {
         Self {
             client: self.client.clone(),
+            db_name: self.db_name.clone(),
             name: self.name.clone(),
             namespace: self.namespace.clone(),
             identifier: self.identifier.clone(),
@@ -1705,6 +1751,7 @@ mod test_utils {
             let client = client_with_handler(handler);
             Self {
                 client,
+                db_name: "default".to_string(),
                 name: name.clone(),
                 namespace: vec![],
                 identifier: name,
@@ -1731,6 +1778,7 @@ mod test_utils {
             let client = client_with_handler_and_interval(handler, read_consistency_interval);
             Self {
                 client,
+                db_name: "default".to_string(),
                 name: name.clone(),
                 namespace: vec![],
                 identifier: name,
@@ -1766,6 +1814,7 @@ mod test_utils {
             let client = client_with_handler_and_config(handler, config);
             Self {
                 client,
+                db_name: "default".to_string(),
                 name: name.clone(),
                 namespace: vec![],
                 identifier: name,
@@ -2230,6 +2279,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
         );
         let namespace = self.namespace.clone();
         let client = self.client.clone();
+        let db_name = self.db_name.clone();
         let name = self.name.clone();
         let identifier = self.identifier.clone();
         let server_version = self.server_version.clone();
@@ -2240,6 +2290,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
             .submit_as_job(statement, namespace.clone(), move || async move {
                 let table = Self::new_with_sql_client(
                     client,
+                    db_name,
                     name,
                     namespace,
                     identifier,
@@ -3396,6 +3447,56 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
         Ok(Some(spec))
     }
 
+    async fn get_table_overrides(&self) -> Result<serde_json::Value> {
+        let request = self
+            .client
+            .get("/admin/table/overrides")
+            .query(&self.table_overrides_query());
+        let (request_id, response) = self.send(request, true).await?;
+        let response = self.client.check_response(&request_id, response).await?;
+        response.json().await.err_to_http(request_id)
+    }
+
+    async fn update_table_overrides(
+        &self,
+        overrides: serde_json::Value,
+    ) -> Result<serde_json::Value> {
+        self.check_mutable().await?;
+        let serde_json::Value::Object(patch) = overrides else {
+            return Err(Error::InvalidInput {
+                message: "table overrides update must be a JSON object".into(),
+            });
+        };
+
+        let current = self.get_table_overrides().await?;
+        let mut current = match current {
+            serde_json::Value::Object(current) => current,
+            other => {
+                return Err(Error::InvalidInput {
+                    message: format!("table overrides response must be an object, got {other}"),
+                });
+            }
+        };
+        Self::merge_table_overrides(&mut current, patch);
+        let merged = serde_json::Value::Object(current);
+        let body = self.table_overrides_set_body(merged);
+        let request = self.client.put("/admin/table/overrides").json(&body);
+        let (request_id, response) = self.send(request, true).await?;
+        let response = self.client.check_response(&request_id, response).await?;
+        response.json().await.err_to_http(request_id)
+    }
+
+    async fn reset_table_overrides(&self) -> Result<serde_json::Value> {
+        self.check_mutable().await?;
+        let request = self
+            .client
+            .post("/admin/table/overrides/reset")
+            .query(&self.table_overrides_query());
+        let (request_id, response) = self.send(request, true).await?;
+        let response = self.client.check_response(&request_id, response).await?;
+        response.json().await.err_to_http(request_id)
+    }
+
     // WAL-PK-FUSION: delete both.
     fn hybrid_pk_fusion_learned(&self) -> bool {
         self.wal_pk_fusion.learned()
@@ -11056,6 +11157,116 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_get_table_overrides() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "GET");
+            assert_eq!(request.url().path(), "/admin/table/overrides");
+            assert_eq!(request.url().query(), Some("db=default&table=my_table"));
+            http::Response::builder()
+                .status(200)
+                .body(
+                    serde_json::json!({
+                        "job_types": {
+                            "cleanup": { "enabled": false }
+                        }
+                    })
+                    .to_string(),
+                )
+                .unwrap()
+        });
+
+        let overrides = table.get_table_overrides().await.unwrap();
+        assert_eq!(overrides["job_types"]["cleanup"]["enabled"], false);
+    }
+
+    #[tokio::test]
+    async fn test_update_table_overrides_merges_existing_fields() {
+        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = calls.clone();
+        let table = Table::new_with_handler("my_table", move |request| {
+            match seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) {
+                0 => {
+                    assert_eq!(request.method(), "GET");
+                    assert_eq!(request.url().path(), "/admin/table/overrides");
+                    http::Response::builder()
+                        .status(200)
+                        .body(
+                            serde_json::json!({
+                                "job_types": {
+                                    "compaction": { "enabled": true },
+                                    "reindex_vector": {
+                                        "table": { "num_workers": 3 }
+                                    }
+                                },
+                                "unknown": { "keep": true }
+                            })
+                            .to_string(),
+                        )
+                        .unwrap()
+                }
+                1 => {
+                    assert_eq!(request.method(), "PUT");
+                    assert_eq!(request.url().path(), "/admin/table/overrides");
+                    let body = request_body_json(&request);
+                    assert_eq!(body["db"], "default");
+                    assert_eq!(body["table"], "my_table");
+                    assert_eq!(body["namespace"], serde_json::json!([]));
+                    assert_eq!(
+                        body["overrides"],
+                        serde_json::json!({
+                            "job_types": {
+                                "compaction": { "enabled": true },
+                                "cleanup": { "enabled": false },
+                                "reindex_vector": {
+                                    "table": { "num_workers": 3 }
+                                }
+                            },
+                            "unknown": { "keep": true }
+                        })
+                    );
+                    http::Response::builder()
+                        .status(200)
+                        .body(body["overrides"].to_string())
+                        .unwrap()
+                }
+                _ => panic!("unexpected request"),
+            }
+        });
+
+        let updated = table
+            .update_table_overrides(serde_json::json!({
+                "job_types": {
+                    "cleanup": { "enabled": false }
+                }
+            }))
+            .await
+            .unwrap();
+        assert_eq!(updated["job_types"]["cleanup"]["enabled"], false);
+        assert_eq!(updated["job_types"]["compaction"]["enabled"], true);
+        assert_eq!(
+            updated["job_types"]["reindex_vector"]["table"]["num_workers"],
+            3
+        );
+        assert_eq!(updated["unknown"]["keep"], true);
+        assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2);
+    }
+
+    #[tokio::test]
+    async fn test_reset_table_overrides() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            assert_eq!(request.url().path(), "/admin/table/overrides/reset");
+            assert_eq!(request.url().query(), Some("db=default&table=my_table"));
+            http::Response::builder().status(200).body("{}").unwrap()
+        });
+
+        assert_eq!(
+            table.reset_table_overrides().await.unwrap(),
+            serde_json::json!({})
+        );
+    }
+
     #[tokio::test]
     async fn test_get_rte_spec_absent() {
         let table = Table::new_with_handler("my_table", |request| {

rust/lancedb/src/table.rs

@@ -42,6 +42,7 @@ use lance_table::io::commit::CommitHandler;
 use lance_table::io::commit::ManifestNamingScheme;
 use lance_table::io::commit::external_manifest::ExternalManifestCommitHandler;
 use serde::{Deserialize, Serialize};
+pub type TableOverrides = serde_json::Value;
 use std::collections::{HashMap, HashSet};
 use std::format;
 use std::path::Path;
@@ -710,6 +711,34 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
             message: "get_rte_spec is not supported on this table type".into(),
         })
     }
+    /// Read Enterprise table overrides.
+    ///
+    /// The default implementation returns `NotSupported`; Enterprise remote
+    /// tables override it.
+    async fn get_table_overrides(&self) -> Result<TableOverrides> {
+        Err(Error::NotSupported {
+            message: "get_table_overrides is supported only on remote Enterprise tables".into(),
+        })
+    }
+    /// Merge Enterprise table overrides into the currently stored object.
+    ///
+    /// The default implementation returns `NotSupported`; Enterprise remote
+    /// tables override it.
+    async fn update_table_overrides(&self, _overrides: TableOverrides) -> Result<TableOverrides> {
+        Err(Error::NotSupported {
+            message: "update_table_overrides is supported only on remote Enterprise tables".into(),
+        })
+    }
+    /// Clear Enterprise table overrides.
+    ///
+    /// The default implementation returns `NotSupported`; Enterprise remote
+    /// tables override it.
+    async fn reset_table_overrides(&self) -> Result<TableOverrides> {
+        Err(Error::NotSupported {
+            message: "reset_table_overrides is supported only on remote Enterprise tables".into(),
+        })
+    }
+
     /// Whether a hybrid query on this table has already been told it cannot
     /// join its legs on `_rowid`.
     ///
@@ -735,6 +764,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
     /// the only cost of being late to notice is a base-only read that is still
     /// correct.
     fn note_hybrid_pk_fusion(&self) {}
+
     /// Freeze every table shard's active memtable into an SSTable.
     ///
     /// The default implementation returns `NotSupported`.
@@ -2110,6 +2140,33 @@ impl Table {
         self.inner.get_rte_spec().await
     }
 
+    /// Read Enterprise table overrides for this table.
+    ///
+    /// Table overrides configure Enterprise-managed background behavior such as
+    /// automatic compaction, cleanup, and reindex execution. Local tables return
+    /// `NotSupported` because they do not have an Enterprise background planner
+    /// consuming these overrides.
+    pub async fn get_table_overrides(&self) -> Result<TableOverrides> {
+        self.inner.get_table_overrides().await
+    }
+
+    /// Merge Enterprise table overrides into the current stored overrides.
+    ///
+    /// This method uses patch semantics: fields present in `overrides` are
+    /// recursively merged into the existing table overrides and unrelated fields
+    /// are preserved.
+    pub async fn update_table_overrides(
+        &self,
+        overrides: TableOverrides,
+    ) -> Result<TableOverrides> {
+        self.inner.update_table_overrides(overrides).await
+    }
+
+    /// Clear Enterprise table overrides for this table.
+    pub async fn reset_table_overrides(&self) -> Result<TableOverrides> {
+        self.inner.reset_table_overrides().await
+    }
+
     /// Converge this table's RTE write path into its base table.
     ///
     /// One `flush` to freeze every memtable into an SSTable, then compaction triggers