From 4a04584ce0723945132cb193ef15ccadfdcd7414 Mon Sep 17 00:00:00 2001 From: Daniel Rammer Date: Tue, 18 Aug 2026 19:48:23 -0500 Subject: [PATCH] fix(java): let the checkpoint loop own its retry budget and decode LSM stats strictly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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` to `Optional`, 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) --- java/README.md | 3 +- .../main/java/com/lancedb/BucketStats.java | 194 ++++++++++++++++++ .../java/com/lancedb/GenerationStats.java | 64 ++++++ .../src/main/java/com/lancedb/JsonFields.java | 109 ++++++++++ .../java/com/lancedb/LanceDbRestClient.java | 7 +- .../java/com/lancedb/LanceDbTableLsm.java | 61 +++--- .../src/main/java/com/lancedb/LsmStats.java | 56 +++++ .../main/java/com/lancedb/MemtableStats.java | 99 +++++++++ .../java/com/lancedb/LanceDbTableLsmTest.java | 150 +++++++++++++- 9 files changed, 698 insertions(+), 45 deletions(-) create mode 100644 java/lancedb-core/src/main/java/com/lancedb/BucketStats.java create mode 100644 java/lancedb-core/src/main/java/com/lancedb/GenerationStats.java create mode 100644 java/lancedb-core/src/main/java/com/lancedb/JsonFields.java create mode 100644 java/lancedb-core/src/main/java/com/lancedb/LsmStats.java create mode 100644 java/lancedb-core/src/main/java/com/lancedb/MemtableStats.java diff --git a/java/README.md b/java/README.md index 5369c7497..c46c8174b 100644 --- a/java/README.md +++ b/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(); ``` diff --git a/java/lancedb-core/src/main/java/com/lancedb/BucketStats.java b/java/lancedb-core/src/main/java/com/lancedb/BucketStats.java new file mode 100644 index 000000000..2a8060c5d --- /dev/null +++ b/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 generations; + private final boolean compacting; + private final List memtables; + + BucketStats( + String shardId, + String status, + long writerEpoch, + long manifestVersion, + long currentGeneration, + long replayAfterWalEntryPosition, + long walEntryPositionLastSeen, + List generations, + boolean compacting, + List 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 generations() { + return generations; + } + + /** + * Whether a pass owns this bucket's compaction latch right now. Says a driver is + * running, not whose, 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> 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. + * + *

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 generations = new ArrayList(); + for (JsonNode generation : JsonFields.requiredArray(node, "generations", CONTEXT)) { + generations.add(GenerationStats.fromJson(generation)); + } + + JsonNode memtablesNode = JsonFields.optionalArray(node, "memtables", CONTEXT); + List memtables = null; + if (memtablesNode != null) { + memtables = new ArrayList(); + 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 + + "}"; + } +} diff --git a/java/lancedb-core/src/main/java/com/lancedb/GenerationStats.java b/java/lancedb-core/src/main/java/com/lancedb/GenerationStats.java new file mode 100644 index 000000000..12222407c --- /dev/null +++ b/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 + "}"; + } +} diff --git a/java/lancedb-core/src/main/java/com/lancedb/JsonFields.java b/java/lancedb-core/src/main/java/com/lancedb/JsonFields.java new file mode 100644 index 000000000..b78e2411a --- /dev/null +++ b/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. + * + *

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. + * + *

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; + } +} diff --git a/java/lancedb-core/src/main/java/com/lancedb/LanceDbRestClient.java b/java/lancedb-core/src/main/java/com/lancedb/LanceDbRestClient.java index f8b390feb..baafbb9df 100644 --- a/java/lancedb-core/src/main/java/com/lancedb/LanceDbRestClient.java +++ b/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(); } /** diff --git a/java/lancedb-core/src/main/java/com/lancedb/LanceDbTableLsm.java b/java/lancedb-core/src/main/java/com/lancedb/LanceDbTableLsm.java index b7c01013e..23b18199e 100644 --- a/java/lancedb-core/src/main/java/com/lancedb/LanceDbTableLsm.java +++ b/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 { *

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. * - *

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. + *

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 getLsmStats(boolean includeGenerationRows) { + public Optional getLsmStats(boolean includeGenerationRows) { Map body = new LinkedHashMap(); 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 getLsmStats() { + public Optional getLsmStats() { return getLsmStats(false); } @@ -202,7 +210,7 @@ public class LanceDbTableLsm { continue; } - Attempt> stats = issue(() -> getLsmStats(false)); + Attempt> stats = issue(() -> getLsmStats(false)); if (stats.lostClaim) { backoff(reissue); continue; @@ -234,7 +242,7 @@ public class LanceDbTableLsm { */ private boolean drainToTargets(Map targets) { while (true) { - Attempt> stats = issue(() -> getLsmStats(false)); + Attempt> 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 newestGenerations(JsonNode stats) { + private static Map newestGenerations(LsmStats stats) { Map targets = new HashMap(); - 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. - * - *

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). diff --git a/java/lancedb-core/src/main/java/com/lancedb/LsmStats.java b/java/lancedb-core/src/main/java/com/lancedb/LsmStats.java new file mode 100644 index 000000000..3496ebc96 --- /dev/null +++ b/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()}. + * + *

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 buckets; + + LsmStats(List buckets) { + this.buckets = Collections.unmodifiableList(buckets); + } + + /** One entry per bucket. */ + public List buckets() { + return buckets; + } + + static LsmStats fromJson(JsonNode node) { + JsonFields.requiredObject(node, CONTEXT); + List buckets = new ArrayList(); + 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 + "}"; + } +} diff --git a/java/lancedb-core/src/main/java/com/lancedb/MemtableStats.java b/java/lancedb-core/src/main/java/com/lancedb/MemtableStats.java new file mode 100644 index 000000000..777e915aa --- /dev/null +++ b/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 indexes; + + MemtableStats(long generation, long rows, long bytes, long batches, List 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 indexes() { + return indexes; + } + + static MemtableStats fromJson(JsonNode node) { + JsonFields.requiredObject(node, CONTEXT); + List indexes = new ArrayList(); + 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 + + "}"; + } +} diff --git a/java/lancedb-core/src/test/java/com/lancedb/LanceDbTableLsmTest.java b/java/lancedb-core/src/test/java/com/lancedb/LanceDbTableLsmTest.java index 70cf1d647..e84fa5421 100644 --- a/java/lancedb-core/src/test/java/com/lancedb/LanceDbTableLsmTest.java +++ b/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 got = lsm.getLsmStats(true); + Optional 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 malformed = new LinkedHashMap(); + 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 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 generationNumbers(BucketStats bucket) { + List numbers = new ArrayList(); + 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()) - .add(new Reply(status, body)); + replies.computeIfAbsent(operation, key -> new ArrayDeque()).add(new Reply(status, body)); } private Reply nextReply(String path) {