From a651b67c76e47638c8f02848ce472ba93cde74d9 Mon Sep 17 00:00:00 2001 From: Daniel Rammer Date: Mon, 17 Aug 2026 14:42:37 -0500 Subject: [PATCH] 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) --- docs/src/js/classes/Table.md | 93 ++++ docs/src/js/globals.md | 4 + docs/src/js/interfaces/BucketStats.md | 116 +++++ docs/src/js/interfaces/GenerationStats.md | 40 ++ docs/src/js/interfaces/LsmStats.md | 22 + docs/src/js/interfaces/MemtableStats.md | 60 +++ docs/src/python/python.md | 2 + java/README.md | 41 ++ java/lancedb-core/pom.xml | 14 + .../LanceDbNamespaceClientBuilder.java | 49 +- .../java/com/lancedb/LanceDbRestClient.java | 114 +++++ .../java/com/lancedb/LanceDbTableLsm.java | 405 ++++++++++++++++ .../main/java/com/lancedb/LsmWriteSpec.java | 260 +++++++++++ .../java/com/lancedb/LanceDbTableLsmTest.java | 434 ++++++++++++++++++ nodejs/__test__/table.test.ts | 53 +++ nodejs/lancedb/index.ts | 4 + nodejs/lancedb/table.ts | 78 ++++ nodejs/src/table.rs | 151 ++++++ python/python/lancedb/__init__.py | 2 + python/python/lancedb/table.py | 2 +- 20 files changed, 1928 insertions(+), 16 deletions(-) create mode 100644 docs/src/js/interfaces/BucketStats.md create mode 100644 docs/src/js/interfaces/GenerationStats.md create mode 100644 docs/src/js/interfaces/LsmStats.md create mode 100644 docs/src/js/interfaces/MemtableStats.md create mode 100644 java/lancedb-core/src/main/java/com/lancedb/LanceDbRestClient.java create mode 100644 java/lancedb-core/src/main/java/com/lancedb/LanceDbTableLsm.java create mode 100644 java/lancedb-core/src/main/java/com/lancedb/LsmWriteSpec.java create mode 100644 java/lancedb-core/src/test/java/com/lancedb/LanceDbTableLsmTest.java diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index 4479bf4e4..06dc8479e 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -213,6 +213,39 @@ version of the table. *** +### checkpointLsm() + +```ts +abstract checkpointLsm(): Promise +``` + +Converge this table's LSM write path into its base table. + +Seals once, then triggers compaction and polls until the L0 that existed +at the start is gone. The target set is fixed at the start, so +generations created *during* 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, and safe to run on a cadence. + +There is no liveness bound — the compactor pool is shared across tables, +so a checkpoint queued behind unrelated work looks exactly like one that +is merging. The caller owns the deadline. + +#### Returns + +`Promise`<`void`> + +#### Example + +```ts +const before = await table.getLsmStats(); +await table.checkpointLsm(); +const after = await table.getLsmStats(); +``` + +*** + ### close() ```ts @@ -250,6 +283,24 @@ It is a no-op when no writers are cached. *** +### compactLsm() + +```ts +abstract compactLsm(): Promise +``` + +Trigger a background L0 → base compaction pass per bucket. + +Returns once the passes are *dispatched*, not once they finish — watch +[Table#getLsmStats](Table.md#getlsmstats) for progress, or use +[Table#checkpointLsm](Table.md#checkpointlsm) to wait for convergence. + +#### Returns + +`Promise`<`void`> + +*** + ### countRows() ```ts @@ -448,6 +499,48 @@ Drop an index from the table. *** +### flushLsm() + +```ts +abstract flushLsm(): Promise +``` + +Seal every bucket's active memtable into a new L0 generation. + +Returns once the seal is committed. Sealing an empty memtable is a no-op, +so this is safe to call repeatedly. + +#### Returns + +`Promise`<`void`> + +*** + +### getLsmStats() + +```ts +abstract getLsmStats(includeGenerationRows?): Promise +``` + +Read live per-bucket LSM state. + +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. + +Resolves to `undefined` only when the LSM write path is not enabled. + +#### Parameters + +* **includeGenerationRows?**: `boolean` + Also count rows per L0 generation. + Off by default because each count opens an uncached Lance dataset. + +#### Returns + +`Promise`<`undefined` \| [`LsmStats`](../interfaces/LsmStats.md)> + +*** + ### getLsmWriteSpec() ```ts diff --git a/docs/src/js/globals.md b/docs/src/js/globals.md index bd2ca54b5..462907cfd 100644 --- a/docs/src/js/globals.md +++ b/docs/src/js/globals.md @@ -58,6 +58,7 @@ - [BranchDiff](interfaces/BranchDiff.md) - [BranchIndexSummary](interfaces/BranchIndexSummary.md) - [BranchRowCountSummary](interfaces/BranchRowCountSummary.md) +- [BucketStats](interfaces/BucketStats.md) - [ClientConfig](interfaces/ClientConfig.md) - [ColumnAlteration](interfaces/ColumnAlteration.md) - [ColumnOrdering](interfaces/ColumnOrdering.md) @@ -81,6 +82,7 @@ - [FtsToken](interfaces/FtsToken.md) - [FullTextQuery](interfaces/FullTextQuery.md) - [FullTextSearchOptions](interfaces/FullTextSearchOptions.md) +- [GenerationStats](interfaces/GenerationStats.md) - [HnswPqOptions](interfaces/HnswPqOptions.md) - [HnswSqOptions](interfaces/HnswSqOptions.md) - [IndexConfig](interfaces/IndexConfig.md) @@ -94,7 +96,9 @@ - [JobInfo](interfaces/JobInfo.md) - [ListNamespacesOptions](interfaces/ListNamespacesOptions.md) - [ListNamespacesResponse](interfaces/ListNamespacesResponse.md) +- [LsmStats](interfaces/LsmStats.md) - [LsmWriteSpec](interfaces/LsmWriteSpec.md) +- [MemtableStats](interfaces/MemtableStats.md) - [MergeBlocker](interfaces/MergeBlocker.md) - [MergeBranchResult](interfaces/MergeBranchResult.md) - [MergePreview](interfaces/MergePreview.md) diff --git a/docs/src/js/interfaces/BucketStats.md b/docs/src/js/interfaces/BucketStats.md new file mode 100644 index 000000000..3f5095672 --- /dev/null +++ b/docs/src/js/interfaces/BucketStats.md @@ -0,0 +1,116 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / BucketStats + +# Interface: BucketStats + +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. + +## Properties + +### compacting + +```ts +compacting: boolean; +``` + +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". + +*** + +### currentGeneration + +```ts +currentGeneration: number; +``` + +The generation the active memtable will become. + +*** + +### generations + +```ts +generations: GenerationStats[]; +``` + +Flushed L0 generations not yet merged into the base table. + +*** + +### manifestVersion + +```ts +manifestVersion: number; +``` + +Version of the shard manifest these numbers were read from. + +*** + +### memtables? + +```ts +optional memtables: MemtableStats[]; +``` + +Oldest first, active last. Absent for a `"Sealed"` bucket, whose +in-memory state is torn down. + +*** + +### replayAfterWalEntryPosition + +```ts +replayAfterWalEntryPosition: number; +``` + +WAL position replay resumes from. + +*** + +### shardId + +```ts +shardId: string; +``` + +The shard this bucket writes. + +*** + +### status + +```ts +status: string; +``` + +`"Active"` or `"Sealed"` (drop-table 2PC in flight). + +*** + +### walEntryPositionLastSeen + +```ts +walEntryPositionLastSeen: number; +``` + +Highest WAL position the writer has seen. The difference against +`replayAfterWalEntryPosition` is the WAL lag. + +*** + +### writerEpoch + +```ts +writerEpoch: number; +``` + +Epoch of the writer that currently owns the shard. diff --git a/docs/src/js/interfaces/GenerationStats.md b/docs/src/js/interfaces/GenerationStats.md new file mode 100644 index 000000000..19dd2afda --- /dev/null +++ b/docs/src/js/interfaces/GenerationStats.md @@ -0,0 +1,40 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / GenerationStats + +# Interface: GenerationStats + +One flushed L0 generation. + +## Properties + +### bytes + +```ts +bytes: number; +``` + +On-disk size of the generation. + +*** + +### generation + +```ts +generation: number; +``` + +The generation number. Increases as memtables are sealed into L0. + +*** + +### rows? + +```ts +optional rows: number; +``` + +Present only when `includeGenerationRows` was requested. Off by default +because each count opens an uncached Lance dataset. diff --git a/docs/src/js/interfaces/LsmStats.md b/docs/src/js/interfaces/LsmStats.md new file mode 100644 index 000000000..76a2f50db --- /dev/null +++ b/docs/src/js/interfaces/LsmStats.md @@ -0,0 +1,22 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / LsmStats + +# Interface: LsmStats + +Live per-bucket LSM state, as returned by `Table#getLsmStats`. + +Nothing here is derived: sums and differences (total L0 bytes, WAL lag) are +the caller's to compute. + +## Properties + +### buckets + +```ts +buckets: BucketStats[]; +``` + +One entry per bucket backing this table. diff --git a/docs/src/js/interfaces/MemtableStats.md b/docs/src/js/interfaces/MemtableStats.md new file mode 100644 index 000000000..fdc1e4467 --- /dev/null +++ b/docs/src/js/interfaces/MemtableStats.md @@ -0,0 +1,60 @@ +[**@lancedb/lancedb**](../README.md) • **Docs** + +*** + +[@lancedb/lancedb](../globals.md) / MemtableStats + +# Interface: MemtableStats + +One in-memory memtable. + +## Properties + +### batches + +```ts +batches: number; +``` + +Record batches currently buffered. + +*** + +### bytes + +```ts +bytes: number; +``` + +Estimated in-memory size. + +*** + +### generation + +```ts +generation: number; +``` + +The generation this memtable will become once sealed. + +*** + +### indexes + +```ts +indexes: string[]; +``` + +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". + +*** + +### rows + +```ts +rows: number; +``` + +Rows currently buffered. diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 3dd6f59f4..1d5975dee 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -52,6 +52,8 @@ listing a storage directory. ::: lancedb.table.Branches +::: lancedb.LsmWriteSpec + ## Expressions Type-safe expression builder for filters and projections. Use these instead diff --git a/java/README.md b/java/README.md index d3560ba4d..5369c7497 100644 --- a/java/README.md +++ b/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: diff --git a/java/lancedb-core/pom.xml b/java/lancedb-core/pom.xml index 09b088e46..dd3fe9ed4 100644 --- a/java/lancedb-core/pom.xml +++ b/java/lancedb-core/pom.xml @@ -33,6 +33,20 @@ arrow-memory-netty + + + org.apache.httpcomponents.client5 + httpclient5 + 5.2.1 + + + + com.fasterxml.jackson.core + jackson-databind + 2.17.1 + + org.junit.jupiter junit-jupiter diff --git a/java/lancedb-core/src/main/java/com/lancedb/LanceDbNamespaceClientBuilder.java b/java/lancedb-core/src/main/java/com/lancedb/LanceDbNamespaceClientBuilder.java index 5e31aaaa1..da241dfd5 100644 --- a/java/lancedb-core/src/main/java/com/lancedb/LanceDbNamespaceClientBuilder.java +++ b/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 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. + * + *

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()}. + * + *

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

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

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

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; + } + } +} diff --git a/java/lancedb-core/src/main/java/com/lancedb/LanceDbTableLsm.java b/java/lancedb-core/src/main/java/com/lancedb/LanceDbTableLsm.java new file mode 100644 index 000000000..b7c01013e --- /dev/null +++ b/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. + * + *

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

These routes are not part of the Lance Namespace specification, so they are issued directly + * rather than through {@link org.lance.namespace.LanceNamespace}. + * + * 

{@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();
+ * }
+ */ +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. + * + *

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

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

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

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

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

Returns once the passes are dispatched, 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. + * + *

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. + * + * @param includeGenerationRows Also count rows per L0 generation. Off by default because each + * count opens an uncached Lance dataset. + */ + 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")) { + return Optional.empty(); + } + return Optional.of(response.get("lsm_stats")); + } + + /** Equivalent to {@code getLsmStats(false)}. */ + public Optional getLsmStats() { + return getLsmStats(false); + } + + /** + * Converge this table's LSM write path into its base table. + * + *

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

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

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> 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 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 targets) { + while (true) { + Attempt> 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 newestGenerations(JsonNode 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); + } + } + 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). + */ + 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. + * + *

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

An exhausted budget propagates the last error as itself rather than a synthesized one — "429 + * after nine tries" beats "checkpoint failed". + */ + private static Attempt issue(Call call) { + int retries = 0; + while (true) { + try { + return new Attempt(call.run(), false); + } catch (LanceDbRestClient.HttpException e) { + if (isLostClaim(e)) { + return new Attempt(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 { + 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 run(); + } +} diff --git a/java/lancedb-core/src/main/java/com/lancedb/LsmWriteSpec.java b/java/lancedb-core/src/main/java/com/lancedb/LsmWriteSpec.java new file mode 100644 index 000000000..da0966910 --- /dev/null +++ b/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}. + * + *

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

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 + * nothing, while a fresh spec here maintains every 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 maintainedIndexes; + private final Map writerConfigDefaults; + + private LsmWriteSpec( + Sharding sharding, + String column, + Integer numBuckets, + List maintainedIndexes, + Map 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. + * + *

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()); + } + + /** + * Identity sharding — shard by the raw value of {@code column} — maintaining every index on the + * table. + * + *

{@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()); + } + + /** 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()); + } + + /** + * Set the indexes the MemWAL keeps up to date as rows are appended. + * + *

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

Note that {@code null} and the empty list mean opposite things here. + */ + public LsmWriteSpec withMaintainedIndexes(List maintainedIndexes) { + return new LsmWriteSpec( + sharding, + column, + numBuckets, + maintainedIndexes == null ? null : new ArrayList(maintainedIndexes), + writerConfigDefaults); + } + + /** + * Set default {@code ShardWriter} configuration recorded in the MemWAL index. + * + *

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 writerConfigDefaults) { + if (writerConfigDefaults == null) { + throw new IllegalArgumentException("writerConfigDefaults cannot be null"); + } + return new LsmWriteSpec( + sharding, + column, + numBuckets, + maintainedIndexes, + new HashMap(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 maintainedIndexes() { + return maintainedIndexes == null ? null : Collections.unmodifiableList(maintainedIndexes); + } + + /** Default {@code ShardWriter} configuration recorded in the MemWAL index. */ + public Map writerConfigDefaults() { + return Collections.unmodifiableMap(writerConfigDefaults); + } + + /** Render this spec as the {@code set_lsm_write_spec} request body. */ + Map toRequestBody() { + Map shardingBody = new LinkedHashMap(); + shardingBody.put("mode", sharding.wireName()); + if (column != null) { + shardingBody.put("column", column); + } + if (numBuckets != null) { + shardingBody.put("num_buckets", numBuckets); + } + + Map body = new LinkedHashMap(); + 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. + * + *

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 maintainedIndexes = new ArrayList(); + JsonNode indexesNode = node.get("maintained_indexes"); + if (indexesNode != null && indexesNode.isArray()) { + for (JsonNode index : indexesNode) { + maintainedIndexes.add(index.asText()); + } + } + + Map defaults = new HashMap(); + 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 + + "}"; + } +} diff --git a/java/lancedb-core/src/test/java/com/lancedb/LanceDbTableLsmTest.java b/java/lancedb-core/src/test/java/com/lancedb/LanceDbTableLsmTest.java new file mode 100644 index 000000000..70cf1d647 --- /dev/null +++ b/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. + * + *

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 requestPaths = Collections.synchronizedList(new ArrayList()); + private final List requestBodies = Collections.synchronizedList(new ArrayList()); + private final Map> replies = new ConcurrentHashMap>(); + + @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.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 defaults = new HashMap(); + 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 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 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()) + .add(new Reply(status, body)); + } + + private Reply nextReply(String path) { + String operation = operationOf(path); + Deque 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; + } + } +} diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts index 5396a251a..80c50f1ac 100644 --- a/nodejs/__test__/table.test.ts +++ b/nodejs/__test__/table.test.ts @@ -3341,6 +3341,59 @@ describe("LSM merge insert", () => { }); }); +describe("LSM convergence and stats", () => { + let tmpDir: tmp.DirResult; + + beforeEach(() => { + tmpDir = tmp.dirSync({ unsafeCleanup: true }); + }); + afterEach(() => tmpDir.removeCallback()); + + async function lsmTable(conn: Connection): Promise { + const table = await conn.createEmptyTable( + "t", + new arrow.Schema([new arrow.Field("id", new arrow.Utf8(), false)]), + ); + await table.setUnenforcedPrimaryKey("id"); + await table.setLsmWriteSpec({ specType: "unsharded" }); + return table; + } + + // These four route through the server that owns the MemWAL, so a local table + // rejects them rather than answering. What is asserted here is that the + // bindings reach the core at all; the behavior against a real endpoint is + // covered by the mocked endpoint tests in rust/lancedb/src/remote/table.rs. + it("rejects flushLsm on a local table", async () => { + const conn = await connect(tmpDir.name); + const table = await lsmTable(conn); + + await expect(table.flushLsm()).rejects.toThrow(/not supported/i); + }); + + it("rejects compactLsm on a local table", async () => { + const conn = await connect(tmpDir.name); + const table = await lsmTable(conn); + + await expect(table.compactLsm()).rejects.toThrow(/not supported/i); + }); + + it("rejects getLsmStats on a local table", async () => { + const conn = await connect(tmpDir.name); + const table = await lsmTable(conn); + + await expect(table.getLsmStats()).rejects.toThrow(/not supported/i); + await expect(table.getLsmStats(true)).rejects.toThrow(/not supported/i); + }); + + it("rejects checkpointLsm on a local table", async () => { + const conn = await connect(tmpDir.name); + const table = await lsmTable(conn); + + // checkpointLsm seals first, so it surfaces flushLsm's rejection. + await expect(table.checkpointLsm()).rejects.toThrow(/not supported/i); + }); +}); + describe("computed columns", () => { let tmpDir: tmp.DirResult; beforeEach(() => { diff --git a/nodejs/lancedb/index.ts b/nodejs/lancedb/index.ts index 9f2e97989..6a5bfe3b4 100644 --- a/nodejs/lancedb/index.ts +++ b/nodejs/lancedb/index.ts @@ -147,6 +147,10 @@ export { FtsToken, TokenizeTableOptions, LsmWriteSpec, + LsmStats, + BucketStats, + GenerationStats, + MemtableStats, ColumnAlteration, FieldMetadataUpdate, } from "./table"; diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index a7dc8def1..964c2cea3 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -31,6 +31,7 @@ import { IndexConfig, IndexStatistics, Job, + LsmStats, Branches as NativeBranches, OptimizeStats, RefreshColumnResult, @@ -50,6 +51,12 @@ import { import { sanitizeType } from "./sanitize"; import { IntoSql, toSQL } from "./util"; export { IndexConfig } from "./native"; +export { + BucketStats, + GenerationStats, + LsmStats, + MemtableStats, +} from "./native"; /** * Progress snapshot for a write operation, delivered to the `progress` @@ -706,6 +713,59 @@ export abstract class Table { * @returns {Promise} */ abstract closeLsmWriters(): Promise; + /** + * Seal every bucket's active memtable into a new L0 generation. + * + * Returns once the seal is committed. Sealing an empty memtable is a no-op, + * so this is safe to call repeatedly. + * @returns {Promise} + */ + abstract flushLsm(): Promise; + /** + * Trigger a background L0 → base compaction pass per bucket. + * + * Returns once the passes are *dispatched*, not once they finish — watch + * {@link Table#getLsmStats} for progress, or use + * {@link Table#checkpointLsm} to wait for convergence. + * @returns {Promise} + */ + abstract compactLsm(): Promise; + /** + * Converge this table's LSM write path into its base table. + * + * Seals once, then triggers compaction and polls until the L0 that existed + * at the start is gone. The target set is fixed at the start, so + * generations created *during* 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, and safe to run on a cadence. + * + * There is no liveness bound — the compactor pool is shared across tables, + * so a checkpoint queued behind unrelated work looks exactly like one that + * is merging. The caller owns the deadline. + * @returns {Promise} + * @example + * ```ts + * const before = await table.getLsmStats(); + * await table.checkpointLsm(); + * const after = await table.getLsmStats(); + * ``` + */ + abstract checkpointLsm(): Promise; + /** + * Read live per-bucket LSM state. + * + * 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. + * + * Resolves to `undefined` only when the LSM write path is not enabled. + * @param {boolean} includeGenerationRows Also count rows per L0 generation. + * Off by default because each count opens an uncached Lance dataset. + * @returns {Promise} + */ + abstract getLsmStats( + includeGenerationRows?: boolean, + ): Promise; /** Retrieve the version of the table */ abstract version(): Promise; @@ -1266,6 +1326,24 @@ export class LocalTable extends Table { return await this.inner.closeLsmWriters(); } + async flushLsm(): Promise { + return await this.inner.flushLsm(); + } + + async compactLsm(): Promise { + return await this.inner.compactLsm(); + } + + async checkpointLsm(): Promise { + return await this.inner.checkpointLsm(); + } + + async getLsmStats( + includeGenerationRows: boolean = false, + ): Promise { + return (await this.inner.getLsmStats(includeGenerationRows)) ?? undefined; + } + async version(): Promise { return await this.inner.version(); } diff --git a/nodejs/src/table.rs b/nodejs/src/table.rs index 4c45be668..b15491202 100644 --- a/nodejs/src/table.rs +++ b/nodejs/src/table.rs @@ -497,6 +497,34 @@ impl Table { self.inner_ref()?.close_lsm_writers().await.default_error() } + #[napi(catch_unwind)] + pub async fn flush_lsm(&self) -> napi::Result<()> { + self.inner_ref()?.flush_lsm().await.default_error() + } + + #[napi(catch_unwind)] + pub async fn compact_lsm(&self) -> napi::Result<()> { + self.inner_ref()?.compact_lsm().await.default_error() + } + + #[napi(catch_unwind)] + pub async fn checkpoint_lsm(&self) -> napi::Result<()> { + self.inner_ref()?.checkpoint_lsm().await.default_error() + } + + #[napi(catch_unwind)] + pub async fn get_lsm_stats( + &self, + include_generation_rows: bool, + ) -> napi::Result> { + let stats = self + .inner_ref()? + .get_lsm_stats(include_generation_rows) + .await + .default_error()?; + Ok(stats.map(LsmStats::from)) + } + #[napi(catch_unwind)] pub async fn version(&self) -> napi::Result { self.inner_ref()? @@ -889,6 +917,129 @@ impl From for LsmWriteSpec { } } +/// One flushed L0 generation. +#[napi(object)] +#[derive(Clone, Debug)] +pub struct GenerationStats { + /// The generation number. Increases as memtables are sealed into L0. + pub generation: i64, + /// On-disk size of the generation. + pub bytes: i64, + /// Present only when `includeGenerationRows` was requested. Off by default + /// because each count opens an uncached Lance dataset. + pub rows: Option, +} + +impl From for GenerationStats { + fn from(g: lancedb::table::GenerationStats) -> Self { + Self { + generation: g.generation as i64, + bytes: g.bytes as i64, + rows: g.rows.map(|r| r as i64), + } + } +} + +/// One in-memory memtable. +#[napi(object)] +#[derive(Clone, Debug)] +pub struct MemtableStats { + /// The generation this memtable will become once sealed. + pub generation: i64, + /// Rows currently buffered. + pub rows: i64, + /// Estimated in-memory size. + pub bytes: i64, + /// Record batches currently buffered. + pub batches: i64, + /// 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". + pub indexes: Vec, +} + +impl From for MemtableStats { + fn from(m: lancedb::table::MemtableStats) -> Self { + Self { + generation: m.generation as i64, + rows: m.rows as i64, + bytes: m.bytes as i64, + batches: m.batches as i64, + indexes: m.indexes, + } + } +} + +/// 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. +#[napi(object)] +#[derive(Clone, Debug)] +pub struct BucketStats { + /// The shard this bucket writes. + pub shard_id: String, + /// `"Active"` or `"Sealed"` (drop-table 2PC in flight). + pub status: String, + /// Epoch of the writer that currently owns the shard. + pub writer_epoch: i64, + /// Version of the shard manifest these numbers were read from. + pub manifest_version: i64, + /// The generation the active memtable will become. + pub current_generation: i64, + /// WAL position replay resumes from. + pub replay_after_wal_entry_position: i64, + /// Highest WAL position the writer has seen. The difference against + /// `replayAfterWalEntryPosition` is the WAL lag. + pub wal_entry_position_last_seen: i64, + /// Flushed L0 generations not yet merged into the base table. + pub generations: Vec, + /// 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". + pub compacting: bool, + /// Oldest first, active last. Absent for a `"Sealed"` bucket, whose + /// in-memory state is torn down. + pub memtables: Option>, +} + +impl From for BucketStats { + fn from(b: lancedb::table::BucketStats) -> Self { + Self { + shard_id: b.shard_id, + status: b.status, + writer_epoch: b.writer_epoch as i64, + manifest_version: b.manifest_version as i64, + current_generation: b.current_generation as i64, + replay_after_wal_entry_position: b.replay_after_wal_entry_position as i64, + wal_entry_position_last_seen: b.wal_entry_position_last_seen as i64, + generations: b.generations.into_iter().map(Into::into).collect(), + compacting: b.compacting, + memtables: b + .memtables + .map(|ms| ms.into_iter().map(Into::into).collect()), + } + } +} + +/// Live per-bucket LSM state, as returned by `Table#getLsmStats`. +/// +/// Nothing here is derived: sums and differences (total L0 bytes, WAL lag) are +/// the caller's to compute. +#[napi(object)] +#[derive(Clone, Debug)] +pub struct LsmStats { + /// One entry per bucket backing this table. + pub buckets: Vec, +} + +impl From for LsmStats { + fn from(stats: lancedb::table::LsmStats) -> Self { + Self { + buckets: stats.buckets.into_iter().map(Into::into).collect(), + } + } +} + /// Statistics about a compaction operation. #[napi(object)] #[derive(Clone, Debug)] diff --git a/python/python/lancedb/__init__.py b/python/python/lancedb/__init__.py index 235049f97..e12ef4e86 100644 --- a/python/python/lancedb/__init__.py +++ b/python/python/lancedb/__init__.py @@ -12,6 +12,7 @@ __version__ = importlib.metadata.version("lancedb") from ._lancedb import connect as lancedb_connect from ._lancedb import FtsToken +from ._lancedb import LsmWriteSpec from ._lancedb import tokenize as _tokenize from .common import URI, sanitize_uri from urllib.parse import urlparse @@ -518,6 +519,7 @@ __all__ = [ "Job", "LanceDBConnection", "LanceNamespaceDBConnection", + "LsmWriteSpec", "RemoteDBConnection", "Session", "Table", diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 4ecf6e836..2237ffdc2 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -4801,7 +4801,7 @@ class AsyncTable: Examples -------- - >>> from lancedb._lancedb import LsmWriteSpec + >>> from lancedb import LsmWriteSpec >>> # table.set_unenforced_primary_key("id") >>> # table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 16)) """