diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index 188769747..1226d5a1c 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -943,6 +943,12 @@ Modeled after ``VACUUM`` in PostgreSQL. you have added or modified 100,000 or more records or run more than 20 data modification operations. +On a LanceDB Enterprise (remote) table, only compaction is currently +supported. Pruning old versions and index optimization are silently +skipped there instead of being attempted, so this call resolves +successfully, with zeroed pruning statistics, once the compaction job +has been submitted; it does not wait for that job to finish. + #### Parameters * **options?**: `Partial`<[`OptimizeOptions`](../interfaces/OptimizeOptions.md)> diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 87eb8d37e..6340a4e4c 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -348,9 +348,14 @@ index statistics, and search. For example, `table.delete(where="id = 1")`, `create_scalar_index` defaults to `replace=True` on both backends. Both scalar and full-text index creation accept `wait_timeout`. LanceDB Enterprise maintenance methods -accept the LanceDB Embedded options: cleanup and compaction warn that they are no-ops, -while `optimize` reports an unsupported operation. Removed Tantivy options -produce the same explanatory errors on both backends. +accept the LanceDB Embedded options: cleanup and compaction warn that they are no-ops. +`optimize` submits a durable compaction job and returns once that submission is +accepted, not once compaction itself finishes. Pruning and index optimization +are not yet supported remotely, so it silently skips them and returns with +zeroed pruning statistics rather than attempting and failing. A non-default +`compaction_options` raises `NotImplementedError` before anything is +submitted, since there is no way yet to forward those settings to the server. +Removed Tantivy options produce the same explanatory errors on both backends. The old LanceDB Enterprise keyword names `predicate` for `delete`, `index_name` for `drop_index`, `index_uuid` for `index_stats`, and `column` for `create_fts_index` diff --git a/nodejs/__test__/remote.test.ts b/nodejs/__test__/remote.test.ts index 9139f2686..b11a79a67 100644 --- a/nodejs/__test__/remote.test.ts +++ b/nodejs/__test__/remote.test.ts @@ -312,7 +312,6 @@ describe("remote connection", () => { const cases: [() => Promise, string][] = [ [() => table.usesV2ManifestPaths(), "uses_v2_manifest_paths"], [() => table.migrateManifestPathsV2(), "migrate_manifest_paths_v2"], - [() => table.optimize(), "optimize"], [ () => table.createIndex("vector", { config: Index.hnswPq() }), "IVF_HNSW_PQ", diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index 4861bf3f0..e84704803 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -1019,6 +1019,12 @@ export abstract class Table { * optimize should be run frequently. A good rule of thumb is to run optimize if * you have added or modified 100,000 or more records or run more than 20 data * modification operations. + * + * On a LanceDB Enterprise (remote) table, only compaction is currently + * supported. Pruning old versions and index optimization are silently + * skipped there instead of being attempted, so this call resolves + * successfully, with zeroed pruning statistics, once the compaction job + * has been submitted; it does not wait for that job to finish. */ abstract optimize(options?: Partial): Promise; /** diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index a08c7664e..9c9fd5156 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -9,7 +9,7 @@ use lancedb::ipc::{ipc_file_to_batches, ipc_file_to_schema}; use lancedb::table::{ AddDataMode, ColumnAlteration as LanceColumnAlteration, FieldMetadataUpdate as LanceFieldMetadataUpdate, FtsToken as LanceDbFtsToken, - NewColumnTransform, OptimizeAction, OptimizeOptions, Ref, Table as LanceDbTable, + NewColumnTransform, Ref, Table as LanceDbTable, }; use napi::Env; use napi::bindgen_prelude::*; @@ -735,37 +735,17 @@ impl Table { }) .transpose()?; - let compaction_stats = inner - .optimize(OptimizeAction::Compact { - options: lancedb::table::CompactionOptions::default(), - remap_options: None, - }) - .await - .default_error()? - .compaction - .unwrap(); - let prune_stats = if let Some(before_timestamp) = before_timestamp { - inner - .optimize_prune_before(before_timestamp, delete_unverified, None) - .await - } else { - inner - .optimize(OptimizeAction::Prune { - older_than: None, - delete_unverified, - error_if_tagged_old_versions: None, - }) - .await - } - .default_error()? - .prune - .unwrap(); - inner - .optimize(lancedb::table::OptimizeAction::Index( - OptimizeOptions::default(), - )) + let stats = inner + .optimize_full( + lancedb::table::CompactionOptions::default(), + None, + before_timestamp, + delete_unverified, + ) .await .default_error()?; + let compaction_stats = stats.compaction.unwrap(); + let prune_stats = stats.prune.unwrap(); Ok(OptimizeStats { compaction: CompactionStats { files_added: compaction_stats.files_added as i64, diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py index dcea1f8eb..db74c5375 100644 --- a/python/python/lancedb/remote/table.py +++ b/python/python/lancedb/remote/table.py @@ -1083,12 +1083,21 @@ class RemoteTable(Table): retrain: bool = False, compaction_options: Optional[CompactionOptions] = None, ): - """optimize() is not supported in LanceDB Enterprise.""" + """Compact the table's data files via a durable, asynchronous + Enterprise job. Pruning old versions and index optimization are not + yet supported remotely, so this call skips them and returns + successfully with zeroed pruning statistics once the compaction job + has been submitted; it does not wait for that job to finish. Passing + a non-default ``compaction_options`` raises ``NotImplementedError`` + before anything is submitted, since there is no way yet to forward + those settings to the server. + """ return LOOP.run( self._table.optimize( cleanup_older_than=cleanup_older_than, delete_unverified=delete_unverified, retrain=retrain, + compaction_options=compaction_options, ) ) diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index e19f2e70f..23bc25b97 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -2935,6 +2935,17 @@ class Table(ABC): """ Optimize the on-disk data and indices for better performance. + .. note:: + + On a LanceDB Enterprise (remote) table, only compaction is + currently supported. Pruning old versions and index optimization + are silently skipped there instead of being attempted, so this + call returns successfully, with zeroed pruning statistics, once + the compaction job has been submitted; it does not wait for that + job to finish. A non-default ``compaction_options`` raises + ``NotImplementedError`` before anything is submitted, since + there is no way yet to forward those settings to the server. + Modeled after ``VACUUM`` in PostgreSQL. Optimization covers three operations: @@ -8195,6 +8206,17 @@ class AsyncTable: """ Optimize the on-disk data and indices for better performance. + .. note:: + + On a LanceDB Enterprise (remote) table, only compaction is + currently supported. Pruning old versions and index optimization + are silently skipped there instead of being attempted, so this + call returns successfully, with zeroed pruning statistics, once + the compaction job has been submitted; it does not wait for that + job to finish. A non-default ``compaction_options`` raises + ``NotImplementedError`` before anything is submitted, since + there is no way yet to forward those settings to the server. + Modeled after ``VACUUM`` in PostgreSQL. Optimization covers three operations: diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index 84b973d90..300c4f849 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -2494,7 +2494,6 @@ def remote_unsupported_operations(): return [ ("uses_v2_manifest_paths", (), {}, "uses_v2_manifest_paths"), ("migrate_v2_manifest_paths", (), {}, "migrate_manifest_paths_v2"), - ("optimize", (), {}, "optimize"), ("create_index", ("vector",), {"config": HnswPq()}, "IVF_HNSW_PQ"), ] @@ -5626,3 +5625,134 @@ def test_local_connections_refuse_property_graphs(tmp_path): list(db.list_property_graphs()) with pytest.raises(NotImplementedError, match=refused): db.drop_property_graph("social") + + +def _create_table_handler(request): + if request.path == "/v1/table/test/create/?mode=create": + request.send_response(200) + request.send_header("Content-Type", "application/json") + request.end_headers() + request.wfile.write(b"{}") + else: + request.send_response(404) + request.end_headers() + + +@contextlib.contextmanager +def _mock_remote_table_with_sql(flight_server): + """A remote table backed by a mock REST server (for create/open) and the + given mock Flight SQL server (for the SQL statements `optimize()` submits + as jobs).""" + with http.server.ThreadingHTTPServer( + ("127.0.0.1", 0), make_mock_http_handler(_create_table_handler) + ) as rest_server: + rest_port = rest_server.server_address[1] + handle = start_server_thread(rest_server) + try: + with flight_server: + db = lancedb.connect( + "db://dev", + api_key="fake", + host_override=f"http://127.0.0.1:{rest_port}", + sql_host_override=f"grpc://127.0.0.1:{flight_server.port}", + ) + yield db.create_table("test", [{"id": 1}]) + finally: + rest_server.shutdown() + handle.join() + + +def test_remote_optimize_skips_prune_and_index(): + """Regression test: a remote table's `optimize()` used to submit a + compaction job and then raise `NotImplementedError` once it went on to + request pruning, which is not supported remotely. Retrying that failure + resubmitted another compaction job. It should instead succeed, with + zeroed pruning statistics, once compaction is submitted. + + The SQL job submission for a query without parameters (like + `OPTIMIZE TABLE ... REWRITE DATA`) goes over plain Flight `do_get`, not + `do_exchange`: the client sends a ticket wrapping the statement and reads + back whatever the server returns as the job's result stream. + """ + flight = pytest.importorskip("pyarrow.flight") + + class AckServer(flight.FlightServerBase): + def __init__(self): + super().__init__("grpc://127.0.0.1:0") + self.tickets = [] + + def do_get(self, context, ticket): + self.tickets.append(ticket.ticket) + return flight.RecordBatchStream( + pa.table({"ok": pa.array([], type=pa.bool_())}) + ) + + server = AckServer() + with _mock_remote_table_with_sql(server) as table: + stats = table.optimize() + + assert len(server.tickets) == 1 + assert b"OPTIMIZE TABLE" in server.tickets[0] + assert stats.prune.bytes_removed == 0 + assert stats.prune.old_versions_removed == 0 + + +def test_remote_optimize_rejects_non_default_compaction_options_before_submitting(): + """Regression test: `RemoteTable.optimize()` used to drop + `compaction_options` instead of forwarding it to the Rust layer, so a + non-default value was silently ignored -- the server-side guard against + unsupported options never saw it. It should now reach that guard and + raise before any SQL statement is submitted.""" + from lancedb.table import CompactionOptions + + flight = pytest.importorskip("pyarrow.flight") + + class UnreachableServer(flight.FlightServerBase): + def __init__(self): + super().__init__("grpc://127.0.0.1:0") + self.tickets = [] + + def do_get(self, context, ticket): + self.tickets.append(ticket.ticket) + return flight.RecordBatchStream( + pa.table({"ok": pa.array([], type=pa.bool_())}) + ) + + server = UnreachableServer() + options: CompactionOptions = {"target_rows_per_fragment": 5} + with _mock_remote_table_with_sql(server) as table: + with pytest.raises(NotImplementedError, match="CompactionOptions"): + table.optimize(compaction_options=options) + + assert server.tickets == [] + + +def test_remote_optimize_rejects_explicit_pruning_before_submitting(): + """Regression test: a remote table's `optimize()` used to silently drop + an explicit pruning request (`cleanup_older_than`), returning success + with zero bytes removed even though nothing was pruned. It should now + raise before any SQL statement is submitted, the same way a non-default + `compaction_options` does.""" + from datetime import timedelta + + flight = pytest.importorskip("pyarrow.flight") + + class UnreachableServer(flight.FlightServerBase): + def __init__(self): + super().__init__("grpc://127.0.0.1:0") + self.tickets = [] + + def do_get(self, context, ticket): + self.tickets.append(ticket.ticket) + return flight.RecordBatchStream( + pa.table({"ok": pa.array([], type=pa.bool_())}) + ) + + server = UnreachableServer() + with _mock_remote_table_with_sql(server) as table: + with pytest.raises(NotImplementedError, match="pruning"): + table.optimize(cleanup_older_than=timedelta(days=1)) + with pytest.raises(NotImplementedError, match="pruning"): + table.optimize(delete_unverified=True) + + assert server.tickets == [] diff --git a/python/python/tests/test_table_signatures.py b/python/python/tests/test_table_signatures.py index 346935b2b..387a89f31 100644 --- a/python/python/tests/test_table_signatures.py +++ b/python/python/tests/test_table_signatures.py @@ -192,7 +192,10 @@ def test_remote_optimize_options(): with pytest.raises(NotImplementedError, match="remote optimize"): table.optimize(cleanup_older_than=age, delete_unverified=True, retrain=True) handle.optimize.assert_awaited_once_with( - cleanup_older_than=age, delete_unverified=True, retrain=True + cleanup_older_than=age, + delete_unverified=True, + retrain=True, + compaction_options=None, ) diff --git a/python/src/table.rs b/python/src/table.rs index 38b3a7442..6ef524310 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -29,8 +29,8 @@ use lancedb::connection::LanceFileVersion; use lancedb::index::scalar::FtsIndexBuilder; use lancedb::table::{ AddDataMode, ColumnAlteration, CompactionMode, CompactionOptions, Duration, - FieldMetadataUpdate, FtsToken as LanceDbFtsToken, IndexRemapMode, NewColumnTransform, - OptimizeAction, OptimizeOptions, Ref, RefreshColumnOptions, Table as LanceDbTable, + FieldMetadataUpdate, FtsToken as LanceDbFtsToken, IndexRemapMode, NewColumnTransform, Ref, + RefreshColumnOptions, Table as LanceDbTable, }; use lancedb::tokenize as lancedb_tokenize; use pyo3::{ @@ -1783,31 +1783,12 @@ impl Table { None }; future_into_py(self_.py(), async move { - let compaction_stats = inner - .optimize(OptimizeAction::Compact { - options: compaction_options, - remap_options: None, - }) - .await - .infer_error()? - .compaction - .unwrap(); - let prune_stats = inner - .optimize(OptimizeAction::Prune { - older_than, - delete_unverified, - error_if_tagged_old_versions: None, - }) - .await - .infer_error()? - .prune - .unwrap(); - inner - .optimize(lancedb::table::OptimizeAction::Index( - OptimizeOptions::default(), - )) + let stats = inner + .optimize_full(compaction_options, older_than, None, delete_unverified) .await .infer_error()?; + let compaction_stats = stats.compaction.unwrap(); + let prune_stats = stats.prune.unwrap(); Ok(OptimizeStats { compaction: CompactionStats { files_added: compaction_stats.files_added as u64, diff --git a/rust/lancedb/src/remote/sql_test.rs b/rust/lancedb/src/remote/sql_test.rs index 7031b1422..ff06ae9d6 100644 --- a/rust/lancedb/src/remote/sql_test.rs +++ b/rust/lancedb/src/remote/sql_test.rs @@ -2277,6 +2277,266 @@ async fn remote_drop_materialized_view_uses_a_query_job() { } } +#[tokio::test] +async fn remote_optimize_submits_a_compaction_job_over_sql() { + let service = TestSqlService { + adaptive: true, + ..Default::default() + }; + let statements = service.statements.clone(); + let headers = service.headers.clone(); + let (sql_client, shutdown) = start_sql_service(service, ClientConfig::default()).await; + let client = client_with_handler(|request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()); + }); + let table = RemoteTable::new_with_sql_client( + client, + "analytics".into(), + "events".into(), + vec!["team".into(), "reports".into()], + "team$reports$events".into(), + Default::default(), + Some(sql_client), + ); + + let stats = table + .optimize(crate::table::optimize::OptimizeAction::Compact { + options: crate::table::optimize::CompactionOptions::default(), + remap_options: None, + }) + .await + .unwrap(); + + // The SQL job adapter does not expose compaction metrics, so + // `compaction` is a zeroed placeholder, not real numbers -- but it must + // be `Some`: callers (including the Python/Node bindings) unwrap it + // unconditionally on success. + assert_eq!( + stats.compaction, + Some(lance::dataset::optimize::CompactionMetrics::default()) + ); + assert!(stats.prune.is_none()); + assert_eq!( + *statements.lock().unwrap(), + ["OPTIMIZE TABLE \"events\" REWRITE DATA"] + ); + assert!( + headers + .lock() + .unwrap() + .iter() + .all(|header| header.namespace_path == "team$reports") + ); + shutdown.send(()).unwrap(); +} + +#[tokio::test] +async fn remote_optimize_on_a_branch_submits_a_with_clause_over_sql() { + let service = TestSqlService { + adaptive: true, + ..Default::default() + }; + let statements = service.statements.clone(); + let (sql_client, shutdown) = start_sql_service(service, ClientConfig::default()).await; + let client = client_with_handler(|request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()); + }); + let table = RemoteTable::new_with_sql_client( + client, + "analytics".into(), + "events".into(), + vec!["team".into(), "reports".into()], + "team$reports$events".into(), + Default::default(), + Some(sql_client), + ) + .with_branch(Some("o'brien".to_string())); + + table + .optimize(crate::table::optimize::OptimizeAction::Compact { + options: crate::table::optimize::CompactionOptions::default(), + remap_options: None, + }) + .await + .unwrap(); + + assert_eq!( + *statements.lock().unwrap(), + ["OPTIMIZE TABLE \"events\" REWRITE DATA WITH (branch = 'o''brien')"] + ); + shutdown.send(()).unwrap(); +} + +/// Regression test for the language bindings' `optimize()` convenience +/// method, which used to submit a compaction job and then fail on the next +/// action because `Prune` and `Index` are not supported remotely -- so a +/// retry on that failure resubmitted another compaction job. `optimize_full` +/// is what both the Python and Node bindings now call instead of +/// orchestrating `Compact`, `Prune`, and `Index` themselves; on a remote +/// table it must submit only the compaction job and return zeroed prune +/// stats rather than attempting `Prune` and `Index` at all. +#[tokio::test] +async fn remote_optimize_full_skips_prune_and_index() { + let service = TestSqlService { + adaptive: true, + ..Default::default() + }; + let statements = service.statements.clone(); + let (sql_client, shutdown) = start_sql_service(service, ClientConfig::default()).await; + let client = client_with_handler(|request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()); + }); + let remote_table = RemoteTable::new_with_sql_client( + client, + "analytics".into(), + "events".into(), + vec!["team".into(), "reports".into()], + "team$reports$events".into(), + Default::default(), + Some(sql_client), + ); + let database = Arc::new(RemoteDatabase::new_mock( + |request: reqwest::Request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()); + }, + )); + let table = crate::Table::new(Arc::new(remote_table), database); + + let stats = table + .optimize_full( + crate::table::optimize::CompactionOptions::default(), + None, + None, + None, + ) + .await + .unwrap(); + + // Compaction was submitted (and its placeholder stats are present)... + assert_eq!( + stats.compaction, + Some(lance::dataset::optimize::CompactionMetrics::default()) + ); + // ...but Prune and Index were skipped rather than attempted and failed, + // so the call succeeds with zeroed prune stats instead of returning an + // error after the job has already been submitted. + assert_eq!( + stats.prune, + Some(lance::dataset::cleanup::RemovalStats::default()) + ); + assert_eq!( + *statements.lock().unwrap(), + ["OPTIMIZE TABLE \"events\" REWRITE DATA"] + ); + shutdown.send(()).unwrap(); +} + +/// `delete_unverified: Some(false)` is what every real Python call sends +/// (the public `optimize()` default is `False`, not `None`), so it must be +/// treated the same as "not asking to prune" -- not rejected -- or a plain +/// `table.optimize()` from Python would never succeed on a remote table. +#[tokio::test] +async fn remote_optimize_full_treats_delete_unverified_false_as_default() { + let service = TestSqlService { + adaptive: true, + ..Default::default() + }; + let statements = service.statements.clone(); + let (sql_client, shutdown) = start_sql_service(service, ClientConfig::default()).await; + let client = client_with_handler(|request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()); + }); + let remote_table = RemoteTable::new_with_sql_client( + client, + "analytics".into(), + "events".into(), + vec!["team".into(), "reports".into()], + "team$reports$events".into(), + Default::default(), + Some(sql_client), + ); + let database = Arc::new(RemoteDatabase::new_mock( + |request: reqwest::Request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()); + }, + )); + let table = crate::Table::new(Arc::new(remote_table), database); + + let stats = table + .optimize_full( + crate::table::optimize::CompactionOptions::default(), + None, + None, + Some(false), + ) + .await + .unwrap(); + + assert_eq!( + stats.prune, + Some(lance::dataset::cleanup::RemovalStats::default()) + ); + assert_eq!(statements.lock().unwrap().len(), 1); + shutdown.send(()).unwrap(); +} + +/// Regression test: a caller who explicitly asks a remote table's +/// `optimize()` to prune (an older-than cutoff, an absolute timestamp, or +/// `delete_unverified=True`) used to get a success result reporting zero +/// bytes removed, even though no pruning was attempted at all -- silently +/// dropping their storage-reclaim request. It must instead reject before +/// anything is submitted, the same way a non-default `CompactionOptions` is. +#[tokio::test] +async fn remote_optimize_full_rejects_explicit_pruning_options() { + async fn assert_rejected_before_submission( + older_than: Option, + before_timestamp: Option>, + delete_unverified: Option, + ) { + let service = TestSqlService { + adaptive: true, + ..Default::default() + }; + let statements = service.statements.clone(); + let (sql_client, shutdown) = start_sql_service(service, ClientConfig::default()).await; + let client = client_with_handler(|request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()); + }); + let remote_table = RemoteTable::new_with_sql_client( + client, + "analytics".into(), + "events".into(), + vec!["team".into(), "reports".into()], + "team$reports$events".into(), + Default::default(), + Some(sql_client), + ); + let database = Arc::new(RemoteDatabase::new_mock( + |request: reqwest::Request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()); + }, + )); + let table = crate::Table::new(Arc::new(remote_table), database); + + let error = table + .optimize_full( + crate::table::optimize::CompactionOptions::default(), + older_than, + before_timestamp, + delete_unverified, + ) + .await + .unwrap_err(); + assert!(matches!(error, Error::NotSupported { .. }), "{error:?}"); + assert!(statements.lock().unwrap().is_empty()); + shutdown.send(()).unwrap(); + } + + assert_rejected_before_submission(Some(chrono::Duration::days(1)), None, None).await; + assert_rejected_before_submission(None, Some(chrono::Utc::now()), None).await; + assert_rejected_before_submission(None, None, Some(true)).await; +} + #[tokio::test] async fn remote_mv_listing_uses_scoped_sql_keyset_pagination() { for defer in [false, true] { diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 9e9d507b7..590f4866f 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -429,6 +429,10 @@ fn quote_sql_identifier(identifier: &str) -> String { format!("\"{}\"", identifier.replace('"', "\"\"")) } +fn quote_sql_string(value: &str) -> String { + format!("'{}'", value.replace('\'', "''")) +} + fn freshness_headers_snapshot( freshness: &Mutex, interval: Option, @@ -854,7 +858,7 @@ impl RemoteTable { /// Return a new handle scoped to `branch`, sharing the client but with fresh /// caches and version/freshness state (the branch tracks its own latest). /// Mirrors `NativeTable`'s handle-per-branch model. - fn with_branch(&self, branch: Option) -> Self { + pub(super) fn with_branch(&self, branch: Option) -> Self { Self { client: self.client.clone(), db_name: self.db_name.clone(), @@ -3661,10 +3665,84 @@ impl BaseTable for RemoteTable { Ok(()) } - async fn optimize(&self, _action: OptimizeAction) -> Result { + /// Only `OptimizeAction::Compact` with default `options` and no custom + /// `remap_options` is supported remotely: it submits a durable + /// compaction job through the server's `OPTIMIZE TABLE ... REWRITE DATA` + /// SQL statement, the same atomic, single-worker, index-safe path the + /// server uses for its own automatic compaction. `.wait()` below waits + /// for that SQL statement's own execution to finish, which only confirms + /// the compaction job was accepted into the server's job registry; the + /// compaction job itself then runs asynchronously on the server, so this + /// call can return well before the table is actually compacted. + /// `OptimizeStats::compaction` is `Some(CompactionMetrics::default())`, + /// a placeholder rather than `None` -- callers (including the + /// Python/Node bindings) unwrap it unconditionally on success -- because + /// the SQL job adapter does not expose per-job compaction metrics. + /// `OptimizeAction::All`, `::Prune`, and `::Index` remain unsupported + /// remotely: there is no SQL statement yet for version pruning or index + /// optimization. A branch-scoped table handle submits a + /// `WITH (branch = '')` clause; the statement does not itself + /// validate the branch exists, so a nonexistent branch surfaces as an + /// async job failure rather than an error from this call. A non-default + /// `CompactionOptions` (for example `target_rows_per_fragment` or + /// `max_source_bytes`) is rejected: the SQL statement has no `WITH` + /// clause to carry it yet, so accepting it would silently ignore + /// settings a caller may be relying on to bound the job. + async fn optimize(&self, action: OptimizeAction) -> Result { self.check_mutable().await?; - Err(Error::NotSupported { - message: "optimize is not supported in LanceDB Enterprise.".into(), + let OptimizeAction::Compact { + options, + remap_options, + } = &action + else { + return Err(Error::NotSupported { + message: "optimize is only supported in LanceDB Enterprise for \ + OptimizeAction::Compact; version pruning and index \ + optimization are not yet supported remotely." + .into(), + }); + }; + if remap_options.is_some() { + return Err(Error::NotSupported { + message: "a custom index remap strategy is not supported in LanceDB \ + Enterprise; omit remap_options to use the server's managed, \ + index-safe compaction path." + .into(), + }); + } + if *options != lance::dataset::optimize::CompactionOptions::default() { + return Err(Error::NotSupported { + message: "custom CompactionOptions are not supported in LanceDB \ + Enterprise; the OPTIMIZE TABLE ... REWRITE DATA SQL \ + statement has no WITH clause to carry them yet, so a \ + non-default value would be silently ignored rather than \ + applied. Omit options to use the server's managed \ + compaction settings." + .into(), + }); + } + let sql_client = self.sql_client.clone().ok_or_else(|| Error::NotSupported { + message: "SQL is unavailable for this remote table client".to_string(), + })?; + let mut statement = format!( + "OPTIMIZE TABLE {} REWRITE DATA", + quote_sql_identifier(&self.name) + ); + if let Some(branch) = &self.branch { + statement.push_str(&format!(" WITH (branch = {})", quote_sql_string(branch))); + } + // `OptimizeStats` carries `lance::dataset::optimize` metric types that + // are not `Serialize`/`Deserialize`/`Clone`, which `Job` requires, + // so the job itself only signals completion; the real (placeholder) + // value is built afterward. + sql_client + .submit_as_job(statement, self.namespace.clone(), || async { Ok(()) }) + .await? + .wait() + .await?; + Ok(OptimizeStats { + compaction: Some(lance::dataset::optimize::CompactionMetrics::default()), + prune: None, }) } async fn add_columns( @@ -6794,10 +6872,9 @@ mod tests { let mut fts = FullTextSearchQuery::new("test".into()); fts.wand_factor = Some(1.0); let results = [ - ( - "optimize", - table.optimize(OptimizeAction::All).await.map(|_| ()), - ), + // `optimize` has its own action-specific messages now and its + // own dedicated tests (`test_optimize_*`), not the generic + // boilerplate this loop checks. ( "Wand factor", table @@ -15535,6 +15612,115 @@ mod tests { assert!(error.to_string().contains("SQL is unavailable")); } + #[tokio::test] + async fn test_optimize_requires_sql() { + let table = Table::new_with_handler("my_table", |request| -> http::Response { + panic!("unexpected REST request: {}", request.url().path()) + }); + let error = table + .optimize(OptimizeAction::Compact { + options: crate::table::optimize::CompactionOptions::default(), + remap_options: None, + }) + .await + .unwrap_err(); + assert!(error.to_string().contains("SQL is unavailable")); + } + + #[tokio::test] + async fn test_optimize_rejects_custom_remap_options() { + let table = Table::new_with_handler("my_table", move |_| -> http::Response { + panic!("a custom remap strategy must be rejected before any request is sent") + }); + + let error = table + .optimize(OptimizeAction::Compact { + options: crate::table::optimize::CompactionOptions::default(), + remap_options: Some(Arc::new(lance::dataset::optimize::IgnoreRemap::default())), + }) + .await + .unwrap_err(); + + assert!(matches!(error, Error::NotSupported { .. })); + assert!(error.to_string().contains("remap")); + } + + #[tokio::test] + async fn test_optimize_rejects_non_default_compaction_options() { + let table = Table::new_with_handler("my_table", move |_| -> http::Response { + panic!("custom compaction options must be rejected before any request is sent") + }); + + let error = table + .optimize(OptimizeAction::Compact { + options: crate::table::optimize::CompactionOptions { + target_rows_per_fragment: 17, + max_source_bytes: Some(1024), + ..Default::default() + }, + remap_options: None, + }) + .await + .unwrap_err(); + + assert!(matches!(error, Error::NotSupported { .. })); + assert!(error.to_string().contains("CompactionOptions")); + } + + #[tokio::test] + async fn test_optimize_rejects_all_prune_and_index() { + for action in [ + OptimizeAction::All, + OptimizeAction::Prune { + older_than: None, + delete_unverified: None, + error_if_tagged_old_versions: None, + }, + OptimizeAction::Index(Default::default()), + ] { + let table = Table::new_with_handler("my_table", move |_| -> http::Response { + panic!( + "an unsupported remote optimize action must be rejected before any request is sent" + ) + }); + + let error = table.optimize(action).await.unwrap_err(); + assert!(matches!(error, Error::NotSupported { .. })); + } + } + + #[tokio::test] + async fn test_optimize_on_a_branch_requires_sql() { + use lance::dataset::refs::Ref; + let table = + Table::new_with_handler("my_table", move |request| match request.url().path() { + "/v1/table/my_table/branches/create/" => http::Response::builder() + .status(200) + .body(b"{}".to_vec()) + .unwrap(), + path => panic!("unexpected request on a branch-scoped optimize call: {path}"), + }); + + let branch = table + .create_branch("dev", Ref::Version(None, None)) + .await + .unwrap(); + + // No SQL client is wired to this REST-only mock, so the branch-scoped + // handle hits the same "SQL is unavailable" check `optimize` runs for + // main -- it does not reject the branch itself, and it does not send + // a REST request (see `test_optimize_requires_sql`). + let error = branch + .optimize(OptimizeAction::Compact { + options: crate::table::optimize::CompactionOptions::default(), + remap_options: None, + }) + .await + .unwrap_err(); + + assert!(error.to_string().contains("SQL is unavailable")); + } + #[tokio::test] async fn test_create_branch_empty_name_rejected_client_side() { use lance::dataset::refs::Ref; diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 8a7133b37..6bc845580 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -19,6 +19,7 @@ pub use lance::dataset::ReadParams; pub use lance::dataset::Version; use lance::dataset::WriteMode; use lance::dataset::builder::DatasetBuilder; +use lance::dataset::cleanup::RemovalStats; pub use lance::dataset::mem_wal::MemIndexRegistry; /// The traits a memtable index plugin implements; see [`MemIndexRegistry`]. pub use lance::dataset::mem_wal::index as mem_index; @@ -1900,6 +1901,13 @@ impl Table { /// optimize should be run frequently. A good rule of thumb is to run optimize if /// you have added or modified 100,000 or more records or run more than 20 data /// modification operations. + /// + /// # LanceDB Enterprise (remote tables) + /// + /// Only [`OptimizeAction::Compact`] with default `options` and no custom + /// `remap_options` is supported; `All`, `Prune`, and `Index` return + /// [`crate::Error::NotSupported`]. See [`OptimizeAction::Compact`] for + /// what a remote call submits and what it does and does not wait for. pub async fn optimize(&self, action: OptimizeAction) -> Result { self.inner.optimize(action).await } @@ -1931,6 +1939,77 @@ impl Table { }) } + /// Run the maintenance cycle used by language bindings' `optimize()` + /// convenience method: compact, then -- on a local table -- prune and + /// optimize indices. + /// + /// This is an internal entry point for language bindings. On a remote + /// table (LanceDB Enterprise) only [`OptimizeAction::Compact`] with its + /// defaults is supported. A caller asking for anything pruning-related + /// (`older_than`, `before_timestamp`, or `delete_unverified: Some(true)`) + /// gets `NotSupported` before anything is submitted, the same way a + /// non-default `CompactionOptions` is already rejected -- rather than a + /// success that silently reports zero bytes pruned when nothing was + /// pruned at all. A plain call with none of those set only compacts: + /// `Prune` and `Index` are skipped here rather than attempted and + /// failed, since by this point the compaction job has already been + /// durably submitted and a caller retrying on that failure would + /// resubmit it. The returned prune stats are zeroed in that case. See + /// [`OptimizeAction::Compact`] for what the submission does and does not + /// wait for. + #[doc(hidden)] + pub async fn optimize_full( + &self, + compaction_options: CompactionOptions, + older_than: Option, + before_timestamp: Option>, + delete_unverified: Option, + ) -> Result { + let is_remote = self.as_native().is_none(); + if is_remote + && (older_than.is_some() + || before_timestamp.is_some() + || delete_unverified == Some(true)) + { + return Err(Error::NotSupported { + message: "pruning (older_than, before_timestamp, or \ + delete_unverified) is not supported in LanceDB \ + Enterprise; omit those options to run compaction \ + only." + .into(), + }); + } + let compaction = self + .optimize(OptimizeAction::Compact { + options: compaction_options, + remap_options: None, + }) + .await? + .compaction; + if is_remote { + return Ok(OptimizeStats { + compaction, + prune: Some(RemovalStats::default()), + }); + } + let prune = if let Some(before_timestamp) = before_timestamp { + self.optimize_prune_before(before_timestamp, delete_unverified, None) + .await? + .prune + } else { + self.optimize(OptimizeAction::Prune { + older_than, + delete_unverified, + error_if_tagged_old_versions: None, + }) + .await? + .prune + }; + self.optimize(OptimizeAction::Index(OptimizeOptions::default())) + .await?; + Ok(OptimizeStats { compaction, prune }) + } + /// Add new columns to the table, providing values to fill in. pub fn add_columns(&self) -> AddColumnsBuilder { AddColumnsBuilder::new(self.inner.clone()) diff --git a/rust/lancedb/src/table/optimize.rs b/rust/lancedb/src/table/optimize.rs index 7c10bb96e..446234a6e 100644 --- a/rust/lancedb/src/table/optimize.rs +++ b/rust/lancedb/src/table/optimize.rs @@ -206,6 +206,23 @@ pub enum OptimizeAction { /// new files. If these operations are run frequently then compaction should run frequently. /// /// If these operations are never run (search only) then compaction is not necessary. + /// + /// # LanceDB Enterprise (remote tables) + /// + /// A remote table only supports `Compact` with default `options` and no + /// custom `remap_options`; both are rejected with + /// [`crate::Error::NotSupported`] otherwise. It submits a durable + /// compaction job to the server over SQL and waits for that submission to + /// be confirmed, not for compaction itself to finish: the job then runs + /// asynchronously on the server, so this call can return well before the + /// data is actually compacted. Because of that, the returned + /// [`OptimizeStats::compaction`] is a zeroed [`CompactionMetrics`] + /// placeholder rather than the real counts a local table reports. + /// + /// A branch-scoped table handle compacts that branch. The branch is not + /// validated to exist before the job is submitted, so a nonexistent + /// branch fails the compaction job asynchronously rather than failing + /// this call. Compact { options: CompactionOptions, remap_options: Option>,