fix(java): let the checkpoint loop own its retry budget and decode LSM stats strictly

Two fixes to the Java MemWAL LSM surface, both reproduced against the
scripted test server before being fixed.

Transport retries. `HttpClients.createDefault()` installs Apache's default
response retry strategy, whose retryable-status list is exactly 429 and 503
— the two statuses `LanceDbTableLsm.isRetryable` owns. Every explicit retry
budget in `checkpointLsm` was therefore doubled on the wire (a 429 held
against flush issued 18 requests, not 9), and `compactLsm` was retried in
place despite the loop being built to fall through to a fresh stats poll.
Automatic retries are now disabled, so the checkpoint loop is the sole owner
of the 421/429/503 transitions.

Stats decoding. `getLsmStats` read the response with Jackson's `path()`,
which yields a missing node that iterates as an empty array. That made
"malformed" indistinguishable from "no buckets", which is indistinguishable
from "drained" — so an empty response body, a `{"lsm_stats": {}}`, or a
bucket missing its required fields all made `checkpointLsm()` report
convergence for a checkpoint that never ran.

Stats now decode into `LsmStats`, `BucketStats`, `GenerationStats` and
`MemtableStats`, mirroring the Rust structs in
`rust/lancedb/src/table/lsm_stats.rs` and the objects the Node binding
already exposes. Decoding is strict and fails closed, matching the serde
contract on the Rust side: absent or null `lsm_stats` means the LSM write
path is off, and anything else present must decode into the full
bucket/generation shape. `newestGeneration` and `outstandingGenerations`
move onto `BucketStats`, matching Rust's `impl BucketStats`.

This changes `getLsmStats` from `Optional<JsonNode>` to `Optional<LsmStats>`,
which also brings Java to the typed surface Node already had rather than
handing back a raw JSON blob.

Testing: 33 passing in lancedb-core, up from 29. The new tests pin the wire
request count against the retry budget and reject five malformed stats
payloads. `testCheckpointRetriesRetryableStatusInPlace` previously passed on
a transport-absorbed 429 and now exercises the real retry path.

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

java/README.md

@@ -56,7 +56,8 @@ lsm.setLsmWriteSpec(LsmWriteSpec.bucket("id", 16));
 lsm.checkpointLsm();
 
 // Inspect live per-bucket state.
-lsm.getLsmStats().ifPresent(stats -> System.out.println(stats.get("buckets")));
+lsm.getLsmStats().ifPresent(stats -> stats.buckets().forEach(bucket ->
+    System.out.println(bucket.shardId() + ": " + bucket.generations().size() + " L0 generations")));
 
 client.close();
 ```

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

@@ -0,0 +1,194 @@
+/*
+ * 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.List;
+import java.util.Optional;
+import java.util.OptionalLong;
+
+/**
+ * Live state of one bucket. A table is N buckets on one node; flattening to a single number hides
+ * the one hot bucket that is usually why someone opened this endpoint.
+ */
+public class BucketStats {
+  private static final String CONTEXT = "bucket stats";
+
+  private final String shardId;
+  private final String status;
+  private final long writerEpoch;
+  private final long manifestVersion;
+  private final long currentGeneration;
+  private final long replayAfterWalEntryPosition;
+  private final long walEntryPositionLastSeen;
+  private final List<GenerationStats> generations;
+  private final boolean compacting;
+  private final List<MemtableStats> memtables;
+
+  BucketStats(
+      String shardId,
+      String status,
+      long writerEpoch,
+      long manifestVersion,
+      long currentGeneration,
+      long replayAfterWalEntryPosition,
+      long walEntryPositionLastSeen,
+      List<GenerationStats> generations,
+      boolean compacting,
+      List<MemtableStats> memtables) {
+    this.shardId = shardId;
+    this.status = status;
+    this.writerEpoch = writerEpoch;
+    this.manifestVersion = manifestVersion;
+    this.currentGeneration = currentGeneration;
+    this.replayAfterWalEntryPosition = replayAfterWalEntryPosition;
+    this.walEntryPositionLastSeen = walEntryPositionLastSeen;
+    this.generations = Collections.unmodifiableList(generations);
+    this.compacting = compacting;
+    this.memtables = memtables == null ? null : Collections.unmodifiableList(memtables);
+  }
+
+  /** The shard this bucket writes. */
+  public String shardId() {
+    return shardId;
+  }
+
+  /** {@code "Active"} or {@code "Sealed"} (drop-table 2PC in flight). */
+  public String status() {
+    return status;
+  }
+
+  /** Epoch of the writer that currently owns the shard. */
+  public long writerEpoch() {
+    return writerEpoch;
+  }
+
+  /** Version of the shard manifest these numbers were read from. */
+  public long manifestVersion() {
+    return manifestVersion;
+  }
+
+  /** The generation the active memtable will become. */
+  public long currentGeneration() {
+    return currentGeneration;
+  }
+
+  /** WAL position replay resumes from. */
+  public long replayAfterWalEntryPosition() {
+    return replayAfterWalEntryPosition;
+  }
+
+  /**
+   * Highest WAL position the writer has seen. The difference against {@link
+   * #replayAfterWalEntryPosition()} is the WAL lag.
+   */
+  public long walEntryPositionLastSeen() {
+    return walEntryPositionLastSeen;
+  }
+
+  /** Flushed L0 generations not yet merged into the base table. */
+  public List<GenerationStats> generations() {
+    return generations;
+  }
+
+  /**
+   * Whether a pass owns this bucket's compaction latch right now. Says <em>a</em> driver is
+   * running, not <em>whose</em>, and the latch is held from dispatch — including while the pass
+   * queues for a pod-wide compactor permit. Read it as "do not pile on", never as "mine is
+   * progressing".
+   */
+  public boolean compacting() {
+    return compacting;
+  }
+
+  /** Oldest first, active last. Empty for a {@code "Sealed"} bucket, whose state is torn down. */
+  public Optional<List<MemtableStats>> memtables() {
+    return Optional.ofNullable(memtables);
+  }
+
+  /** The newest flushed generation, or empty when L0 is empty. */
+  OptionalLong newestGeneration() {
+    OptionalLong newest = OptionalLong.empty();
+    for (GenerationStats generation : generations) {
+      if (!newest.isPresent() || generation.generation() > newest.getAsLong()) {
+        newest = OptionalLong.of(generation.generation());
+      }
+    }
+    return newest;
+  }
+
+  /**
+   * 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. Compaction drains
+   * oldest-first, so this decreases monotonically.
+   */
+  long outstandingGenerations(long target) {
+    long count = 0;
+    for (GenerationStats generation : generations) {
+      if (generation.generation() <= target) {
+        count++;
+      }
+    }
+    return count;
+  }
+
+  static BucketStats fromJson(JsonNode node) {
+    JsonFields.requiredObject(node, CONTEXT);
+    List<GenerationStats> generations = new ArrayList<GenerationStats>();
+    for (JsonNode generation : JsonFields.requiredArray(node, "generations", CONTEXT)) {
+      generations.add(GenerationStats.fromJson(generation));
+    }
+
+    JsonNode memtablesNode = JsonFields.optionalArray(node, "memtables", CONTEXT);
+    List<MemtableStats> memtables = null;
+    if (memtablesNode != null) {
+      memtables = new ArrayList<MemtableStats>();
+      for (JsonNode memtable : memtablesNode) {
+        memtables.add(MemtableStats.fromJson(memtable));
+      }
+    }
+
+    return new BucketStats(
+        JsonFields.requiredText(node, "shard_id", CONTEXT),
+        JsonFields.requiredText(node, "status", CONTEXT),
+        JsonFields.requiredLong(node, "writer_epoch", CONTEXT),
+        JsonFields.requiredLong(node, "manifest_version", CONTEXT),
+        JsonFields.requiredLong(node, "current_generation", CONTEXT),
+        JsonFields.requiredLong(node, "replay_after_wal_entry_position", CONTEXT),
+        JsonFields.requiredLong(node, "wal_entry_position_last_seen", CONTEXT),
+        generations,
+        JsonFields.requiredBoolean(node, "compacting", CONTEXT),
+        memtables);
+  }
+
+  @Override
+  public String toString() {
+    return "BucketStats{shardId="
+        + shardId
+        + ", status="
+        + status
+        + ", currentGeneration="
+        + currentGeneration
+        + ", generations="
+        + generations
+        + ", compacting="
+        + compacting
+        + "}";
+  }
+}

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

@@ -0,0 +1,64 @@
+/*
+ * 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.OptionalLong;
+
+/** One flushed L0 generation. */
+public class GenerationStats {
+  private static final String CONTEXT = "generation stats";
+
+  private final long generation;
+  private final long bytes;
+  private final Long rows;
+
+  GenerationStats(long generation, long bytes, Long rows) {
+    this.generation = generation;
+    this.bytes = bytes;
+    this.rows = rows;
+  }
+
+  /** The generation number. Increases as memtables are sealed into L0. */
+  public long generation() {
+    return generation;
+  }
+
+  /** On-disk size of the generation. */
+  public long bytes() {
+    return bytes;
+  }
+
+  /**
+   * Rows in this generation, present only when {@code includeGenerationRows} was requested. Off by
+   * default because each count opens an uncached Lance dataset.
+   */
+  public OptionalLong rows() {
+    return rows == null ? OptionalLong.empty() : OptionalLong.of(rows);
+  }
+
+  static GenerationStats fromJson(JsonNode node) {
+    JsonFields.requiredObject(node, CONTEXT);
+    return new GenerationStats(
+        JsonFields.requiredLong(node, "generation", CONTEXT),
+        JsonFields.requiredLong(node, "bytes", CONTEXT),
+        JsonFields.optionalLong(node, "rows", CONTEXT));
+  }
+
+  @Override
+  public String toString() {
+    return "GenerationStats{generation=" + generation + ", bytes=" + bytes + ", rows=" + rows + "}";
+  }
+}

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

@@ -0,0 +1,109 @@
+/*
+ * 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;
+
+/**
+ * Strict readers for decoding LanceDB JSON responses.
+ *
+ * <p>Every reader fails closed: a missing, null, or wrong-typed field throws rather than
+ * defaulting. That mirrors the serde decoding the Rust client applies to the same payloads in
+ * {@code rust/lancedb/src/table/lsm_stats.rs}, where a required field has no default and a
+ * malformed response is an error rather than a zero.
+ *
+ * <p>The alternative — Jackson's {@code path()}, which yields a missing node that reads as an empty
+ * array or a zero — is unsafe here because {@link LanceDbTableLsm#checkpointLsm()} decides
+ * convergence from these numbers. A defaulted {@code generations} array is indistinguishable from a
+ * drained one, so a malformed response would report a checkpoint that never happened.
+ */
+final class JsonFields {
+  private JsonFields() {}
+
+  /** The node itself, once confirmed to be a JSON object. */
+  static JsonNode requiredObject(JsonNode node, String context) {
+    if (node == null || !node.isObject()) {
+      throw new IllegalStateException(context + " is not a JSON object: " + node);
+    }
+    return node;
+  }
+
+  static String requiredText(JsonNode owner, String field, String context) {
+    JsonNode value = required(owner, field, context);
+    if (!value.isTextual()) {
+      throw new IllegalStateException(fieldIs(context, field, "a string", value));
+    }
+    return value.asText();
+  }
+
+  static long requiredLong(JsonNode owner, String field, String context) {
+    JsonNode value = required(owner, field, context);
+    if (!value.isIntegralNumber()) {
+      throw new IllegalStateException(fieldIs(context, field, "an integer", value));
+    }
+    return value.asLong();
+  }
+
+  static boolean requiredBoolean(JsonNode owner, String field, String context) {
+    JsonNode value = required(owner, field, context);
+    if (!value.isBoolean()) {
+      throw new IllegalStateException(fieldIs(context, field, "a boolean", value));
+    }
+    return value.asBoolean();
+  }
+
+  static JsonNode requiredArray(JsonNode owner, String field, String context) {
+    JsonNode value = required(owner, field, context);
+    if (!value.isArray()) {
+      throw new IllegalStateException(fieldIs(context, field, "an array", value));
+    }
+    return value;
+  }
+
+  /** Null when the field is absent or JSON null, mirroring a serde {@code Option}. */
+  static Long optionalLong(JsonNode owner, String field, String context) {
+    JsonNode value = owner.get(field);
+    if (value == null || value.isNull()) {
+      return null;
+    }
+    if (!value.isIntegralNumber()) {
+      throw new IllegalStateException(fieldIs(context, field, "an integer", value));
+    }
+    return value.asLong();
+  }
+
+  /** Null when the field is absent or JSON null, mirroring a serde {@code Option}. */
+  static JsonNode optionalArray(JsonNode owner, String field, String context) {
+    JsonNode value = owner.get(field);
+    if (value == null || value.isNull()) {
+      return null;
+    }
+    if (!value.isArray()) {
+      throw new IllegalStateException(fieldIs(context, field, "an array", value));
+    }
+    return value;
+  }
+
+  private static JsonNode required(JsonNode owner, String field, String context) {
+    JsonNode value = owner.get(field);
+    if (value == null || value.isNull()) {
+      throw new IllegalStateException(context + " is missing required field '" + field + "'");
+    }
+    return value;
+  }
+
+  private static String fieldIs(String context, String field, String expected, JsonNode value) {
+    return context + " field '" + field + "' is not " + expected + ": " + value;
+  }
+}

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

@@ -49,7 +49,12 @@ public class LanceDbRestClient implements Closeable {
     this.baseUri = baseUri.endsWith("/") ? baseUri.substring(0, baseUri.length() - 1) : baseUri;
     this.apiKey = apiKey;
     this.database = database;
-    this.http = HttpClients.createDefault();
+    // Automatic retries off, deliberately. The default strategy retries 429 and 503 —
+    // exactly the two statuses LanceDbTableLsm.checkpointLsm() acts on — which would
+    // silently double its explicit retry budget and would also retry compact_lsm in
+    // place, where the loop is designed to fall through to a fresh stats poll instead.
+    // The checkpoint loop owns the 421/429/503 transitions; the transport must not.
+    this.http = HttpClients.custom().disableAutomaticRetries().build();
   }
 
   /**

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

@@ -19,6 +19,7 @@ import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalLong;
 
 /**
  * The MemWAL LSM write path for one LanceDB Cloud or Enterprise table.
@@ -154,24 +155,31 @@ public class LanceDbTableLsm {
    * <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.
+   * <p>Empty only when the LSM write path is not enabled — that is, when the server sends an absent
+   * or null {@code lsm_stats}. A stats object that is present is decoded strictly, and a malformed
+   * one throws rather than decoding to something empty, because {@link #checkpointLsm} reads
+   * convergence out of these numbers and cannot tell a defaulted array from a drained one.
    *
    * @param includeGenerationRows Also count rows per L0 generation. Off by default because each
    *     count opens an uncached Lance dataset.
+   * @throws IllegalStateException if the response is absent or does not decode.
    */
-  public Optional<JsonNode> getLsmStats(boolean includeGenerationRows) {
+  public Optional<LsmStats> 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")) {
+    if (response == null) {
+      throw new IllegalStateException("get_lsm_stats returned an empty response body");
+    }
+    JsonNode stats = response.get("lsm_stats");
+    if (stats == null || stats.isNull()) {
       return Optional.empty();
     }
-    return Optional.of(response.get("lsm_stats"));
+    return Optional.of(LsmStats.fromJson(stats));
   }
 
   /** Equivalent to {@code getLsmStats(false)}. */
-  public Optional<JsonNode> getLsmStats() {
+  public Optional<LsmStats> getLsmStats() {
     return getLsmStats(false);
   }
 
@@ -202,7 +210,7 @@ public class LanceDbTableLsm {
         continue;
       }
 
-      Attempt<Optional<JsonNode>> stats = issue(() -> getLsmStats(false));
+      Attempt<Optional<LsmStats>> stats = issue(() -> getLsmStats(false));
       if (stats.lostClaim) {
         backoff(reissue);
         continue;
@@ -234,7 +242,7 @@ public class LanceDbTableLsm {
    */
   private boolean drainToTargets(Map<String, Long> targets) {
     while (true) {
-      Attempt<Optional<JsonNode>> stats = issue(() -> getLsmStats(false));
+      Attempt<Optional<LsmStats>> stats = issue(() -> getLsmStats(false));
       if (stats.lostClaim) {
         return false;
       }
@@ -248,15 +256,15 @@ public class LanceDbTableLsm {
       // 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());
+      for (BucketStats bucket : stats.value.get().buckets()) {
+        Long target = targets.get(bucket.shardId());
         if (target == null) {
           continue;
         }
-        long remaining = outstandingGenerations(bucket, target);
+        long remaining = bucket.outstandingGenerations(target);
         if (remaining > 0) {
           outstanding += remaining;
-          allCompacting &= bucket.path("compacting").asBoolean(false);
+          allCompacting &= bucket.compacting();
         }
       }
       if (outstanding == 0) {
@@ -284,36 +292,17 @@ public class LanceDbTableLsm {
   }
 
   /** The newest generation held by each bucket, skipping buckets holding none. */
-  private static Map<String, Long> newestGenerations(JsonNode stats) {
+  private static Map<String, Long> newestGenerations(LsmStats 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);
+    for (BucketStats bucket : stats.buckets()) {
+      OptionalLong newest = bucket.newestGeneration();
+      if (newest.isPresent()) {
+        targets.put(bucket.shardId(), newest.getAsLong());
       }
     }
     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).

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

@@ -0,0 +1,56 @@
+/*
+ * 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.List;
+
+/**
+ * Live per-bucket LSM state, as returned by {@link LanceDbTableLsm#getLsmStats()}.
+ *
+ * <p>Nothing here is derived: sums and differences (total L0 bytes, WAL lag) are the caller's to
+ * compute. There is no "LSM is off" shape — that case is an empty {@link java.util.Optional},
+ * because a stats object of zeros would read as measurements.
+ */
+public class LsmStats {
+  private static final String CONTEXT = "lsm stats";
+
+  private final List<BucketStats> buckets;
+
+  LsmStats(List<BucketStats> buckets) {
+    this.buckets = Collections.unmodifiableList(buckets);
+  }
+
+  /** One entry per bucket. */
+  public List<BucketStats> buckets() {
+    return buckets;
+  }
+
+  static LsmStats fromJson(JsonNode node) {
+    JsonFields.requiredObject(node, CONTEXT);
+    List<BucketStats> buckets = new ArrayList<BucketStats>();
+    for (JsonNode bucket : JsonFields.requiredArray(node, "buckets", CONTEXT)) {
+      buckets.add(BucketStats.fromJson(bucket));
+    }
+    return new LsmStats(buckets);
+  }
+
+  @Override
+  public String toString() {
+    return "LsmStats{buckets=" + buckets + "}";
+  }
+}

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

@@ -0,0 +1,99 @@
+/*
+ * 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.List;
+
+/** One in-memory memtable. */
+public class MemtableStats {
+  private static final String CONTEXT = "memtable stats";
+
+  private final long generation;
+  private final long rows;
+  private final long bytes;
+  private final long batches;
+  private final List<String> indexes;
+
+  MemtableStats(long generation, long rows, long bytes, long batches, List<String> indexes) {
+    this.generation = generation;
+    this.rows = rows;
+    this.bytes = bytes;
+    this.batches = batches;
+    this.indexes = Collections.unmodifiableList(indexes);
+  }
+
+  /** The generation this memtable will become once sealed. */
+  public long generation() {
+    return generation;
+  }
+
+  /** Rows currently buffered. */
+  public long rows() {
+    return rows;
+  }
+
+  /** Estimated in-memory size. */
+  public long bytes() {
+    return bytes;
+  }
+
+  /** Record batches currently buffered. */
+  public long batches() {
+    return batches;
+  }
+
+  /**
+   * Names of the indexes this memtable carries. An absent name is the whole answer to "why is my
+   * fresh-tier search on that column brute-force".
+   */
+  public List<String> indexes() {
+    return indexes;
+  }
+
+  static MemtableStats fromJson(JsonNode node) {
+    JsonFields.requiredObject(node, CONTEXT);
+    List<String> indexes = new ArrayList<String>();
+    for (JsonNode index : JsonFields.requiredArray(node, "indexes", CONTEXT)) {
+      if (!index.isTextual()) {
+        throw new IllegalStateException(CONTEXT + " has a non-string index name: " + index);
+      }
+      indexes.add(index.asText());
+    }
+    return new MemtableStats(
+        JsonFields.requiredLong(node, "generation", CONTEXT),
+        JsonFields.requiredLong(node, "rows", CONTEXT),
+        JsonFields.requiredLong(node, "bytes", CONTEXT),
+        JsonFields.requiredLong(node, "batches", CONTEXT),
+        indexes);
+  }
+
+  @Override
+  public String toString() {
+    return "MemtableStats{generation="
+        + generation
+        + ", rows="
+        + rows
+        + ", bytes="
+        + bytes
+        + ", batches="
+        + batches
+        + ", indexes="
+        + indexes
+        + "}";
+  }
+}

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

@@ -23,6 +23,7 @@ import org.junit.jupiter.api.Test;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.UncheckedIOException;
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayDeque;
@@ -31,6 +32,7 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.Deque;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -57,6 +59,24 @@ public class LanceDbTableLsmTest {
 
   @BeforeEach
   public void setUp() throws IOException {
+    start();
+  }
+
+  /** Tear down and restart the scripted server, for a test that scripts several exchanges. */
+  private void setUpFresh() {
+    try {
+      client.close();
+      server.stop(0);
+      requestPaths.clear();
+      requestBodies.clear();
+      replies.clear();
+      start();
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  private void start() throws IOException {
     server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
     server.createContext(
         "/",
@@ -210,12 +230,50 @@ public class LanceDbTableLsmTest {
   public void testGetLsmStats() throws Exception {
     enqueue("get_lsm_stats", 200, stats(bucket("shard-0", false, 7L, 8L)));
 
-    Optional<JsonNode> got = lsm.getLsmStats(true);
+    Optional<LsmStats> 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());
+    BucketStats decoded = got.get().buckets().get(0);
+    assertEquals("shard-0", decoded.shardId());
+    assertEquals("Active", decoded.status());
+    assertEquals(1, decoded.writerEpoch());
+    assertEquals(2, decoded.manifestVersion());
+    assertEquals(9, decoded.currentGeneration());
+    assertFalse(decoded.compacting());
+    assertEquals(Arrays.asList(7L, 8L), generationNumbers(decoded));
+    assertEquals(1024, decoded.generations().get(0).bytes());
+    assertFalse(decoded.generations().get(0).rows().isPresent(), "rows absent unless requested");
+    assertFalse(decoded.memtables().isPresent(), "absent memtables stay absent");
+  }
+
+  /** The optional fields decode when the server does send them. */
+  @Test
+  public void testGetLsmStatsDecodesOptionalFields() {
+    enqueue(
+        "get_lsm_stats",
+        200,
+        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+            + "\"writer_epoch\":1,\"manifest_version\":2,\"current_generation\":9,"
+            + "\"replay_after_wal_entry_position\":3,\"wal_entry_position_last_seen\":11,"
+            + "\"generations\":[{\"generation\":7,\"bytes\":1024,\"rows\":42}],"
+            + "\"compacting\":true,\"memtables\":[{\"generation\":8,\"rows\":5,"
+            + "\"bytes\":64,\"batches\":2,\"indexes\":[\"id_idx\"]}]}]}}");
+
+    BucketStats decoded = lsm.getLsmStats(true).get().buckets().get(0);
+
+    assertEquals(3, decoded.replayAfterWalEntryPosition());
+    assertEquals(11, decoded.walEntryPositionLastSeen());
+    assertTrue(decoded.compacting());
+    assertEquals(42, decoded.generations().get(0).rows().getAsLong());
+    assertTrue(decoded.memtables().isPresent());
+    MemtableStats memtable = decoded.memtables().get().get(0);
+    assertEquals(8, memtable.generation());
+    assertEquals(5, memtable.rows());
+    assertEquals(64, memtable.bytes());
+    assertEquals(2, memtable.batches());
+    assertEquals(Arrays.asList("id_idx"), memtable.indexes());
   }
 
   @Test
@@ -352,16 +410,96 @@ public class LanceDbTableLsmTest {
   public void testCheckpointGivesUpAfterRepeatedLostClaims() {
     enqueue("flush_lsm", 421, "no claim");
 
-    IllegalStateException e =
-        assertThrows(IllegalStateException.class, () -> lsm.checkpointLsm());
+    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");
   }
 
+  // ===========================================================================
+  // strict decoding
+  // ===========================================================================
+
+  /**
+   * A stats payload that does not decode must fail closed. Every one of these bodies used to be
+   * read as "no buckets", which is indistinguishable from a drained table, so {@code checkpointLsm}
+   * reported convergence for a checkpoint that never ran.
+   */
+  @Test
+  public void testCheckpointRejectsMalformedStats() {
+    Map<String, String> malformed = new LinkedHashMap<String, String>();
+    malformed.put("no response body at all", "");
+    malformed.put("stats object with no buckets", "{\"lsm_stats\":{}}");
+    malformed.put("bucket missing its required fields", "{\"lsm_stats\":{\"buckets\":[{}]}}");
+    malformed.put(
+        "bucket missing generations",
+        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+            + "\"writer_epoch\":1,\"manifest_version\":2,\"current_generation\":9,"
+            + "\"replay_after_wal_entry_position\":0,\"wal_entry_position_last_seen\":0,"
+            + "\"compacting\":false}]}}");
+    malformed.put(
+        "generation with a non-numeric generation number",
+        "{\"lsm_stats\":{\"buckets\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
+            + "\"writer_epoch\":1,\"manifest_version\":2,\"current_generation\":9,"
+            + "\"replay_after_wal_entry_position\":0,\"wal_entry_position_last_seen\":0,"
+            + "\"generations\":[{\"generation\":\"7\",\"bytes\":1024}],"
+            + "\"compacting\":false}]}}");
+
+    for (Map.Entry<String, String> each : malformed.entrySet()) {
+      setUpFresh();
+      enqueue("flush_lsm", 200, "");
+      enqueue("get_lsm_stats", 200, each.getValue());
+
+      assertThrows(
+          IllegalStateException.class,
+          () -> lsm.checkpointLsm(),
+          each.getKey() + " must not report convergence");
+    }
+  }
+
+  /** The one shape that legitimately means "this table has no LSM write path". */
+  @Test
+  public void testCheckpointTreatsNullStatsAsNotWalBacked() {
+    enqueue("flush_lsm", 200, "");
+    enqueue("get_lsm_stats", 200, "{\"lsm_stats\":null}");
+
+    lsm.checkpointLsm();
+
+    assertEquals(1, countCalls("get_lsm_stats"));
+  }
+
+  // ===========================================================================
+  // retry budget
+  // ===========================================================================
+
+  /**
+   * The transport must not retry on the checkpoint loop's behalf. Apache HttpClient's default
+   * strategy retries exactly 429 and 503 — the two statuses {@code isRetryable} owns — which
+   * doubled every budget here and also retried {@code compact_lsm} in place, where the loop is
+   * built to fall through to a fresh stats poll instead.
+   */
+  @Test
+  public void testCheckpointRetryBudgetIsNotDoubledByTheTransport() {
+    enqueue("flush_lsm", 429, "latch held");
+
+    LanceDbRestClient.HttpException e =
+        assertThrows(LanceDbRestClient.HttpException.class, () -> lsm.checkpointLsm());
+
+    assertEquals(429, e.statusCode(), "the exhausted budget propagates the last error as itself");
+    assertEquals(9, countCalls("flush_lsm"), "the initial request plus MAX_RETRIES, and no more");
+  }
+
   // ===========================================================================
   // harness
   // ===========================================================================
 
+  private static List<Long> generationNumbers(BucketStats bucket) {
+    List<Long> numbers = new ArrayList<Long>();
+    for (GenerationStats generation : bucket.generations()) {
+      numbers.add(generation.generation());
+    }
+    return numbers;
+  }
+
   /** Build an {@code lsm_stats} response body from bucket fragments. */
   private static String stats(String... buckets) {
     return "{\"lsm_stats\":{\"buckets\":[" + String.join(",", buckets) + "]}}";
@@ -388,9 +526,7 @@ public class LanceDbTableLsmTest {
 
   /** 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));
+    replies.computeIfAbsent(operation, key -> new ArrayDeque<Reply>()).add(new Reply(status, body));
   }
 
   private Reply nextReply(String path) {