feat: support remote table compaction via SQL OPTIMIZE TABLE statement (#4802)

`Table::optimize()` previously returned `NotSupported` for remote
tables. This PR wires it to the LanceDB Enterprise server's `OPTIMIZE
TABLE ... REWRITE DATA` SQL statement, submitted over Flight SQL.

Only `OptimizeAction::Compact` without a custom `remap_options` or
non-default `CompactionOptions` is supported remotely.
`OptimizeAction::All`, `::Prune`, and `::Index` all return a clear
`NotSupported` error instead of silently doing the wrong thing. A
branch-scoped table handle submits a `WITH (branch = '<branch>')`
clause; the statement doesn't validate the branch exists itself, so an
unknown branch surfaces as an async job failure rather than an error
from this call.

The server's SQL job adapter doesn't expose compaction metrics, so
`OptimizeStats::compaction` is `Some(CompactionMetrics::default())`, a
placeholder rather than real numbers — it must still be `Some`, since
callers (including the Python/Node bindings) unwrap it unconditionally
on success. This mirrors how `refresh_materialized_view_async` already
handles the same limitation for its own metrics.

The Python and Node `optimize()` convenience methods call a new shared
`Table::optimize_full`, which runs `Compact` and then, on a remote
table, skips `Prune` and `Index` instead of attempting and failing on
them — returning zeroed prune stats rather than submitting a compaction
job and then raising. Python's `RemoteTable.optimize()` also now
forwards `compaction_options` to the Rust layer, so a non-default value
is rejected before submission instead of being silently dropped.

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

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)>

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`

nodejs/__test__/remote.test.ts

@@ -312,7 +312,6 @@ describe("remote connection", () => {
         const cases: [() => Promise<unknown>, 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",

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<OptimizeOptions>): Promise<OptimizeStats>;
   /**

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,

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,
             )
         )
 

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:

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 == []

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,
     )
 
 

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,

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<String> {
+        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<String> {
+        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<String> {
+        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<String> {
+            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<String> {
+        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<String> {
+            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<chrono::Duration>,
+        before_timestamp: Option<chrono::DateTime<chrono::Utc>>,
+        delete_unverified: Option<bool>,
+    ) {
+        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<String> {
+            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<String> {
+                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] {

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<FreshnessState>,
     interval: Option<Duration>,
@@ -854,7 +858,7 @@ impl<S: HttpSend> RemoteTable<S> {
     /// 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<String>) -> Self {
+    pub(super) fn with_branch(&self, branch: Option<String>) -> Self {
         Self {
             client: self.client.clone(),
             db_name: self.db_name.clone(),
@@ -3661,10 +3665,84 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
 
         Ok(())
     }
-    async fn optimize(&self, _action: OptimizeAction) -> Result<OptimizeStats> {
+    /// 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 = '<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<OptimizeStats> {
         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<T>` 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<String> {
+            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<String> {
+            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<String> {
+            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<String> {
+                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;

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<OptimizeStats> {
         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<chrono::Duration>,
+        before_timestamp: Option<chrono::DateTime<chrono::Utc>>,
+        delete_unverified: Option<bool>,
+    ) -> Result<OptimizeStats> {
+        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())

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<Arc<dyn IndexRemapperOptions>>,