feat: bring the MemWAL LSM surface to parity across the SDKs

Four of the eight LSM methods are remote-only in the core: `impl BaseTable
for NativeTable` implements only set/unset/get_lsm_write_spec and
close_lsm_writers, while flush_lsm, compact_lsm and get_lsm_stats fall
through to trait defaults returning NotSupported. That is why Node had
bound the four that work locally and stopped, and why the remaining four
had no binding-level coverage anywhere.

Node: add napi bindings for flush_lsm, compact_lsm, checkpoint_lsm and
get_lsm_stats, with typed LsmStats/BucketStats/GenerationStats/
MemtableStats objects mirroring the existing LsmWriteSpec object in the
same file. Tests assert each binding reaches the core and surfaces
NotSupported locally; behavior against a real endpoint stays covered by
the mocked-endpoint tests in rust/lancedb/src/remote/table.rs.

Python: LsmWriteSpec was importable only from the private lancedb._lancedb
-- it appeared in table.py solely under `if TYPE_CHECKING:`. Export it as
lancedb.LsmWriteSpec, add it to __all__, and list it in the API reference,
which had no mention of it and so rendered it nowhere.

Java: add the LSM routes to lancedb-core. Java reaches LanceDB purely over
REST through the generated namespace client, and these routes are not in
the Lance Namespace spec, so they are issued through a small dedicated
client. LsmWriteSpec is deliberately not org.lance.memwal.
InitializeMemWalParams: that type defaults to maintaining no indexes where
a spec here defaults to maintaining every index, and it cannot express the
null that asks the server to resolve the set. checkpointLsm is ported from
rust/lancedb/src/table/checkpoint.rs with its constants and status
semantics intact -- 429/503 retried in place, 421 restarting from flush.

Note: `mvnw spotless:apply` cannot run on JDK 21 (google-java-format 1.7,
pinned in java/pom.xml, predates JDK 16's compiler API change). This is
pre-existing and reproduces on a pristine main checkout; the Java sources
here were formatted by hand to the checkstyle rules.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

java/README.md

@@ -29,6 +29,47 @@ LanceNamespace namespaceClient = LanceDbNamespaceClientBuilder.newBuilder()
     .build();
 ```
 
+## MemWAL LSM write path
+
+Most table operations reach LanceDB through the `LanceNamespace` above, which is
+generated from the Lance Namespace specification. The MemWAL LSM routes are not part
+of that specification, so they are issued through a separate client:
+
+```java
+import com.lancedb.LanceDbRestClient;
+import com.lancedb.LanceDbTableLsm;
+import com.lancedb.LsmWriteSpec;
+
+LanceDbRestClient client = LanceDbNamespaceClientBuilder.newBuilder()
+    .apiKey("your_lancedb_cloud_api_key")
+    .database("your_database_name")
+    .buildRestClient();
+
+LanceDbTableLsm lsm = new LanceDbTableLsm(client, "my_table");
+
+// Route future merge_insert upserts through the MemWAL, hash-bucketed by `id`.
+lsm.setLsmWriteSpec(LsmWriteSpec.bucket("id", 16));
+
+// ... merge_insert traffic ...
+
+// Converge the fresh tier into the base table.
+lsm.checkpointLsm();
+
+// Inspect live per-bucket state.
+lsm.getLsmStats().ifPresent(stats -> System.out.println(stats.get("buckets")));
+
+client.close();
+```
+
+`maintainedIndexes` is tri-state, and the null default is the opposite of what a Java
+reader usually expects:
+
+| Value | Meaning |
+| --- | --- |
+| unset (null) | Maintain **every** index the MemWAL can, resolved on install |
+| `Collections.emptyList()` | Maintain **none** |
+| `Arrays.asList("id_idx")` | Maintain exactly those |
+
 ## Development
 
 Build:

java/lancedb-core/pom.xml

@@ -33,6 +33,20 @@
             <artifactId>arrow-memory-netty</artifactId>
         </dependency>
 
+        <!-- Transport for the LanceDB routes outside the Lance Namespace spec.
+             Versions match what lance-namespace-apache-client resolves to. -->
+        <dependency>
+            <groupId>org.apache.httpcomponents.client5</groupId>
+            <artifactId>httpclient5</artifactId>
+            <version>5.2.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>2.17.1</version>
+        </dependency>
+
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>

java/lancedb-core/src/main/java/com/lancedb/LanceDbNamespaceClientBuilder.java

@@ -136,29 +136,48 @@ public class LanceDbNamespaceClientBuilder {
    * @throws IllegalStateException if required parameters are missing
    */
   public LanceNamespace build() {
-    // Validate required fields
+    validate();
+
+    // Build configuration map
+    Map<String, String> config = new HashMap<>(additionalConfig);
+    config.put("header.x-lancedb-database", database);
+    config.put("header.x-api-key", apiKey);
+    config.put("uri", resolveUri());
+
+    return LanceNamespace.connect("rest", config, null);
+  }
+
+  /**
+   * Build a {@link LanceDbRestClient} for the same endpoint.
+   *
+   * <p>Needed only for LanceDB routes that the Lance Namespace specification does not cover — the
+   * MemWAL LSM write path, reached through {@link LanceDbTableLsm}. Every other table operation
+   * belongs on the {@link LanceNamespace} from {@link #build()}.
+   *
+   * <p>The returned client owns an HTTP connection pool; close it when you are done with it.
+   *
+   * @return A configured LanceDbRestClient
+   * @throws IllegalStateException if required parameters are missing
+   */
+  public LanceDbRestClient buildRestClient() {
+    validate();
+    return new LanceDbRestClient(resolveUri(), apiKey, database);
+  }
+
+  private void validate() {
     if (apiKey == null) {
       throw new IllegalStateException("API key is required");
     }
     if (database == null) {
       throw new IllegalStateException("Database is required");
     }
+  }
 
-    // Build configuration map
-    Map<String, String> config = new HashMap<>(additionalConfig);
-    config.put("header.x-lancedb-database", database);
-    config.put("header.x-api-key", apiKey);
-
-    // Determine base URL
-    String uri;
+  /** The custom endpoint when set, else the LanceDB Cloud URL for this database and region. */
+  private String resolveUri() {
     if (endpoint.isPresent()) {
-      uri = endpoint.get();
-    } else {
-      String effectiveRegion = region.orElse(DEFAULT_REGION);
-      uri = String.format(CLOUD_URL_PATTERN, database, effectiveRegion);
+      return endpoint.get();
     }
-    config.put("uri", uri);
-
-    return LanceNamespace.connect("rest", config, null);
+    return String.format(CLOUD_URL_PATTERN, database, region.orElse(DEFAULT_REGION));
   }
 }

java/lancedb-core/src/main/java/com/lancedb/LanceDbRestClient.java

@@ -0,0 +1,114 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.lancedb;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.hc.client5.http.classic.methods.HttpPost;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.hc.core5.http.io.entity.StringEntity;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+
+/**
+ * Minimal HTTP client for LanceDB Cloud and Enterprise routes that the Lance Namespace
+ * specification does not cover.
+ *
+ * <p>Most table operations reach LanceDB through {@link org.lance.namespace.LanceNamespace}, which
+ * is generated from the namespace spec. A handful of routes — the MemWAL LSM write path in
+ * particular — are served by the same endpoint but are not part of that spec, so they are issued
+ * directly here. See {@link LanceDbTableLsm}.
+ *
+ * <p>Obtain one from {@link LanceDbNamespaceClientBuilder#buildRestClient()}.
+ */
+public class LanceDbRestClient implements Closeable {
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  private final String baseUri;
+  private final String apiKey;
+  private final String database;
+  private final CloseableHttpClient http;
+
+  LanceDbRestClient(String baseUri, String apiKey, String database) {
+    this.baseUri = baseUri.endsWith("/") ? baseUri.substring(0, baseUri.length() - 1) : baseUri;
+    this.apiKey = apiKey;
+    this.database = database;
+    this.http = HttpClients.createDefault();
+  }
+
+  /**
+   * POST {@code path}, sending {@code body} as JSON when it is non-null.
+   *
+   * @param path Absolute request path, beginning with {@code /}.
+   * @param body Object to serialize as the request body, or null to send no body.
+   * @return The parsed response body, or null when the response carried no content.
+   * @throws HttpException if the server returned a non-2xx status.
+   */
+  public JsonNode post(String path, Object body) {
+    HttpPost request = new HttpPost(baseUri + path);
+    request.setHeader("x-api-key", apiKey);
+    request.setHeader("x-lancedb-database", database);
+    try {
+      if (body != null) {
+        request.setEntity(
+            new StringEntity(MAPPER.writeValueAsString(body), ContentType.APPLICATION_JSON));
+      }
+      return http.execute(
+          request,
+          response -> {
+            String text =
+                response.getEntity() == null ? "" : EntityUtils.toString(response.getEntity());
+            int status = response.getCode();
+            if (status < 200 || status >= 300) {
+              throw new HttpException(status, "LanceDB request to " + path + " failed: " + text);
+            }
+            return text.isEmpty() ? null : MAPPER.readTree(text);
+          });
+    } catch (IOException e) {
+      throw new UncheckedIOException("LanceDB request to " + path + " failed", e);
+    }
+  }
+
+  @Override
+  public void close() throws IOException {
+    http.close();
+  }
+
+  /**
+   * A non-2xx response.
+   *
+   * <p>The status is exposed because callers act on it: {@link LanceDbTableLsm#checkpointLsm()}
+   * treats 429 and 503 as retryable and 421 as a lost node claim.
+   */
+  public static class HttpException extends RuntimeException {
+    private static final long serialVersionUID = 1L;
+
+    private final int statusCode;
+
+    public HttpException(int statusCode, String message) {
+      super(message);
+      this.statusCode = statusCode;
+    }
+
+    /** The HTTP status the failed response carried. */
+    public int statusCode() {
+      return statusCode;
+    }
+  }
+}

java/lancedb-core/src/main/java/com/lancedb/LanceDbTableLsm.java

@@ -0,0 +1,405 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.lancedb;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * The MemWAL LSM write path for one LanceDB Cloud or Enterprise table.
+ *
+ * <p>Installing an {@link LsmWriteSpec} routes {@code mergeInsert} upserts through Lance's MemWAL —
+ * an LSM-style append — instead of the standard merge path. Rows land in an in-memory memtable,
+ * seal into L0 generations, and are merged into the base table by compaction.
+ *
+ * <p>These routes are not part of the Lance Namespace specification, so they are issued directly
+ * rather than through {@link org.lance.namespace.LanceNamespace}.
+ *
+ * <pre>{@code
+ * LanceDbRestClient client = LanceDbNamespaceClientBuilder.newBuilder()
+ *     .apiKey("your_lancedb_cloud_api_key")
+ *     .database("your_database_name")
+ *     .buildRestClient();
+ *
+ * LanceDbTableLsm lsm = new LanceDbTableLsm(client, "my_table");
+ * lsm.setLsmWriteSpec(LsmWriteSpec.bucket("id", 16));
+ * // ... merge_insert traffic ...
+ * lsm.checkpointLsm();
+ * }</pre>
+ */
+public class LanceDbTableLsm {
+
+  /**
+   * Interval between {@code get_lsm_stats} polls during a checkpoint. One interval is roughly one
+   * compaction pass, the granularity at which the answer can change.
+   */
+  private static final long POLL_INTERVAL_MS = 5_000L;
+
+  /**
+   * Cap on re-issues from {@code flushLsm} after a 421, so a crash-looping node cannot turn flush →
+   * compact → 421 → flush into a spin.
+   *
+   * <p>Deliberately not shared with {@link #MAX_RETRIES}: a claim that keeps evaporating is a
+   * broken node, while contention is routine and wants a real budget.
+   */
+  private static final int MAX_REISSUES = 3;
+
+  /**
+   * Retryable faults tolerated on a <em>single</em> request, reset on every success — scattered
+   * contention across a long checkpoint must not accumulate toward a cap.
+   */
+  private static final int MAX_RETRIES = 8;
+
+  private static final long RETRY_BACKOFF_BASE_MS = 100L;
+  private static final long RETRY_BACKOFF_MAX_MS = 5_000L;
+
+  private final LanceDbRestClient client;
+  private final String tableIdentifier;
+
+  /**
+   * Bind the LSM routes for one table.
+   *
+   * @param client Transport for the LanceDB endpoint.
+   * @param tableIdentifier The table's full identifier, {@code $}-delimited when it sits inside a
+   *     namespace, such as {@code analytics$events}.
+   */
+  public LanceDbTableLsm(LanceDbRestClient client, String tableIdentifier) {
+    if (client == null) {
+      throw new IllegalArgumentException("Client cannot be null");
+    }
+    if (tableIdentifier == null || tableIdentifier.trim().isEmpty()) {
+      throw new IllegalArgumentException("Table identifier cannot be null or empty");
+    }
+    this.client = client;
+    this.tableIdentifier = tableIdentifier;
+  }
+
+  /**
+   * Install an {@link LsmWriteSpec} on this table, selecting the MemWAL LSM write path for future
+   * {@code mergeInsert} calls.
+   *
+   * <p>All variants require the table to have an unenforced primary key; bucket sharding
+   * additionally requires it to be the single column being bucketed.
+   */
+  public void setLsmWriteSpec(LsmWriteSpec spec) {
+    if (spec == null) {
+      throw new IllegalArgumentException("Spec cannot be null");
+    }
+    client.post(route("set_lsm_write_spec"), spec.toRequestBody());
+  }
+
+  /**
+   * Remove the {@link LsmWriteSpec} from this table, reverting to the standard {@code mergeInsert}
+   * write path.
+   *
+   * <p>Errors if no spec is currently set.
+   */
+  public void unsetLsmWriteSpec() {
+    client.post(route("unset_lsm_write_spec"), null);
+  }
+
+  /**
+   * Read the {@link LsmWriteSpec} currently installed on this table.
+   *
+   * <p>Empty when the LSM write path is not enabled. The returned spec mirrors what was installed,
+   * except that {@link LsmWriteSpec#maintainedIndexes()} always reports the concrete list resolved
+   * when the spec was set — a null selection never round-trips.
+   */
+  public Optional<LsmWriteSpec> getLsmWriteSpec() {
+    JsonNode response = client.post(route("get_lsm_write_spec"), null);
+    if (response == null || !response.hasNonNull("lsm_write_spec")) {
+      return Optional.empty();
+    }
+    return Optional.of(LsmWriteSpec.fromJson(response.get("lsm_write_spec")));
+  }
+
+  /**
+   * Seal every bucket's active memtable into a new L0 generation.
+   *
+   * <p>Returns once the seal is committed. Sealing an empty memtable is a no-op, so this is safe to
+   * call repeatedly.
+   */
+  public void flushLsm() {
+    client.post(route("flush_lsm"), null);
+  }
+
+  /**
+   * Trigger a background L0 → base compaction pass per bucket.
+   *
+   * <p>Returns once the passes are <em>dispatched</em>, not once they finish — watch {@link
+   * #getLsmStats}, or use {@link #checkpointLsm} to wait for convergence.
+   */
+  public void compactLsm() {
+    client.post(route("compact_lsm"), null);
+  }
+
+  /**
+   * Read live per-bucket LSM state.
+   *
+   * <p>Answers "how far behind is my fresh tier", "which bucket is hot", and "why is my fresh-tier
+   * vector search brute-force". Mutates no table state.
+   *
+   * <p>Empty only when the LSM write path is not enabled. The returned node is the server's {@code
+   * lsm_stats} object, carrying a {@code buckets} array.
+   *
+   * @param includeGenerationRows Also count rows per L0 generation. Off by default because each
+   *     count opens an uncached Lance dataset.
+   */
+  public Optional<JsonNode> getLsmStats(boolean includeGenerationRows) {
+    Map<String, Object> body = new LinkedHashMap<String, Object>();
+    body.put("include_generation_rows", includeGenerationRows);
+    JsonNode response = client.post(route("get_lsm_stats"), body);
+    if (response == null || !response.hasNonNull("lsm_stats")) {
+      return Optional.empty();
+    }
+    return Optional.of(response.get("lsm_stats"));
+  }
+
+  /** Equivalent to {@code getLsmStats(false)}. */
+  public Optional<JsonNode> getLsmStats() {
+    return getLsmStats(false);
+  }
+
+  /**
+   * Converge this table's LSM write path into its base table.
+   *
+   * <p>Seals once, fixes a target watermark from the resulting L0, then triggers compaction and
+   * polls until that L0 is gone. The target set is fixed at the start, so generations created
+   * <em>during</em> the checkpoint are ignored — that is what lets it terminate under write load,
+   * and what makes it best-effort: it converges the fresh tier as of some instant. Idempotent,
+   * abandonable at any point, safe on a cadence.
+   *
+   * <p>The loop runs here, not on the server: {@link #compactLsm} dispatches a pass and returns, so
+   * nothing holds a socket and a client can vanish mid-operation with nothing to reconcile.
+   * Completion is read from generation numbers in the shard manifest — durable state, unlike a
+   * count in a compact response, which a concurrent write invalidates.
+   *
+   * <p>No liveness bound — the caller owns the deadline. The compactor pool is shared across
+   * tables, so a checkpoint queued behind unrelated work looks exactly like one that is merging.
+   */
+  public void checkpointLsm() {
+    for (int reissue = 0; reissue <= MAX_REISSUES; reissue++) {
+      // The seal turns everything written before this call into a generation, so the
+      // watermark has to be read after it. Idempotent: sealing an empty memtable is a
+      // no-op, so a re-issue does not churn empty generations.
+      if (issueVoid(this::flushLsm)) {
+        backoff(reissue);
+        continue;
+      }
+
+      Attempt<Optional<JsonNode>> stats = issue(() -> getLsmStats(false));
+      if (stats.lostClaim) {
+        backoff(reissue);
+        continue;
+      }
+      if (!stats.value.isPresent()) {
+        // Not WAL-backed; flushLsm would have errored first but for a race.
+        return;
+      }
+
+      Map<String, Long> targets = newestGenerations(stats.value.get());
+      if (targets.isEmpty()) {
+        return;
+      }
+
+      if (drainToTargets(targets)) {
+        return;
+      }
+      backoff(reissue);
+    }
+    throw new IllegalStateException(
+        "checkpointLsm: the owning node kept losing its claim; re-issued from flush the maximum "
+            + "number of times");
+  }
+
+  /**
+   * Trigger and poll until no bucket holds a generation at or below its target.
+   *
+   * @return true when the drain finished, false when the table needs re-claiming from flush.
+   */
+  private boolean drainToTargets(Map<String, Long> targets) {
+    while (true) {
+      Attempt<Optional<JsonNode>> stats = issue(() -> getLsmStats(false));
+      if (stats.lostClaim) {
+        return false;
+      }
+      if (!stats.value.isPresent()) {
+        return true;
+      }
+
+      // `compacting` is the bucket's compaction latch, held from dispatch until the pass
+      // ends — including while it waits on a pod-wide permit. So it answers one question
+      // only: do not pile on. Buckets with nothing outstanding are skipped, not counted
+      // as idle.
+      long outstanding = 0;
+      boolean allCompacting = true;
+      for (JsonNode bucket : stats.value.get().path("buckets")) {
+        Long target = targets.get(bucket.path("shard_id").asText());
+        if (target == null) {
+          continue;
+        }
+        long remaining = outstandingGenerations(bucket, target);
+        if (remaining > 0) {
+          outstanding += remaining;
+          allCompacting &= bucket.path("compacting").asBoolean(false);
+        }
+      }
+      if (outstanding == 0) {
+        return true;
+      }
+
+      if (!allCompacting) {
+        try {
+          compactLsm();
+        } catch (LanceDbRestClient.HttpException e) {
+          if (isLostClaim(e)) {
+            return false;
+          }
+          if (!isRetryable(e)) {
+            throw e;
+          }
+          // A 429 here means the server could latch no bucket at all, which the poll
+          // above already handles. Not retried in place: the latch it would contend for
+          // is the one doing the work, so fall through and re-read — POLL_INTERVAL_MS is
+          // the backoff.
+        }
+      }
+      sleep(POLL_INTERVAL_MS);
+    }
+  }
+
+  /** The newest generation held by each bucket, skipping buckets holding none. */
+  private static Map<String, Long> newestGenerations(JsonNode stats) {
+    Map<String, Long> targets = new HashMap<String, Long>();
+    for (JsonNode bucket : stats.path("buckets")) {
+      long newest = Long.MIN_VALUE;
+      for (JsonNode generation : bucket.path("generations")) {
+        newest = Math.max(newest, generation.path("generation").asLong());
+      }
+      if (newest != Long.MIN_VALUE) {
+        targets.put(bucket.path("shard_id").asText(), newest);
+      }
+    }
+    return targets;
+  }
+
+  /**
+   * How many generations at or below {@code target} are still in L0.
+   *
+   * <p>A count, not a boolean: one pass drains a bounded prefix rather than the whole target set,
+   * so a boolean would read as "no progress" for every pass but the last.
+   */
+  private static long outstandingGenerations(JsonNode bucket, long target) {
+    long count = 0;
+    for (JsonNode generation : bucket.path("generations")) {
+      if (generation.path("generation").asLong() <= target) {
+        count++;
+      }
+    }
+    return count;
+  }
+
+  /**
+   * 429 (latch held, pool saturated, or the pod replaying its WAL) and 503 (a draining node, or a
+   * proxy between here and it).
+   */
+  private static boolean isRetryable(LanceDbRestClient.HttpException e) {
+    return e.statusCode() == 429 || e.statusCode() == 503;
+  }
+
+  /**
+   * 421: the owning node holds no claim. Only {@code flush} re-claims and replays, so this cannot
+   * be retried in place — the caller has to start over.
+   */
+  private static boolean isLostClaim(LanceDbRestClient.HttpException e) {
+    return e.statusCode() == 421;
+  }
+
+  /**
+   * Issue one LSM request, retrying in place while the fault is retryable.
+   *
+   * <p>The two recoverable faults have separate budgets: contention clears on its own and retries
+   * here against {@link #MAX_RETRIES}, while a 421 needs {@code flush} to re-claim, which only the
+   * caller can drive.
+   *
+   * <p>An exhausted budget propagates the last error as itself rather than a synthesized one — "429
+   * after nine tries" beats "checkpoint failed".
+   */
+  private static <T> Attempt<T> issue(Call<T> call) {
+    int retries = 0;
+    while (true) {
+      try {
+        return new Attempt<T>(call.run(), false);
+      } catch (LanceDbRestClient.HttpException e) {
+        if (isLostClaim(e)) {
+          return new Attempt<T>(null, true);
+        }
+        if (!isRetryable(e) || retries >= MAX_RETRIES) {
+          throw e;
+        }
+        backoff(retries);
+        retries++;
+      }
+    }
+  }
+
+  /** {@link #issue} for a call with no return value. Returns true when the claim was lost. */
+  private static boolean issueVoid(Runnable call) {
+    return issue(
+            () -> {
+              call.run();
+              return Boolean.TRUE;
+            })
+        .lostClaim;
+  }
+
+  /** Sleep before re-issuing a retryable request. Doubles up to {@link #RETRY_BACKOFF_MAX_MS}. */
+  private static void backoff(int attempt) {
+    long delay = RETRY_BACKOFF_BASE_MS << Math.min(attempt, 8);
+    sleep(Math.min(delay, RETRY_BACKOFF_MAX_MS));
+  }
+
+  private static void sleep(long millis) {
+    try {
+      Thread.sleep(millis);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new IllegalStateException("Interrupted while waiting on the LSM checkpoint", e);
+    }
+  }
+
+  private String route(String operation) {
+    return "/v1/table/" + tableIdentifier + "/" + operation + "/";
+  }
+
+  /** What one LSM request produced: its value, or word that the owning node holds no claim. */
+  private static final class Attempt<T> {
+    private final T value;
+    private final boolean lostClaim;
+
+    private Attempt(T value, boolean lostClaim) {
+      this.value = value;
+      this.lostClaim = lostClaim;
+    }
+  }
+
+  @FunctionalInterface
+  private interface Call<T> {
+    T run();
+  }
+}

java/lancedb-core/src/main/java/com/lancedb/LsmWriteSpec.java

@@ -0,0 +1,260 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.lancedb;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Specification selecting Lance's MemWAL LSM-style write path for {@code mergeInsert}.
+ *
+ * <p>Construct via {@link #bucket}, {@link #identity}, or {@link #unsharded}, then optionally chain
+ * {@link #withMaintainedIndexes} and {@link #withWriterConfigDefaults}. Install it with {@link
+ * LanceDbTableLsm#setLsmWriteSpec} and remove it with {@link LanceDbTableLsm#unsetLsmWriteSpec}.
+ *
+ * <p>This is deliberately not {@code org.lance.memwal.InitializeMemWalParams}. That type is Lance's
+ * own, and its maintained-index default is the opposite of this one: it defaults to maintaining
+ * <em>nothing</em>, while a fresh spec here maintains <em>every</em> index. It also cannot express
+ * the null that asks the server to resolve the set.
+ */
+public class LsmWriteSpec {
+
+  /** How writes are routed to MemWAL shards. */
+  public enum Sharding {
+    /** Hash-bucket writes by a scalar column. */
+    BUCKET("bucket"),
+    /** Shard by the raw value of a scalar column. */
+    IDENTITY("identity"),
+    /** Route every write to a single shard. */
+    UNSHARDED("unsharded");
+
+    private final String wireName;
+
+    Sharding(String wireName) {
+      this.wireName = wireName;
+    }
+
+    String wireName() {
+      return wireName;
+    }
+
+    static Sharding fromWireName(String name) {
+      for (Sharding s : values()) {
+        if (s.wireName.equals(name)) {
+          return s;
+        }
+      }
+      throw new IllegalArgumentException("Unknown sharding mode: " + name);
+    }
+  }
+
+  private final Sharding sharding;
+  private final String column;
+  private final Integer numBuckets;
+  private final List<String> maintainedIndexes;
+  private final Map<String, String> writerConfigDefaults;
+
+  private LsmWriteSpec(
+      Sharding sharding,
+      String column,
+      Integer numBuckets,
+      List<String> maintainedIndexes,
+      Map<String, String> writerConfigDefaults) {
+    this.sharding = sharding;
+    this.column = column;
+    this.numBuckets = numBuckets;
+    this.maintainedIndexes = maintainedIndexes;
+    this.writerConfigDefaults = writerConfigDefaults;
+  }
+
+  /**
+   * Hash-bucket sharding by a scalar column, maintaining every index on the table.
+   *
+   * <p>Iceberg-compatible Murmur3-x86-32 (seed 0) is used, so each row's {@code bucket(column,
+   * numBuckets)} value is stable across processes.
+   *
+   * @param column A non-nested column with a supported scalar type.
+   * @param numBuckets The number of buckets, in {@code [1, 1024]}.
+   */
+  public static LsmWriteSpec bucket(String column, int numBuckets) {
+    if (column == null || column.trim().isEmpty()) {
+      throw new IllegalArgumentException("Column cannot be null or empty");
+    }
+    return new LsmWriteSpec(
+        Sharding.BUCKET, column, numBuckets, null, new HashMap<String, String>());
+  }
+
+  /**
+   * Identity sharding — shard by the raw value of {@code column} — maintaining every index on the
+   * table.
+   *
+   * <p>{@code column} must be a deterministic function of the unenforced primary key: every row
+   * with a given primary key must always produce the same {@code column} value, or upserts of that
+   * key can land in different shards and a stale version can win.
+   */
+  public static LsmWriteSpec identity(String column) {
+    if (column == null || column.trim().isEmpty()) {
+      throw new IllegalArgumentException("Column cannot be null or empty");
+    }
+    return new LsmWriteSpec(Sharding.IDENTITY, column, null, null, new HashMap<String, String>());
+  }
+
+  /** No sharding — every write goes to a single MemWAL shard — maintaining every index. */
+  public static LsmWriteSpec unsharded() {
+    return new LsmWriteSpec(Sharding.UNSHARDED, null, null, null, new HashMap<String, String>());
+  }
+
+  /**
+   * Set the indexes the MemWAL keeps up to date as rows are appended.
+   *
+   * <p>Pass {@code null} — the default for a fresh spec — to maintain every index the MemWAL can,
+   * resolved when the spec is installed. That is a snapshot: indexes created later are not
+   * maintained until the spec is unset and set again. Pass an empty list to maintain none.
+   *
+   * <p>Note that {@code null} and the empty list mean opposite things here.
+   */
+  public LsmWriteSpec withMaintainedIndexes(List<String> maintainedIndexes) {
+    return new LsmWriteSpec(
+        sharding,
+        column,
+        numBuckets,
+        maintainedIndexes == null ? null : new ArrayList<String>(maintainedIndexes),
+        writerConfigDefaults);
+  }
+
+  /**
+   * Set default {@code ShardWriter} configuration recorded in the MemWAL index.
+   *
+   * <p>A sparse override map — only the keys you set are recorded. Recognized keys include {@code
+   * durable_write}, {@code max_wal_buffer_size}, {@code max_memtable_size}, {@code
+   * max_memtable_rows}, {@code max_memtable_batches}, {@code manifest_scan_batch_size}, {@code
+   * max_unflushed_memtable_bytes}, and {@code enable_memtable}. Duration knobs carry an {@code _ms}
+   * suffix, such as {@code max_wal_flush_interval_ms}.
+   */
+  public LsmWriteSpec withWriterConfigDefaults(Map<String, String> writerConfigDefaults) {
+    if (writerConfigDefaults == null) {
+      throw new IllegalArgumentException("writerConfigDefaults cannot be null");
+    }
+    return new LsmWriteSpec(
+        sharding,
+        column,
+        numBuckets,
+        maintainedIndexes,
+        new HashMap<String, String>(writerConfigDefaults));
+  }
+
+  /** How writes are routed to shards. */
+  public Sharding sharding() {
+    return sharding;
+  }
+
+  /** The sharding column for {@link Sharding#BUCKET} and {@link Sharding#IDENTITY}, else null. */
+  public String column() {
+    return column;
+  }
+
+  /** The bucket count for {@link Sharding#BUCKET}, else null. */
+  public Integer numBuckets() {
+    return numBuckets;
+  }
+
+  /**
+   * The indexes the MemWAL maintains, or null to have the server resolve every maintainable index
+   * on install. An empty list means none.
+   */
+  public List<String> maintainedIndexes() {
+    return maintainedIndexes == null ? null : Collections.unmodifiableList(maintainedIndexes);
+  }
+
+  /** Default {@code ShardWriter} configuration recorded in the MemWAL index. */
+  public Map<String, String> writerConfigDefaults() {
+    return Collections.unmodifiableMap(writerConfigDefaults);
+  }
+
+  /** Render this spec as the {@code set_lsm_write_spec} request body. */
+  Map<String, Object> toRequestBody() {
+    Map<String, Object> shardingBody = new LinkedHashMap<String, Object>();
+    shardingBody.put("mode", sharding.wireName());
+    if (column != null) {
+      shardingBody.put("column", column);
+    }
+    if (numBuckets != null) {
+      shardingBody.put("num_buckets", numBuckets);
+    }
+
+    Map<String, Object> body = new LinkedHashMap<String, Object>();
+    body.put("sharding", shardingBody);
+    // Null is meaningful: it asks the server to resolve every maintainable index.
+    body.put("maintained_indexes", maintainedIndexes);
+    body.put("writer_config_defaults", writerConfigDefaults);
+    return body;
+  }
+
+  /**
+   * Rebuild a spec from a {@code get_lsm_write_spec} response body.
+   *
+   * <p>The server always reports a concrete maintained-index list, so a null selection never
+   * round-trips.
+   */
+  static LsmWriteSpec fromJson(JsonNode node) {
+    JsonNode shardingNode = node.get("sharding");
+    if (shardingNode == null || shardingNode.get("mode") == null) {
+      throw new IllegalStateException("get_lsm_write_spec response has no sharding mode");
+    }
+    Sharding sharding = Sharding.fromWireName(shardingNode.get("mode").asText());
+
+    String column = shardingNode.hasNonNull("column") ? shardingNode.get("column").asText() : null;
+    Integer numBuckets =
+        shardingNode.hasNonNull("num_buckets") ? shardingNode.get("num_buckets").asInt() : null;
+
+    List<String> maintainedIndexes = new ArrayList<String>();
+    JsonNode indexesNode = node.get("maintained_indexes");
+    if (indexesNode != null && indexesNode.isArray()) {
+      for (JsonNode index : indexesNode) {
+        maintainedIndexes.add(index.asText());
+      }
+    }
+
+    Map<String, String> defaults = new HashMap<String, String>();
+    JsonNode defaultsNode = node.get("writer_config_defaults");
+    if (defaultsNode != null && defaultsNode.isObject()) {
+      defaultsNode
+          .fieldNames()
+          .forEachRemaining(name -> defaults.put(name, defaultsNode.get(name).asText()));
+    }
+
+    return new LsmWriteSpec(sharding, column, numBuckets, maintainedIndexes, defaults);
+  }
+
+  @Override
+  public String toString() {
+    return "LsmWriteSpec{sharding="
+        + sharding
+        + ", column="
+        + column
+        + ", numBuckets="
+        + numBuckets
+        + ", maintainedIndexes="
+        + maintainedIndexes
+        + ", writerConfigDefaults="
+        + writerConfigDefaults
+        + "}";
+  }
+}

java/lancedb-core/src/test/java/com/lancedb/LanceDbTableLsmTest.java

@@ -0,0 +1,434 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.lancedb;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Unit tests for the MemWAL LSM routes, run against a scripted local HTTP server.
+ *
+ * <p>The wire assertions mirror the Rust mocked-endpoint tests in {@code
+ * rust/lancedb/src/remote/table.rs}, which are the contract these routes have to match.
+ */
+public class LanceDbTableLsmTest {
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  private HttpServer server;
+  private LanceDbRestClient client;
+  private LanceDbTableLsm lsm;
+
+  private final List<String> requestPaths = Collections.synchronizedList(new ArrayList<String>());
+  private final List<String> requestBodies = Collections.synchronizedList(new ArrayList<String>());
+  private final Map<String, Deque<Reply>> replies = new ConcurrentHashMap<String, Deque<Reply>>();
+
+  @BeforeEach
+  public void setUp() throws IOException {
+    server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+    server.createContext(
+        "/",
+        exchange -> {
+          String path = exchange.getRequestURI().getPath();
+          requestPaths.add(path);
+          requestBodies.add(readAll(exchange.getRequestBody()));
+
+          Reply reply = nextReply(path);
+          byte[] out = reply.body.getBytes(StandardCharsets.UTF_8);
+          exchange.sendResponseHeaders(reply.status, out.length == 0 ? -1 : out.length);
+          if (out.length > 0) {
+            exchange.getResponseBody().write(out);
+          }
+          exchange.close();
+        });
+    server.start();
+
+    client =
+        LanceDbNamespaceClientBuilder.newBuilder()
+            .apiKey("test-key")
+            .database("test-db")
+            .endpoint("http://127.0.0.1:" + server.getAddress().getPort())
+            .buildRestClient();
+    lsm = new LanceDbTableLsm(client, "my_table");
+  }
+
+  @AfterEach
+  public void tearDown() throws IOException {
+    client.close();
+    server.stop(0);
+  }
+
+  // ===========================================================================
+  // set / unset / get spec
+  // ===========================================================================
+
+  @Test
+  public void testSetLsmWriteSpecUnsharded() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    lsm.setLsmWriteSpec(LsmWriteSpec.unsharded());
+
+    assertEquals("/v1/table/my_table/set_lsm_write_spec/", requestPaths.get(0));
+    JsonNode body = MAPPER.readTree(requestBodies.get(0));
+    assertEquals("unsharded", body.get("sharding").get("mode").asText());
+    assertFalse(body.get("sharding").has("column"));
+    assertFalse(body.get("sharding").has("num_buckets"));
+  }
+
+  @Test
+  public void testSetLsmWriteSpecBucket() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    lsm.setLsmWriteSpec(
+        LsmWriteSpec.bucket("id", 16).withMaintainedIndexes(Arrays.asList("id_idx")));
+
+    JsonNode body = MAPPER.readTree(requestBodies.get(0));
+    assertEquals("bucket", body.get("sharding").get("mode").asText());
+    assertEquals("id", body.get("sharding").get("column").asText());
+    assertEquals(16, body.get("sharding").get("num_buckets").asInt());
+    assertEquals(1, body.get("maintained_indexes").size());
+    assertEquals("id_idx", body.get("maintained_indexes").get(0).asText());
+  }
+
+  @Test
+  public void testSetLsmWriteSpecIdentity() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    lsm.setLsmWriteSpec(LsmWriteSpec.identity("tenant"));
+
+    JsonNode body = MAPPER.readTree(requestBodies.get(0));
+    assertEquals("identity", body.get("sharding").get("mode").asText());
+    assertEquals("tenant", body.get("sharding").get("column").asText());
+    assertFalse(body.get("sharding").has("num_buckets"));
+  }
+
+  /**
+   * The tri-state that motivated a LanceDB-owned spec type: a null selection asks the server to
+   * resolve every maintainable index, while an empty list asks for none. They must not collapse.
+   */
+  @Test
+  public void testMaintainedIndexesNullAndEmptyAreDistinctOnTheWire() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    lsm.setLsmWriteSpec(LsmWriteSpec.unsharded());
+    JsonNode fresh = MAPPER.readTree(requestBodies.get(0));
+    assertTrue(fresh.has("maintained_indexes"), "the key must be present");
+    assertTrue(fresh.get("maintained_indexes").isNull(), "a fresh spec sends null, not []");
+
+    lsm.setLsmWriteSpec(
+        LsmWriteSpec.unsharded().withMaintainedIndexes(Collections.<String>emptyList()));
+    JsonNode none = MAPPER.readTree(requestBodies.get(1));
+    assertTrue(none.get("maintained_indexes").isArray());
+    assertEquals(0, none.get("maintained_indexes").size());
+  }
+
+  @Test
+  public void testSetLsmWriteSpecWriterConfigDefaults() throws Exception {
+    enqueue("set_lsm_write_spec", 200, "");
+
+    Map<String, String> defaults = new HashMap<String, String>();
+    defaults.put("max_memtable_rows", "50000");
+    lsm.setLsmWriteSpec(LsmWriteSpec.unsharded().withWriterConfigDefaults(defaults));
+
+    JsonNode body = MAPPER.readTree(requestBodies.get(0));
+    assertEquals("50000", body.get("writer_config_defaults").get("max_memtable_rows").asText());
+  }
+
+  @Test
+  public void testUnsetLsmWriteSpec() {
+    enqueue("unset_lsm_write_spec", 200, "");
+
+    lsm.unsetLsmWriteSpec();
+
+    assertEquals("/v1/table/my_table/unset_lsm_write_spec/", requestPaths.get(0));
+    assertEquals("", requestBodies.get(0));
+  }
+
+  @Test
+  public void testGetLsmWriteSpec() {
+    enqueue(
+        "get_lsm_write_spec",
+        200,
+        "{\"lsm_write_spec\":{\"sharding\":{\"mode\":\"bucket\",\"column\":\"id\","
+            + "\"num_buckets\":16},\"maintained_indexes\":[\"id_idx\"],"
+            + "\"writer_config_defaults\":{\"durable_write\":\"true\"}}}");
+
+    Optional<LsmWriteSpec> spec = lsm.getLsmWriteSpec();
+
+    assertTrue(spec.isPresent());
+    assertEquals(LsmWriteSpec.Sharding.BUCKET, spec.get().sharding());
+    assertEquals("id", spec.get().column());
+    assertEquals(Integer.valueOf(16), spec.get().numBuckets());
+    assertEquals(Arrays.asList("id_idx"), spec.get().maintainedIndexes());
+    assertEquals("true", spec.get().writerConfigDefaults().get("durable_write"));
+  }
+
+  @Test
+  public void testGetLsmWriteSpecAbsent() {
+    enqueue("get_lsm_write_spec", 200, "{\"lsm_write_spec\":null}");
+
+    assertFalse(lsm.getLsmWriteSpec().isPresent());
+  }
+
+  // ===========================================================================
+  // stats
+  // ===========================================================================
+
+  @Test
+  public void testGetLsmStats() throws Exception {
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
+
+    Optional<JsonNode> got = lsm.getLsmStats(true);
+
+    assertEquals("/v1/table/my_table/get_lsm_stats/", requestPaths.get(0));
+    assertTrue(MAPPER.readTree(requestBodies.get(0)).get("include_generation_rows").asBoolean());
+    assertTrue(got.isPresent());
+    assertEquals("shard-0", got.get().get("buckets").get(0).get("shard_id").asText());
+  }
+
+  @Test
+  public void testGetLsmStatsAbsentWhenLsmDisabled() {
+    enqueue("get_lsm_stats", 200, "{\"lsm_stats\":null}");
+
+    assertFalse(lsm.getLsmStats().isPresent());
+  }
+
+  @Test
+  public void testGetLsmStatsDefaultsToExcludingGenerationRows() throws Exception {
+    enqueue("get_lsm_stats", 200, stats());
+
+    lsm.getLsmStats();
+
+    assertFalse(MAPPER.readTree(requestBodies.get(0)).get("include_generation_rows").asBoolean());
+  }
+
+  // ===========================================================================
+  // flush / compact
+  // ===========================================================================
+
+  @Test
+  public void testFlushAndCompactRoutes() {
+    enqueue("flush_lsm", 200, "");
+    enqueue("compact_lsm", 200, "");
+
+    lsm.flushLsm();
+    lsm.compactLsm();
+
+    assertEquals("/v1/table/my_table/flush_lsm/", requestPaths.get(0));
+    assertEquals("/v1/table/my_table/compact_lsm/", requestPaths.get(1));
+  }
+
+  @Test
+  public void testHttpErrorCarriesStatus() {
+    enqueue("flush_lsm", 404, "no such table");
+
+    LanceDbRestClient.HttpException e =
+        assertThrows(LanceDbRestClient.HttpException.class, () -> lsm.flushLsm());
+    assertEquals(404, e.statusCode());
+  }
+
+  // ===========================================================================
+  // checkpoint
+  // ===========================================================================
+
+  @Test
+  public void testCheckpointReturnsWhenLsmDisabled() {
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 200, "{\"lsm_stats\":null}");
+
+    lsm.checkpointLsm();
+
+    assertEquals(0, countCalls("compact_lsm"), "nothing to compact when the LSM path is off");
+  }
+
+  @Test
+  public void testCheckpointReturnsWhenNoGenerationsOutstanding() {
+    enqueue("flush_lsm", 200, "");
+    // A bucket with no L0 generations yields no target, so the drain never starts.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));
+
+    lsm.checkpointLsm();
+
+    assertEquals(0, countCalls("compact_lsm"));
+  }
+
+  @Test
+  public void testCheckpointConvergesOnceTargetGenerationsAreGone() {
+    enqueue("flush_lsm", 200, "");
+    // Watermark read: shard-0 holds generations 7 and 8, so target = 8.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
+    // First drain poll: both still outstanding, nothing compacting -> dispatch a pass.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
+    // Second drain poll: drained past the target -> done.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 9L)));
+    enqueue("compact_lsm", 200, "");
+
+    lsm.checkpointLsm();
+
+    assertEquals(1, countCalls("compact_lsm"), "one pass dispatched");
+    assertEquals(3, countCalls("get_lsm_stats"), "watermark read plus two drain polls");
+  }
+
+  @Test
+  public void testCheckpointDoesNotPileOnWhileEveryTargetBucketIsCompacting() {
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", true, 4L)));
+    // Still compacting on the first poll, so no pass is dispatched; then it drains.
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", true, 4L)));
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 5L)));
+
+    lsm.checkpointLsm();
+
+    assertEquals(0, countCalls("compact_lsm"), "a latched bucket is left alone");
+  }
+
+  @Test
+  public void testCheckpointRetriesFromFlushAfterLostClaim() {
+    // 421 on the watermark read: the node lost its claim, so the whole thing restarts
+    // from flush rather than retrying the read in place.
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 421, "no claim");
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));
+
+    lsm.checkpointLsm();
+
+    assertEquals(2, countCalls("flush_lsm"), "re-issued from flush");
+  }
+
+  @Test
+  public void testCheckpointRetriesRetryableStatusInPlace() {
+    enqueue("flush_lsm", 429, "latch held");
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false)));
+
+    lsm.checkpointLsm();
+
+    assertEquals(2, countCalls("flush_lsm"), "429 retried in place, not re-issued");
+  }
+
+  @Test
+  public void testCheckpointPropagatesTerminalStatus() {
+    enqueue("flush_lsm", 400, "bad request");
+
+    LanceDbRestClient.HttpException e =
+        assertThrows(LanceDbRestClient.HttpException.class, () -> lsm.checkpointLsm());
+    assertEquals(400, e.statusCode());
+    assertEquals(1, countCalls("flush_lsm"), "a terminal status is not retried");
+  }
+
+  @Test
+  public void testCheckpointGivesUpAfterRepeatedLostClaims() {
+    enqueue("flush_lsm", 421, "no claim");
+
+    IllegalStateException e =
+        assertThrows(IllegalStateException.class, () -> lsm.checkpointLsm());
+    assertTrue(e.getMessage().contains("kept losing its claim"), e.getMessage());
+    assertEquals(4, countCalls("flush_lsm"), "the initial attempt plus MAX_REISSUES");
+  }
+
+  // ===========================================================================
+  // harness
+  // ===========================================================================
+
+  /** Build an {@code lsm_stats} response body from bucket fragments. */
+  private static String stats(String... buckets) {
+    return "{\"lsm_stats\":{\"buckets\":[" + String.join(",", buckets) + "]}}";
+  }
+
+  private static String bucket(String shardId, boolean compacting, Long... generations) {
+    StringBuilder gens = new StringBuilder();
+    for (Long generation : generations) {
+      if (gens.length() > 0) {
+        gens.append(",");
+      }
+      gens.append("{\"generation\":").append(generation).append(",\"bytes\":1024}");
+    }
+    return "{\"shard_id\":\""
+        + shardId
+        + "\",\"status\":\"Active\",\"writer_epoch\":1,\"manifest_version\":2,"
+        + "\"current_generation\":9,\"replay_after_wal_entry_position\":0,"
+        + "\"wal_entry_position_last_seen\":0,\"generations\":["
+        + gens
+        + "],\"compacting\":"
+        + compacting
+        + "}";
+  }
+
+  /** Queue a reply for an operation. The last queued reply repeats once the queue drains. */
+  private void enqueue(String operation, int status, String body) {
+    replies
+        .computeIfAbsent(operation, key -> new ArrayDeque<Reply>())
+        .add(new Reply(status, body));
+  }
+
+  private Reply nextReply(String path) {
+    String operation = operationOf(path);
+    Deque<Reply> queued = replies.get(operation);
+    if (queued == null || queued.isEmpty()) {
+      return new Reply(200, "");
+    }
+    return queued.size() > 1 ? queued.poll() : queued.peek();
+  }
+
+  private long countCalls(String operation) {
+    return requestPaths.stream().filter(path -> operationOf(path).equals(operation)).count();
+  }
+
+  /** {@code /v1/table/my_table/flush_lsm/} -> {@code flush_lsm}. */
+  private static String operationOf(String path) {
+    String[] segments = path.split("/");
+    return segments.length == 0 ? "" : segments[segments.length - 1];
+  }
+
+  private static String readAll(InputStream in) throws IOException {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    byte[] buffer = new byte[4096];
+    int read;
+    while ((read = in.read(buffer)) != -1) {
+      out.write(buffer, 0, read);
+    }
+    return new String(out.toByteArray(), StandardCharsets.UTF_8);
+  }
+
+  private static final class Reply {
+    private final int status;
+    private final String body;
+
+    private Reply(int status, String body) {
+      this.status = status;
+      this.body = body;
+    }
+  }
+}