diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 239e80cdf..656a2be5c 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/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: ... diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py index 7454b3e73..318c798c6 100644 --- a/python/python/lancedb/remote/table.py +++ b/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. diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index b68f15912..d60e89ad6 100644 --- a/python/python/lancedb/table.py +++ b/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].""" diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index acbe7542c..77e76684c 100644 --- a/python/python/tests/test_remote_db.py +++ b/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, diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index f4efb1787..88f5f834c 100644 --- a/python/python/tests/test_table.py +++ b/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( diff --git a/python/src/table.rs b/python/src/table.rs index f484b3f77..440e494b7 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -1849,6 +1849,45 @@ impl Table { }) } + pub fn get_table_overrides(self_: PyRef<'_, Self>) -> PyResult> { + 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> { + 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> { + 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. /// diff --git a/rust/lancedb/src/remote/client.rs b/rust/lancedb/src/remote/client.rs index ae4c1aaa5..cec6696c9 100644 --- a/rust/lancedb/src/remote/client.rs +++ b/rust/lancedb/src/remote/client.rs @@ -824,6 +824,11 @@ impl RestfulLanceDbClient { 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 { if let Some(ref provider) = self.header_provider { diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 8098f70ba..99961986b 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -223,6 +223,7 @@ pub struct RemoteDatabase { pub(super) client: RestfulLanceDbClient, // Cache existence and server capabilities, not mutable per-handle table state. table_cache: Cache, + db_name: String, uri: String, /// Headers to pass to the namespace client for authentication namespace_headers: HashMap, @@ -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 Database for RemoteDatabase { 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 Database for RemoteDatabase { 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 Database for RemoteDatabase { 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 Database for RemoteDatabase { 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, diff --git a/rust/lancedb/src/remote/sql_test.rs b/rust/lancedb/src/remote/sql_test.rs index 3a701d2e6..228e2ddd3 100644 --- a/rust/lancedb/src/remote/sql_test.rs +++ b/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 { panic!("unexpected REST request: {}", request.url().path()); }), + "analytics".into(), "daily_sales".into(), vec!["analytics".into()], "analytics$daily_sales".into(), diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 4a7e384eb..681ea453e 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -491,6 +491,7 @@ impl Tags for RemoteTags<'_, S> { pub struct RemoteTable { client: RestfulLanceDbClient, + db_name: String, name: String, namespace: Vec, identifier: String, @@ -519,6 +520,48 @@ impl std::fmt::Debug for RemoteTable { } impl RemoteTable { + 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, + patch: serde_json::Map, + ) { + 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> { self.check_mutable().await?; let request = self @@ -686,6 +729,7 @@ impl RemoteTable { pub(super) fn new_with_sql_client( client: RestfulLanceDbClient, + db_name: String, name: String, namespace: Vec, identifier: String, @@ -694,6 +738,7 @@ impl RemoteTable { ) -> Self { Self { client, + db_name, name, namespace, identifier, @@ -729,6 +774,7 @@ impl RemoteTable { fn with_branch(&self, branch: Option) -> 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 BaseTable for RemoteTable { ); 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 BaseTable for RemoteTable { .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 BaseTable for RemoteTable { Ok(Some(spec)) } + async fn get_table_overrides(&self) -> Result { + 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 { + 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 { + 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| { diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 6f1c92147..180efb410 100644 --- a/rust/lancedb/src/table.rs +++ b/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 { + 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 { + 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 { + 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 { + 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 { + self.inner.update_table_overrides(overrides).await + } + + /// Clear Enterprise table overrides for this table. + pub async fn reset_table_overrides(&self) -> Result { + 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