mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-21 19:22:27 +00:00
refactor!: rename MemWAL APIs to RTE (realtime engine) (#4598)
## What Renames the user-facing MemWAL surface to **RTE** (realtime engine) across Rust, Python, TypeScript, and Java, including the REST wire format. This follows the LSM → MemWAL rename in #4075. | Before | After | |---|---| | `MemWalWriteSpec` / `MemWalStats` | `RTEWriteSpec` / `RTEStats` | | `enable_memwal` / `disable_memwal` / `get_memwal_spec` | `enable_rte` / `disable_rte` / `get_rte_spec` | | `flush_memwal` / `compact_memwal` / `checkpoint_memwal` | `flush_rte` / `compact_rte` / `checkpoint_rte` | | `get_memwal_stats` / `close_memwal_writers` | `get_rte_stats` / `close_rte_writers` | | `use_memwal` (query, merge insert) | `use_rte` | | TS/Java `enableMemWal`, `useMemWal`, ... | `enableRTE`, `useRTE`, ... (napi `js_name` set explicitly) | | Java `LanceDbTableMemWal` | `LanceDbTableRTE` | | `table/memwal_stats.rs` | `table/rte_stats.rs` | Docs and comments now say "realtime engine (RTE)". Text about Lance's underlying mechanism keeps "MemWAL" ("Lance's MemWAL", "MemWAL index"). **Left alone:** Lance-owned names (`MemWalIndexDetails`, `DatasetMemWalExt`, `InitializeMemWal*`, `mem_wal` paths), internal `Lsm*` names and modules, and the `does not support with_row_id` refusal text that the hybrid fallback matches on. ## Compatibility Breaking, for an experimental API. The wire format changes, so this must roll out together with the matching sophon change (lancedb/sophon#8870): - Routes: `/enable_memwal/` → `/enable_rte/`, `/disable_memwal/` → `/disable_rte/`, `/get_memwal_spec/` → `/get_rte_spec/`, `/get_memwal_stats/` → `/get_rte_stats/`, `/flush_memwal/` → `/flush_rte/`, `/compact_memwal/` → `/compact_rte/` - Response keys: `memwal_spec` → `rte_spec`, `memwal_stats` → `rte_stats` - Request field: `use_memwal` → `use_rte` on query and merge-insert bodies The SQL keyword becomes `ALTER TABLE ... SET RTE` server-side. The guides under `docs/web` will be updated separately. ## Validation - Rust: `cargo clippy --features remote --tests --examples` clean; `cargo test --features remote -p lancedb --lib`: 1533 passed - Java: `mvn test`: 37 passed (Spotless applied with JDK 11) - Python: `ruff format` / `ruff check` clean; targeted tests (`test_rte_spec`, `test_merge_insert_lsm`, `test_remote_db`, `test_wal_hybrid`, `test_query`, `test_permutation`, `test_table`): 708 passed - TypeScript: `pnpm build`; `table`, `query` tests: 606 passed; `biome check` clean; `docs/src/js` regenerated with `pnpm run docs` 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dan Rammer
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
1772f4ef83
commit
92825f6f13
@@ -0,0 +1,28 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<projectDescription>
|
||||
<name>lancedb-parent</name>
|
||||
<comment></comment>
|
||||
<projects>
|
||||
</projects>
|
||||
<buildSpec>
|
||||
<buildCommand>
|
||||
<name>org.eclipse.m2e.core.maven2Builder</name>
|
||||
<arguments>
|
||||
</arguments>
|
||||
</buildCommand>
|
||||
</buildSpec>
|
||||
<natures>
|
||||
<nature>org.eclipse.m2e.core.maven2Nature</nature>
|
||||
</natures>
|
||||
<filteredResources>
|
||||
<filter>
|
||||
<id>1786980408448</id>
|
||||
<name></name>
|
||||
<type>30</type>
|
||||
<matcher>
|
||||
<id>org.eclipse.core.resources.regexFilterMatcher</id>
|
||||
<arguments>node_modules|\.git|__CREATED_BY_JAVA_LANGUAGE_SERVER__</arguments>
|
||||
</matcher>
|
||||
</filter>
|
||||
</filteredResources>
|
||||
</projectDescription>
|
||||
@@ -0,0 +1,2 @@
|
||||
eclipse.preferences.version=1
|
||||
encoding/<project>=UTF-8
|
||||
@@ -0,0 +1,4 @@
|
||||
activeProfiles=
|
||||
eclipse.preferences.version=1
|
||||
resolveWorkspaceProjects=true
|
||||
version=1
|
||||
+9
-9
@@ -29,34 +29,34 @@ LanceNamespace namespaceClient = LanceDbNamespaceClientBuilder.newBuilder()
|
||||
.build();
|
||||
```
|
||||
|
||||
## MemWAL write path
|
||||
## RTE 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
|
||||
generated from the Lance Namespace specification. The RTE LSM routes are not part
|
||||
of that specification, so they are issued through a separate client:
|
||||
|
||||
```java
|
||||
import com.lancedb.LanceDbRestClient;
|
||||
import com.lancedb.LanceDbTableMemWal;
|
||||
import com.lancedb.MemWalWriteSpec;
|
||||
import com.lancedb.LanceDbTableRTE;
|
||||
import com.lancedb.RTEWriteSpec;
|
||||
|
||||
LanceDbRestClient client = LanceDbNamespaceClientBuilder.newBuilder()
|
||||
.apiKey("your_lancedb_cloud_api_key")
|
||||
.database("your_database_name")
|
||||
.buildRestClient();
|
||||
|
||||
LanceDbTableMemWal memWal = new LanceDbTableMemWal(client, "my_table");
|
||||
LanceDbTableRTE rte = new LanceDbTableRTE(client, "my_table");
|
||||
|
||||
// Route future merge_insert upserts through the MemWAL, hash-bucketed by `id`.
|
||||
memWal.enableMemWal(MemWalWriteSpec.bucket("id", 16));
|
||||
// Route future merge_insert upserts through the RTE, hash-bucketed by `id`.
|
||||
rte.enableRTE(RTEWriteSpec.bucket("id", 16));
|
||||
|
||||
// ... merge_insert traffic ...
|
||||
|
||||
// Converge the fresh tier into the base table.
|
||||
memWal.checkpointMemWal();
|
||||
rte.checkpointRTE();
|
||||
|
||||
// Inspect live per-table-shard state.
|
||||
memWal.getMemWalStats().ifPresent(stats -> stats.tableShards().forEach(tableShard ->
|
||||
rte.getRTEStats().ifPresent(stats -> stats.tableShards().forEach(tableShard ->
|
||||
System.out.println(tableShard.shardId() + ": " + tableShard.sstables().size() + " SSTables")));
|
||||
|
||||
client.close();
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<classpath>
|
||||
<classpathentry kind="src" output="target/classes" path="src/main/java">
|
||||
<attributes>
|
||||
<attribute name="optional" value="true"/>
|
||||
<attribute name="maven.pomderived" value="true"/>
|
||||
</attributes>
|
||||
</classpathentry>
|
||||
<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources">
|
||||
<attributes>
|
||||
<attribute name="maven.pomderived" value="true"/>
|
||||
<attribute name="optional" value="true"/>
|
||||
</attributes>
|
||||
</classpathentry>
|
||||
<classpathentry kind="src" output="target/test-classes" path="src/test/java">
|
||||
<attributes>
|
||||
<attribute name="optional" value="true"/>
|
||||
<attribute name="maven.pomderived" value="true"/>
|
||||
<attribute name="test" value="true"/>
|
||||
</attributes>
|
||||
</classpathentry>
|
||||
<classpathentry excluding="**" kind="src" output="target/test-classes" path="src/test/resources">
|
||||
<attributes>
|
||||
<attribute name="maven.pomderived" value="true"/>
|
||||
<attribute name="test" value="true"/>
|
||||
<attribute name="optional" value="true"/>
|
||||
</attributes>
|
||||
</classpathentry>
|
||||
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-11">
|
||||
<attributes>
|
||||
<attribute name="maven.pomderived" value="true"/>
|
||||
</attributes>
|
||||
</classpathentry>
|
||||
<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
|
||||
<attributes>
|
||||
<attribute name="maven.pomderived" value="true"/>
|
||||
</attributes>
|
||||
</classpathentry>
|
||||
<classpathentry kind="src" path="target/generated-sources/annotations">
|
||||
<attributes>
|
||||
<attribute name="optional" value="true"/>
|
||||
<attribute name="maven.pomderived" value="true"/>
|
||||
<attribute name="ignore_optional_problems" value="true"/>
|
||||
<attribute name="m2e-apt" value="true"/>
|
||||
</attributes>
|
||||
</classpathentry>
|
||||
<classpathentry kind="src" output="target/test-classes" path="target/generated-test-sources/test-annotations">
|
||||
<attributes>
|
||||
<attribute name="optional" value="true"/>
|
||||
<attribute name="maven.pomderived" value="true"/>
|
||||
<attribute name="ignore_optional_problems" value="true"/>
|
||||
<attribute name="m2e-apt" value="true"/>
|
||||
<attribute name="test" value="true"/>
|
||||
</attributes>
|
||||
</classpathentry>
|
||||
<classpathentry kind="output" path="target/classes"/>
|
||||
</classpath>
|
||||
@@ -0,0 +1,34 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<projectDescription>
|
||||
<name>lancedb-core</name>
|
||||
<comment></comment>
|
||||
<projects>
|
||||
</projects>
|
||||
<buildSpec>
|
||||
<buildCommand>
|
||||
<name>org.eclipse.jdt.core.javabuilder</name>
|
||||
<arguments>
|
||||
</arguments>
|
||||
</buildCommand>
|
||||
<buildCommand>
|
||||
<name>org.eclipse.m2e.core.maven2Builder</name>
|
||||
<arguments>
|
||||
</arguments>
|
||||
</buildCommand>
|
||||
</buildSpec>
|
||||
<natures>
|
||||
<nature>org.eclipse.jdt.core.javanature</nature>
|
||||
<nature>org.eclipse.m2e.core.maven2Nature</nature>
|
||||
</natures>
|
||||
<filteredResources>
|
||||
<filter>
|
||||
<id>1786980408445</id>
|
||||
<name></name>
|
||||
<type>30</type>
|
||||
<matcher>
|
||||
<id>org.eclipse.core.resources.regexFilterMatcher</id>
|
||||
<arguments>node_modules|\.git|__CREATED_BY_JAVA_LANGUAGE_SERVER__</arguments>
|
||||
</matcher>
|
||||
</filter>
|
||||
</filteredResources>
|
||||
</projectDescription>
|
||||
@@ -0,0 +1,5 @@
|
||||
eclipse.preferences.version=1
|
||||
encoding//src/main/java=UTF-8
|
||||
encoding//src/test/java=UTF-8
|
||||
encoding//src/test/resources=UTF-8
|
||||
encoding/<project>=UTF-8
|
||||
@@ -0,0 +1,2 @@
|
||||
eclipse.preferences.version=1
|
||||
org.eclipse.jdt.apt.aptEnabled=false
|
||||
@@ -0,0 +1,9 @@
|
||||
eclipse.preferences.version=1
|
||||
org.eclipse.jdt.core.compiler.codegen.targetPlatform=11
|
||||
org.eclipse.jdt.core.compiler.compliance=11
|
||||
org.eclipse.jdt.core.compiler.problem.enablePreviewFeatures=disabled
|
||||
org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
|
||||
org.eclipse.jdt.core.compiler.problem.reportPreviewFeatures=ignore
|
||||
org.eclipse.jdt.core.compiler.processAnnotations=disabled
|
||||
org.eclipse.jdt.core.compiler.release=disabled
|
||||
org.eclipse.jdt.core.compiler.source=11
|
||||
@@ -0,0 +1,4 @@
|
||||
activeProfiles=
|
||||
eclipse.preferences.version=1
|
||||
resolveWorkspaceProjects=true
|
||||
version=1
|
||||
@@ -20,11 +20,11 @@ import com.fasterxml.jackson.databind.JsonNode;
|
||||
*
|
||||
* <p>Every reader fails closed: a missing, null, or wrong-typed field throws rather than
|
||||
* defaulting. That mirrors the serde decoding the Rust client applies to the same payloads in
|
||||
* {@code rust/lancedb/src/table/memwal_stats.rs}, where a required field has no default and a
|
||||
* {@code rust/lancedb/src/table/rte_stats.rs}, where a required field has no default and a
|
||||
* malformed response is an error rather than a zero.
|
||||
*
|
||||
* <p>The alternative — Jackson's {@code path()}, which yields a missing node that reads as an empty
|
||||
* array or a zero — is unsafe here because {@link LanceDbTableMemWal#checkpointMemWal()} decides
|
||||
* array or a zero — is unsafe here because {@link LanceDbTableRTE#checkpointRTE()} decides
|
||||
* convergence from these numbers. A defaulted {@code sstables} array is indistinguishable from a
|
||||
* drained one, so a malformed response would report a checkpoint that never happened.
|
||||
*/
|
||||
|
||||
@@ -151,8 +151,8 @@ public class LanceDbNamespaceClientBuilder {
|
||||
* Build a {@link LanceDbRestClient} for the same endpoint.
|
||||
*
|
||||
* <p>Needed only for LanceDB routes that the Lance Namespace specification does not cover — the
|
||||
* MemWAL write path, reached through {@link LanceDbTableMemWal}. Every other table
|
||||
* operation belongs on the {@link LanceNamespace} from {@link #build()}.
|
||||
* RTE write path, reached through {@link LanceDbTableRTE}. Every other table operation belongs on
|
||||
* the {@link LanceNamespace} from {@link #build()}.
|
||||
*
|
||||
* <p>The returned client owns an HTTP connection pool; close it when you are done with it.
|
||||
*
|
||||
|
||||
@@ -31,9 +31,9 @@ import java.io.UncheckedIOException;
|
||||
* specification does not cover.
|
||||
*
|
||||
* <p>Most table operations reach LanceDB through {@link org.lance.namespace.LanceNamespace}, which
|
||||
* is generated from the namespace spec. A handful of routes — the MemWAL 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 LanceDbTableMemWal}.
|
||||
* is generated from the namespace spec. A handful of routes — the RTE 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 LanceDbTableRTE}.
|
||||
*
|
||||
* <p>Obtain one from {@link LanceDbNamespaceClientBuilder#buildRestClient()}.
|
||||
*/
|
||||
@@ -50,8 +50,8 @@ public class LanceDbRestClient implements Closeable {
|
||||
this.apiKey = apiKey;
|
||||
this.database = database;
|
||||
// Automatic retries off, deliberately. The default strategy retries 429 and 503 —
|
||||
// exactly the two statuses LanceDbTableMemWal.checkpointMemWal() acts on — which would
|
||||
// silently double its explicit retry budget and would also retry compact_memwal in
|
||||
// exactly the two statuses LanceDbTableRTE.checkpointRTE() acts on — which would
|
||||
// silently double its explicit retry budget and would also retry compact_rte 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();
|
||||
@@ -98,9 +98,8 @@ public class LanceDbRestClient implements Closeable {
|
||||
/**
|
||||
* A non-2xx response.
|
||||
*
|
||||
* <p>The status is exposed because callers act on it: {@link
|
||||
* LanceDbTableMemWal#checkpointMemWal()} treats 429 and 503 as retryable and 421 as a lost node
|
||||
* claim.
|
||||
* <p>The status is exposed because callers act on it: {@link LanceDbTableRTE#checkpointRTE()}
|
||||
* 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;
|
||||
|
||||
+58
-58
@@ -22,11 +22,11 @@ import java.util.Optional;
|
||||
import java.util.OptionalLong;
|
||||
|
||||
/**
|
||||
* The MemWAL write path for one LanceDB Cloud or Enterprise table.
|
||||
* The RTE write path for one LanceDB Cloud or Enterprise table.
|
||||
*
|
||||
* <p>Installing an {@link MemWalWriteSpec} 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, freeze into SSTables, and are merged into the base table by compaction.
|
||||
* <p>Installing an {@link RTEWriteSpec} routes {@code mergeInsert} upserts through Lance's RTE — an
|
||||
* LSM-style append — instead of the standard merge path. Rows land in an in-memory memtable, freeze
|
||||
* into SSTables, and are merged into the base table by compaction.
|
||||
*
|
||||
* <p>These routes are not part of the Lance Namespace specification, so they are issued directly
|
||||
* rather than through {@link org.lance.namespace.LanceNamespace}.
|
||||
@@ -37,23 +37,23 @@ import java.util.OptionalLong;
|
||||
* .database("your_database_name")
|
||||
* .buildRestClient();
|
||||
*
|
||||
* LanceDbTableMemWal memWal = new LanceDbTableMemWal(client, "my_table");
|
||||
* memWal.enableMemWal(MemWalWriteSpec.bucket("id", 16));
|
||||
* LanceDbTableRTE rte = new LanceDbTableRTE(client, "my_table");
|
||||
* rte.enableRTE(RTEWriteSpec.bucket("id", 16));
|
||||
* // ... merge_insert traffic ...
|
||||
* memWal.checkpointMemWal();
|
||||
* rte.checkpointRTE();
|
||||
* }</pre>
|
||||
*/
|
||||
public class LanceDbTableMemWal {
|
||||
public class LanceDbTableRTE {
|
||||
|
||||
/**
|
||||
* Interval between {@code get_memwal_stats} polls during a checkpoint. One interval is roughly
|
||||
* one compaction pass, the granularity at which the answer can change.
|
||||
* Interval between {@code get_rte_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 flushMemWal} after a 421, so a crash-looping node cannot turn
|
||||
* flush → compact → 421 → flush into a spin.
|
||||
* Cap on re-issues from {@code flushRTE} after a 421, so a crash-looping node cannot turn flush →
|
||||
* compact → 421 → flush into a spin.
|
||||
*
|
||||
* <p>Deliberately not shared with {@link #MAX_RETRIES}: a claim that keeps evaporating is a
|
||||
* broken node, while contention is routine and wants a real budget.
|
||||
@@ -79,7 +79,7 @@ public class LanceDbTableMemWal {
|
||||
* @param tableIdentifier The table's full identifier, {@code $}-delimited when it sits inside a
|
||||
* namespace, such as {@code analytics$events}.
|
||||
*/
|
||||
public LanceDbTableMemWal(LanceDbRestClient client, String tableIdentifier) {
|
||||
public LanceDbTableRTE(LanceDbRestClient client, String tableIdentifier) {
|
||||
if (client == null) {
|
||||
throw new IllegalArgumentException("Client cannot be null");
|
||||
}
|
||||
@@ -91,46 +91,46 @@ public class LanceDbTableMemWal {
|
||||
}
|
||||
|
||||
/**
|
||||
* Install an {@link MemWalWriteSpec} on this table, selecting the MemWAL write path for
|
||||
* future {@code mergeInsert} calls.
|
||||
* Install an {@link RTEWriteSpec} on this table, selecting the RTE write path for future {@code
|
||||
* mergeInsert} calls.
|
||||
*
|
||||
* <p>All variants require the table to have an unenforced primary key; bucket sharding
|
||||
* additionally requires it to be the single column being bucketed.
|
||||
*/
|
||||
public void enableMemWal(MemWalWriteSpec spec) {
|
||||
public void enableRTE(RTEWriteSpec spec) {
|
||||
if (spec == null) {
|
||||
throw new IllegalArgumentException("Spec cannot be null");
|
||||
}
|
||||
client.post(route("enable_memwal"), spec.toRequestBody());
|
||||
client.post(route("enable_rte"), spec.toRequestBody());
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove the {@link MemWalWriteSpec} from this table, reverting to the standard {@code
|
||||
* mergeInsert} write path.
|
||||
* Remove the {@link RTEWriteSpec} from this table, reverting to the standard {@code mergeInsert}
|
||||
* write path.
|
||||
*
|
||||
* <p>Errors if no spec is currently set.
|
||||
*/
|
||||
public void disableMemWal() {
|
||||
client.post(route("disable_memwal"), null);
|
||||
public void disableRTE() {
|
||||
client.post(route("disable_rte"), null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the {@link MemWalWriteSpec} currently installed on this table.
|
||||
* Read the {@link RTEWriteSpec} currently installed on this table.
|
||||
*
|
||||
* <p>Empty when the MemWAL write path is not enabled at all — no spec has been set, or one was
|
||||
* removed with {@link #disableMemWal()}. That is a different answer from a present spec whose
|
||||
* {@link MemWalWriteSpec#maintainedIndexes()} is null, which selects indexes automatically.
|
||||
* <p>Empty when the RTE write path is not enabled at all — no spec has been set, or one was
|
||||
* removed with {@link #disableRTE()}. That is a different answer from a present spec whose {@link
|
||||
* RTEWriteSpec#maintainedIndexes()} is null, which selects indexes automatically.
|
||||
*
|
||||
* <p>The spec read back is the one that was installed, selection included: null maintains every
|
||||
* supported index the table has now or gains later, an empty list maintains none, and a non-empty
|
||||
* list maintains exactly those. All three round-trip.
|
||||
*/
|
||||
public Optional<MemWalWriteSpec> getMemWalSpec() {
|
||||
JsonNode response = client.post(route("get_memwal_spec"), null);
|
||||
if (response == null || !response.hasNonNull("memwal_spec")) {
|
||||
public Optional<RTEWriteSpec> getRTESpec() {
|
||||
JsonNode response = client.post(route("get_rte_spec"), null);
|
||||
if (response == null || !response.hasNonNull("rte_spec")) {
|
||||
return Optional.empty();
|
||||
}
|
||||
return Optional.of(MemWalWriteSpec.fromJson(response.get("memwal_spec")));
|
||||
return Optional.of(RTEWriteSpec.fromJson(response.get("rte_spec")));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -139,56 +139,56 @@ public class LanceDbTableMemWal {
|
||||
* <p>Returns once the freeze is committed. Freezing an empty memtable is a no-op, so this is safe
|
||||
* to call repeatedly.
|
||||
*/
|
||||
public void flushMemWal() {
|
||||
client.post(route("flush_memwal"), null);
|
||||
public void flushRTE() {
|
||||
client.post(route("flush_rte"), null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Trigger a background SSTable compaction pass per table shard.
|
||||
*
|
||||
* <p>Returns once the passes are <em>dispatched</em>, not once they finish — watch {@link
|
||||
* #getMemWalStats}, or use {@link #checkpointMemWal} to wait for convergence.
|
||||
* #getRTEStats}, or use {@link #checkpointRTE} to wait for convergence.
|
||||
*/
|
||||
public void compactMemWal() {
|
||||
client.post(route("compact_memwal"), null);
|
||||
public void compactRTE() {
|
||||
client.post(route("compact_rte"), null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Read live per-table-shard MemWAL state.
|
||||
* Read live per-table-shard RTE state.
|
||||
*
|
||||
* <p>Answers "how far behind is my fresh tier", "which table shard is hot", and "why is my
|
||||
* fresh-tier vector search brute-force". Mutates no table state.
|
||||
*
|
||||
* <p>Empty only when the MemWAL write path is not enabled — that is, when the server sends an
|
||||
* absent or null {@code memwal_stats}. A stats object that is present is decoded strictly, and a
|
||||
* malformed one throws rather than decoding to something empty, because {@link #checkpointMemWal}
|
||||
* reads convergence out of these numbers and cannot tell a defaulted array from a drained one.
|
||||
* <p>Empty only when the RTE write path is not enabled — that is, when the server sends an absent
|
||||
* or null {@code rte_stats}. A stats object that is present is decoded strictly, and a malformed
|
||||
* one throws rather than decoding to something empty, because {@link #checkpointRTE} reads
|
||||
* convergence out of these numbers and cannot tell a defaulted array from a drained one.
|
||||
*
|
||||
* @param includeSstableRows Also count rows per SSTable. Off by default because each count opens
|
||||
* an uncached Lance dataset.
|
||||
* @throws IllegalStateException if the response is absent or does not decode.
|
||||
*/
|
||||
public Optional<MemWalStats> getMemWalStats(boolean includeSstableRows) {
|
||||
public Optional<RTEStats> getRTEStats(boolean includeSstableRows) {
|
||||
Map<String, Object> body = new LinkedHashMap<String, Object>();
|
||||
body.put("include_sstable_rows", includeSstableRows);
|
||||
JsonNode response = client.post(route("get_memwal_stats"), body);
|
||||
JsonNode response = client.post(route("get_rte_stats"), body);
|
||||
if (response == null) {
|
||||
throw new IllegalStateException("get_memwal_stats returned an empty response body");
|
||||
throw new IllegalStateException("get_rte_stats returned an empty response body");
|
||||
}
|
||||
JsonNode stats = response.get("memwal_stats");
|
||||
JsonNode stats = response.get("rte_stats");
|
||||
if (stats == null || stats.isNull()) {
|
||||
return Optional.empty();
|
||||
}
|
||||
return Optional.of(MemWalStats.fromJson(stats));
|
||||
return Optional.of(RTEStats.fromJson(stats));
|
||||
}
|
||||
|
||||
/** Equivalent to {@code getMemWalStats(false)}. */
|
||||
public Optional<MemWalStats> getMemWalStats() {
|
||||
return getMemWalStats(false);
|
||||
/** Equivalent to {@code getRTEStats(false)}. */
|
||||
public Optional<RTEStats> getRTEStats() {
|
||||
return getRTEStats(false);
|
||||
}
|
||||
|
||||
/**
|
||||
* Converge this table's MemWAL write path into its base table.
|
||||
* Converge this table's RTE write path into its base table.
|
||||
*
|
||||
* <p>Freezes once, fixes a target watermark from the resulting SSTables, then triggers compaction
|
||||
* and polls until those SSTables are gone. The target set is fixed at the start, so sstables
|
||||
@@ -196,31 +196,31 @@ public class LanceDbTableMemWal {
|
||||
* load, and what makes it best-effort: it converges the fresh tier as of some instant.
|
||||
* Idempotent, abandonable at any point, safe on a cadence.
|
||||
*
|
||||
* <p>The loop runs here, not on the server: {@link #compactMemWal} dispatches a pass and returns,
|
||||
* so nothing holds a socket and a client can vanish mid-operation with nothing to reconcile.
|
||||
* <p>The loop runs here, not on the server: {@link #compactRTE} dispatches a pass and returns, so
|
||||
* nothing holds a socket and a client can vanish mid-operation with nothing to reconcile.
|
||||
* Completion is read from generation numbers in the shard manifest — durable state, unlike a
|
||||
* count in a compact response, which a concurrent write invalidates.
|
||||
*
|
||||
* <p>No liveness bound — the caller owns the deadline. The compactor pool is shared across
|
||||
* tables, so a checkpoint queued behind unrelated work looks exactly like one that is merging.
|
||||
*/
|
||||
public void checkpointMemWal() {
|
||||
public void checkpointRTE() {
|
||||
for (int reissue = 0; reissue <= MAX_REISSUES; reissue++) {
|
||||
// The freeze turns everything written before this call into an SSTable, so the
|
||||
// watermark has to be read after it. Idempotent: freezing an empty memtable is a
|
||||
// no-op, so a re-issue does not churn empty sstables.
|
||||
if (issueVoid(this::flushMemWal)) {
|
||||
if (issueVoid(this::flushRTE)) {
|
||||
backoff(reissue);
|
||||
continue;
|
||||
}
|
||||
|
||||
Attempt<Optional<MemWalStats>> stats = issue(() -> getMemWalStats(false));
|
||||
Attempt<Optional<RTEStats>> stats = issue(() -> getRTEStats(false));
|
||||
if (stats.lostClaim) {
|
||||
backoff(reissue);
|
||||
continue;
|
||||
}
|
||||
if (!stats.value.isPresent()) {
|
||||
// Not WAL-backed; flushMemWal would have errored first but for a race.
|
||||
// Not WAL-backed; flushRTE would have errored first but for a race.
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -235,7 +235,7 @@ public class LanceDbTableMemWal {
|
||||
backoff(reissue);
|
||||
}
|
||||
throw new IllegalStateException(
|
||||
"checkpointMemWal: the owning node kept losing its claim; re-issued from flush the maximum "
|
||||
"checkpointRTE: the owning node kept losing its claim; re-issued from flush the maximum "
|
||||
+ "number of times");
|
||||
}
|
||||
|
||||
@@ -246,7 +246,7 @@ public class LanceDbTableMemWal {
|
||||
*/
|
||||
private boolean drainToTargets(Map<String, Long> targets) {
|
||||
while (true) {
|
||||
Attempt<Optional<MemWalStats>> stats = issue(() -> getMemWalStats(false));
|
||||
Attempt<Optional<RTEStats>> stats = issue(() -> getRTEStats(false));
|
||||
if (stats.lostClaim) {
|
||||
return false;
|
||||
}
|
||||
@@ -277,7 +277,7 @@ public class LanceDbTableMemWal {
|
||||
|
||||
if (!allCompacting) {
|
||||
try {
|
||||
compactMemWal();
|
||||
compactRTE();
|
||||
} catch (LanceDbRestClient.HttpException e) {
|
||||
if (isLostClaim(e)) {
|
||||
return false;
|
||||
@@ -296,7 +296,7 @@ public class LanceDbTableMemWal {
|
||||
}
|
||||
|
||||
/** The newest SSTable generation held by each table shard, skipping table shards holding none. */
|
||||
private static Map<String, Long> newestSstableGenerations(MemWalStats stats) {
|
||||
private static Map<String, Long> newestSstableGenerations(RTEStats stats) {
|
||||
Map<String, Long> targets = new HashMap<String, Long>();
|
||||
for (TableShardStats tableShard : stats.tableShards()) {
|
||||
OptionalLong newest = tableShard.newestSstableGeneration();
|
||||
+7
-7
@@ -20,18 +20,18 @@ import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Live per-table-shard MemWAL state, as returned by {@link LanceDbTableMemWal#getMemWalStats()}.
|
||||
* Live per-table-shard RTE state, as returned by {@link LanceDbTableRTE#getRTEStats()}.
|
||||
*
|
||||
* <p>Nothing here is derived: sums and differences (total SSTable 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 MemWalStats {
|
||||
private static final String CONTEXT = "memwal stats";
|
||||
public class RTEStats {
|
||||
private static final String CONTEXT = "rte stats";
|
||||
|
||||
private final List<TableShardStats> tableShards;
|
||||
|
||||
MemWalStats(List<TableShardStats> tableShards) {
|
||||
RTEStats(List<TableShardStats> tableShards) {
|
||||
this.tableShards = Collections.unmodifiableList(tableShards);
|
||||
}
|
||||
|
||||
@@ -40,17 +40,17 @@ public class MemWalStats {
|
||||
return tableShards;
|
||||
}
|
||||
|
||||
static MemWalStats fromJson(JsonNode node) {
|
||||
static RTEStats fromJson(JsonNode node) {
|
||||
JsonFields.requiredObject(node, CONTEXT);
|
||||
List<TableShardStats> tableShards = new ArrayList<TableShardStats>();
|
||||
for (JsonNode tableShard : JsonFields.requiredArray(node, "table_shards", CONTEXT)) {
|
||||
tableShards.add(TableShardStats.fromJson(tableShard));
|
||||
}
|
||||
return new MemWalStats(tableShards);
|
||||
return new RTEStats(tableShards);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "MemWalStats{tableShards=" + tableShards + "}";
|
||||
return "RTEStats{tableShards=" + tableShards + "}";
|
||||
}
|
||||
}
|
||||
+26
-26
@@ -23,22 +23,23 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Specification selecting Lance's MemWAL LSM-style write path for {@code mergeInsert}.
|
||||
* Specification enabling the realtime engine (RTE), which routes {@code mergeInsert} through
|
||||
* Lance's MemWAL LSM-style write path.
|
||||
*
|
||||
* <p>Construct via {@link #bucket}, {@link #identity}, or {@link #unsharded}, then optionally chain
|
||||
* {@link #withMaintainedIndexes} and {@link #withWriterConfigDefaults}. Install it with {@link
|
||||
* LanceDbTableMemWal#enableMemWal} and remove it with {@link LanceDbTableMemWal#disableMemWal}.
|
||||
* LanceDbTableRTE#enableRTE} and remove it with {@link LanceDbTableRTE#disableRTE}.
|
||||
*
|
||||
* <p>This is deliberately not {@code org.lance.memwal.InitializeMemWalParams}. That type is Lance's
|
||||
* own, and its maintained-index default is the opposite of this one: it defaults to maintaining
|
||||
* <em>nothing</em>, while a fresh spec here maintains <em>every</em> index. It also cannot express
|
||||
* the null that asks the server to resolve the set.
|
||||
*/
|
||||
public class MemWalWriteSpec {
|
||||
public class RTEWriteSpec {
|
||||
|
||||
private static final String CONTEXT = "get_memwal_spec response";
|
||||
private static final String CONTEXT = "get_rte_spec response";
|
||||
|
||||
/** How writes are routed to MemWAL shards. */
|
||||
/** How writes are routed to RTE shards. */
|
||||
public enum Sharding {
|
||||
/** Hash-bucket writes by a scalar column. */
|
||||
BUCKET("bucket"),
|
||||
@@ -73,7 +74,7 @@ public class MemWalWriteSpec {
|
||||
private final List<String> maintainedIndexes;
|
||||
private final Map<String, String> writerConfigDefaults;
|
||||
|
||||
private MemWalWriteSpec(
|
||||
private RTEWriteSpec(
|
||||
Sharding sharding,
|
||||
String column,
|
||||
Integer numBuckets,
|
||||
@@ -95,11 +96,11 @@ public class MemWalWriteSpec {
|
||||
* @param column A non-nested column with a supported scalar type.
|
||||
* @param numBuckets The number of buckets, in {@code [1, 1024]}.
|
||||
*/
|
||||
public static MemWalWriteSpec bucket(String column, int numBuckets) {
|
||||
public static RTEWriteSpec bucket(String column, int numBuckets) {
|
||||
if (column == null || column.trim().isEmpty()) {
|
||||
throw new IllegalArgumentException("Column cannot be null or empty");
|
||||
}
|
||||
return new MemWalWriteSpec(
|
||||
return new RTEWriteSpec(
|
||||
Sharding.BUCKET, column, numBuckets, null, new HashMap<String, String>());
|
||||
}
|
||||
|
||||
@@ -111,29 +112,28 @@ public class MemWalWriteSpec {
|
||||
* 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 MemWalWriteSpec identity(String column) {
|
||||
public static RTEWriteSpec identity(String column) {
|
||||
if (column == null || column.trim().isEmpty()) {
|
||||
throw new IllegalArgumentException("Column cannot be null or empty");
|
||||
}
|
||||
return new MemWalWriteSpec(
|
||||
Sharding.IDENTITY, column, null, null, new HashMap<String, String>());
|
||||
return new RTEWriteSpec(Sharding.IDENTITY, column, null, null, new HashMap<String, String>());
|
||||
}
|
||||
|
||||
/** No sharding — every write goes to a single MemWAL shard — maintaining every index. */
|
||||
public static MemWalWriteSpec unsharded() {
|
||||
return new MemWalWriteSpec(Sharding.UNSHARDED, null, null, null, new HashMap<String, String>());
|
||||
/** No sharding — every write goes to a single RTE shard — maintaining every index. */
|
||||
public static RTEWriteSpec unsharded() {
|
||||
return new RTEWriteSpec(Sharding.UNSHARDED, null, null, null, new HashMap<String, String>());
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the indexes the MemWAL keeps up to date as rows are appended.
|
||||
* Set the indexes the RTE keeps up to date as rows are appended.
|
||||
*
|
||||
* <p>Pass {@code null} — the default for a fresh spec — to maintain every index the table has,
|
||||
* including ones created later. Pass an empty list to maintain none.
|
||||
*
|
||||
* <p>Note that {@code null} and the empty list mean opposite things here.
|
||||
*/
|
||||
public MemWalWriteSpec withMaintainedIndexes(List<String> maintainedIndexes) {
|
||||
return new MemWalWriteSpec(
|
||||
public RTEWriteSpec withMaintainedIndexes(List<String> maintainedIndexes) {
|
||||
return new RTEWriteSpec(
|
||||
sharding,
|
||||
column,
|
||||
numBuckets,
|
||||
@@ -150,11 +150,11 @@ public class MemWalWriteSpec {
|
||||
* max_unflushed_memtable_bytes}, and {@code enable_memtable}. Duration knobs carry an {@code _ms}
|
||||
* suffix, such as {@code max_wal_flush_interval_ms}.
|
||||
*/
|
||||
public MemWalWriteSpec withWriterConfigDefaults(Map<String, String> writerConfigDefaults) {
|
||||
public RTEWriteSpec withWriterConfigDefaults(Map<String, String> writerConfigDefaults) {
|
||||
if (writerConfigDefaults == null) {
|
||||
throw new IllegalArgumentException("writerConfigDefaults cannot be null");
|
||||
}
|
||||
return new MemWalWriteSpec(
|
||||
return new RTEWriteSpec(
|
||||
sharding,
|
||||
column,
|
||||
numBuckets,
|
||||
@@ -178,7 +178,7 @@ public class MemWalWriteSpec {
|
||||
}
|
||||
|
||||
/**
|
||||
* Which indexes the MemWAL maintains.
|
||||
* Which indexes the RTE maintains.
|
||||
*
|
||||
* <p>Null selects automatically: every supported index the table has now or gains later. An empty
|
||||
* list maintains none. A non-empty list maintains exactly those.
|
||||
@@ -192,7 +192,7 @@ public class MemWalWriteSpec {
|
||||
return Collections.unmodifiableMap(writerConfigDefaults);
|
||||
}
|
||||
|
||||
/** Render this spec as the {@code enable_memwal} request body. */
|
||||
/** Render this spec as the {@code enable_rte} request body. */
|
||||
Map<String, Object> toRequestBody() {
|
||||
Map<String, Object> shardingBody = new LinkedHashMap<String, Object>();
|
||||
shardingBody.put("mode", sharding.wireName());
|
||||
@@ -212,15 +212,15 @@ public class MemWalWriteSpec {
|
||||
}
|
||||
|
||||
/**
|
||||
* Rebuild a spec from a {@code get_memwal_spec} response body.
|
||||
* Rebuild a spec from a {@code get_rte_spec} response body.
|
||||
*
|
||||
* <p>A null {@code maintained_indexes} selects every index the table has; an empty list selects
|
||||
* none.
|
||||
*/
|
||||
static MemWalWriteSpec fromJson(JsonNode node) {
|
||||
static RTEWriteSpec fromJson(JsonNode node) {
|
||||
JsonNode shardingNode = node.get("sharding");
|
||||
if (shardingNode == null || shardingNode.get("mode") == null) {
|
||||
throw new IllegalStateException("get_memwal_spec response has no sharding mode");
|
||||
throw new IllegalStateException("get_rte_spec response has no sharding mode");
|
||||
}
|
||||
Sharding sharding = Sharding.fromWireName(shardingNode.get("mode").asText());
|
||||
|
||||
@@ -253,12 +253,12 @@ public class MemWalWriteSpec {
|
||||
.forEachRemaining(name -> defaults.put(name, defaultsNode.get(name).asText()));
|
||||
}
|
||||
|
||||
return new MemWalWriteSpec(sharding, column, numBuckets, maintainedIndexes, defaults);
|
||||
return new RTEWriteSpec(sharding, column, numBuckets, maintainedIndexes, defaults);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "MemWalWriteSpec{sharding="
|
||||
return "RTEWriteSpec{sharding="
|
||||
+ sharding
|
||||
+ ", column="
|
||||
+ column
|
||||
+129
-133
@@ -41,17 +41,17 @@ 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.
|
||||
* Unit tests for the RTE LSM routes, run against a scripted local HTTP server.
|
||||
*
|
||||
* <p>The wire assertions mirror the Rust mocked-endpoint tests in {@code
|
||||
* rust/lancedb/src/remote/table.rs}, which are the contract these routes have to match.
|
||||
*/
|
||||
public class LanceDbTableMemWalTest {
|
||||
public class LanceDbTableRTETest {
|
||||
private static final ObjectMapper MAPPER = new ObjectMapper();
|
||||
|
||||
private HttpServer server;
|
||||
private LanceDbRestClient client;
|
||||
private LanceDbTableMemWal memWal;
|
||||
private LanceDbTableRTE rte;
|
||||
|
||||
private final List<String> requestPaths = Collections.synchronizedList(new ArrayList<String>());
|
||||
private final List<String> requestBodies = Collections.synchronizedList(new ArrayList<String>());
|
||||
@@ -101,7 +101,7 @@ public class LanceDbTableMemWalTest {
|
||||
.database("test-db")
|
||||
.endpoint("http://127.0.0.1:" + server.getAddress().getPort())
|
||||
.buildRestClient();
|
||||
memWal = new LanceDbTableMemWal(client, "my_table");
|
||||
rte = new LanceDbTableRTE(client, "my_table");
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
@@ -115,12 +115,12 @@ public class LanceDbTableMemWalTest {
|
||||
// ===========================================================================
|
||||
|
||||
@Test
|
||||
public void testEnableMemWalUnsharded() throws Exception {
|
||||
enqueue("enable_memwal", 200, "");
|
||||
public void testEnableRTEUnsharded() throws Exception {
|
||||
enqueue("enable_rte", 200, "");
|
||||
|
||||
memWal.enableMemWal(MemWalWriteSpec.unsharded());
|
||||
rte.enableRTE(RTEWriteSpec.unsharded());
|
||||
|
||||
assertEquals("/v1/table/my_table/enable_memwal/", requestPaths.get(0));
|
||||
assertEquals("/v1/table/my_table/enable_rte/", requestPaths.get(0));
|
||||
JsonNode body = MAPPER.readTree(requestBodies.get(0));
|
||||
assertEquals("unsharded", body.get("sharding").get("mode").asText());
|
||||
assertFalse(body.get("sharding").has("column"));
|
||||
@@ -128,11 +128,10 @@ public class LanceDbTableMemWalTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testEnableMemWalBucket() throws Exception {
|
||||
enqueue("enable_memwal", 200, "");
|
||||
public void testEnableRTEBucket() throws Exception {
|
||||
enqueue("enable_rte", 200, "");
|
||||
|
||||
memWal.enableMemWal(
|
||||
MemWalWriteSpec.bucket("id", 16).withMaintainedIndexes(Arrays.asList("id_idx")));
|
||||
rte.enableRTE(RTEWriteSpec.bucket("id", 16).withMaintainedIndexes(Arrays.asList("id_idx")));
|
||||
|
||||
JsonNode body = MAPPER.readTree(requestBodies.get(0));
|
||||
assertEquals("bucket", body.get("sharding").get("mode").asText());
|
||||
@@ -143,10 +142,10 @@ public class LanceDbTableMemWalTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testEnableMemWalIdentity() throws Exception {
|
||||
enqueue("enable_memwal", 200, "");
|
||||
public void testEnableRTEIdentity() throws Exception {
|
||||
enqueue("enable_rte", 200, "");
|
||||
|
||||
memWal.enableMemWal(MemWalWriteSpec.identity("tenant"));
|
||||
rte.enableRTE(RTEWriteSpec.identity("tenant"));
|
||||
|
||||
JsonNode body = MAPPER.readTree(requestBodies.get(0));
|
||||
assertEquals("identity", body.get("sharding").get("mode").asText());
|
||||
@@ -160,55 +159,54 @@ public class LanceDbTableMemWalTest {
|
||||
*/
|
||||
@Test
|
||||
public void testMaintainedIndexesNullAndEmptyAreDistinctOnTheWire() throws Exception {
|
||||
enqueue("enable_memwal", 200, "");
|
||||
enqueue("enable_rte", 200, "");
|
||||
|
||||
memWal.enableMemWal(MemWalWriteSpec.unsharded());
|
||||
rte.enableRTE(RTEWriteSpec.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 []");
|
||||
|
||||
memWal.enableMemWal(
|
||||
MemWalWriteSpec.unsharded().withMaintainedIndexes(Collections.<String>emptyList()));
|
||||
rte.enableRTE(RTEWriteSpec.unsharded().withMaintainedIndexes(Collections.<String>emptyList()));
|
||||
JsonNode none = MAPPER.readTree(requestBodies.get(1));
|
||||
assertTrue(none.get("maintained_indexes").isArray());
|
||||
assertEquals(0, none.get("maintained_indexes").size());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testEnableMemWalWriterConfigDefaults() throws Exception {
|
||||
enqueue("enable_memwal", 200, "");
|
||||
public void testEnableRTEWriterConfigDefaults() throws Exception {
|
||||
enqueue("enable_rte", 200, "");
|
||||
|
||||
Map<String, String> defaults = new HashMap<String, String>();
|
||||
defaults.put("max_memtable_rows", "50000");
|
||||
memWal.enableMemWal(MemWalWriteSpec.unsharded().withWriterConfigDefaults(defaults));
|
||||
rte.enableRTE(RTEWriteSpec.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 testDisableMemWal() {
|
||||
enqueue("disable_memwal", 200, "");
|
||||
public void testDisableRTE() {
|
||||
enqueue("disable_rte", 200, "");
|
||||
|
||||
memWal.disableMemWal();
|
||||
rte.disableRTE();
|
||||
|
||||
assertEquals("/v1/table/my_table/disable_memwal/", requestPaths.get(0));
|
||||
assertEquals("/v1/table/my_table/disable_rte/", requestPaths.get(0));
|
||||
assertEquals("", requestBodies.get(0));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetMemWalSpec() {
|
||||
public void testGetRTESpec() {
|
||||
enqueue(
|
||||
"get_memwal_spec",
|
||||
"get_rte_spec",
|
||||
200,
|
||||
"{\"memwal_spec\":{\"sharding\":{\"mode\":\"bucket\",\"column\":\"id\","
|
||||
"{\"rte_spec\":{\"sharding\":{\"mode\":\"bucket\",\"column\":\"id\","
|
||||
+ "\"num_buckets\":16},\"maintained_indexes\":[\"id_idx\"],"
|
||||
+ "\"writer_config_defaults\":{\"durable_write\":\"true\"}}}");
|
||||
|
||||
Optional<MemWalWriteSpec> spec = memWal.getMemWalSpec();
|
||||
Optional<RTEWriteSpec> spec = rte.getRTESpec();
|
||||
|
||||
assertTrue(spec.isPresent());
|
||||
assertEquals(MemWalWriteSpec.Sharding.BUCKET, spec.get().sharding());
|
||||
assertEquals(RTEWriteSpec.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());
|
||||
@@ -217,28 +215,28 @@ public class LanceDbTableMemWalTest {
|
||||
|
||||
/** A null selection, every index the table has, reads back as null rather than an empty list. */
|
||||
@Test
|
||||
public void testGetMemWalSpecRoundTripsEveryIndexSelection() {
|
||||
public void testGetRTESpecRoundTripsEveryIndexSelection() {
|
||||
enqueue(
|
||||
"get_memwal_spec",
|
||||
"get_rte_spec",
|
||||
200,
|
||||
"{\"memwal_spec\":{\"sharding\":{\"mode\":\"unsharded\"},"
|
||||
"{\"rte_spec\":{\"sharding\":{\"mode\":\"unsharded\"},"
|
||||
+ "\"maintained_indexes\":null,\"writer_config_defaults\":{}}}");
|
||||
|
||||
Optional<MemWalWriteSpec> spec = memWal.getMemWalSpec();
|
||||
Optional<RTEWriteSpec> spec = rte.getRTESpec();
|
||||
|
||||
assertTrue(spec.isPresent());
|
||||
assertNull(spec.get().maintainedIndexes());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetMemWalSpecRoundTripsNoIndexSelection() {
|
||||
public void testGetRTESpecRoundTripsNoIndexSelection() {
|
||||
enqueue(
|
||||
"get_memwal_spec",
|
||||
"get_rte_spec",
|
||||
200,
|
||||
"{\"memwal_spec\":{\"sharding\":{\"mode\":\"unsharded\"},"
|
||||
"{\"rte_spec\":{\"sharding\":{\"mode\":\"unsharded\"},"
|
||||
+ "\"maintained_indexes\":[],\"writer_config_defaults\":{}}}");
|
||||
|
||||
Optional<MemWalWriteSpec> spec = memWal.getMemWalSpec();
|
||||
Optional<RTEWriteSpec> spec = rte.getRTESpec();
|
||||
|
||||
assertTrue(spec.isPresent());
|
||||
assertEquals(Collections.emptyList(), spec.get().maintainedIndexes());
|
||||
@@ -249,32 +247,32 @@ public class LanceDbTableMemWalTest {
|
||||
* rejected: reading a string or an object as null would turn a malformed response into a policy.
|
||||
*/
|
||||
@Test
|
||||
public void testGetMemWalSpecRejectsANonArrayMaintainedIndexes() {
|
||||
public void testGetRTESpecRejectsANonArrayMaintainedIndexes() {
|
||||
enqueue(
|
||||
"get_memwal_spec",
|
||||
"get_rte_spec",
|
||||
200,
|
||||
"{\"memwal_spec\":{\"sharding\":{\"mode\":\"unsharded\"},"
|
||||
"{\"rte_spec\":{\"sharding\":{\"mode\":\"unsharded\"},"
|
||||
+ "\"maintained_indexes\":\"id_idx\",\"writer_config_defaults\":{}}}");
|
||||
|
||||
assertThrows(IllegalStateException.class, () -> memWal.getMemWalSpec());
|
||||
assertThrows(IllegalStateException.class, () -> rte.getRTESpec());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetMemWalSpecRejectsNonStringIndexNames() {
|
||||
public void testGetRTESpecRejectsNonStringIndexNames() {
|
||||
enqueue(
|
||||
"get_memwal_spec",
|
||||
"get_rte_spec",
|
||||
200,
|
||||
"{\"memwal_spec\":{\"sharding\":{\"mode\":\"unsharded\"},"
|
||||
"{\"rte_spec\":{\"sharding\":{\"mode\":\"unsharded\"},"
|
||||
+ "\"maintained_indexes\":[7],\"writer_config_defaults\":{}}}");
|
||||
|
||||
assertThrows(IllegalStateException.class, () -> memWal.getMemWalSpec());
|
||||
assertThrows(IllegalStateException.class, () -> rte.getRTESpec());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetMemWalSpecAbsent() {
|
||||
enqueue("get_memwal_spec", 200, "{\"memwal_spec\":null}");
|
||||
public void testGetRTESpecAbsent() {
|
||||
enqueue("get_rte_spec", 200, "{\"rte_spec\":null}");
|
||||
|
||||
assertFalse(memWal.getMemWalSpec().isPresent());
|
||||
assertFalse(rte.getRTESpec().isPresent());
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
@@ -282,12 +280,12 @@ public class LanceDbTableMemWalTest {
|
||||
// ===========================================================================
|
||||
|
||||
@Test
|
||||
public void testGetMemWalStats() throws Exception {
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
|
||||
public void testGetRTEStats() throws Exception {
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
|
||||
|
||||
Optional<MemWalStats> got = memWal.getMemWalStats(true);
|
||||
Optional<RTEStats> got = rte.getRTEStats(true);
|
||||
|
||||
assertEquals("/v1/table/my_table/get_memwal_stats/", requestPaths.get(0));
|
||||
assertEquals("/v1/table/my_table/get_rte_stats/", requestPaths.get(0));
|
||||
assertTrue(MAPPER.readTree(requestBodies.get(0)).get("include_sstable_rows").asBoolean());
|
||||
assertTrue(got.isPresent());
|
||||
TableShardStats decoded = got.get().tableShards().get(0);
|
||||
@@ -305,18 +303,18 @@ public class LanceDbTableMemWalTest {
|
||||
|
||||
/** The optional fields decode when the server does send them. */
|
||||
@Test
|
||||
public void testGetMemWalStatsDecodesOptionalFields() {
|
||||
public void testGetRTEStatsDecodesOptionalFields() {
|
||||
enqueue(
|
||||
"get_memwal_stats",
|
||||
"get_rte_stats",
|
||||
200,
|
||||
"{\"memwal_stats\":{\"table_shards\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
|
||||
"{\"rte_stats\":{\"table_shards\":[{\"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,"
|
||||
+ "\"sstables\":[{\"generation\":7,\"bytes\":1024,\"rows\":42}],"
|
||||
+ "\"compacting\":true,\"memtables\":[{\"generation\":8,\"rows\":5,"
|
||||
+ "\"bytes\":64,\"batches\":2,\"indexes\":[\"id_idx\"]}]}]}}");
|
||||
|
||||
TableShardStats decoded = memWal.getMemWalStats(true).get().tableShards().get(0);
|
||||
TableShardStats decoded = rte.getRTEStats(true).get().tableShards().get(0);
|
||||
|
||||
assertEquals(3, decoded.replayAfterWalEntryPosition());
|
||||
assertEquals(11, decoded.walEntryPositionLastSeen());
|
||||
@@ -332,17 +330,17 @@ public class LanceDbTableMemWalTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetMemWalStatsAbsentWhenMemWalDisabled() {
|
||||
enqueue("get_memwal_stats", 200, "{\"memwal_stats\":null}");
|
||||
public void testGetRTEStatsAbsentWhenRTEDisabled() {
|
||||
enqueue("get_rte_stats", 200, "{\"rte_stats\":null}");
|
||||
|
||||
assertFalse(memWal.getMemWalStats().isPresent());
|
||||
assertFalse(rte.getRTEStats().isPresent());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetMemWalStatsDefaultsToExcludingGenerationRows() throws Exception {
|
||||
enqueue("get_memwal_stats", 200, stats());
|
||||
public void testGetRTEStatsDefaultsToExcludingGenerationRows() throws Exception {
|
||||
enqueue("get_rte_stats", 200, stats());
|
||||
|
||||
memWal.getMemWalStats();
|
||||
rte.getRTEStats();
|
||||
|
||||
assertFalse(MAPPER.readTree(requestBodies.get(0)).get("include_sstable_rows").asBoolean());
|
||||
}
|
||||
@@ -353,22 +351,22 @@ public class LanceDbTableMemWalTest {
|
||||
|
||||
@Test
|
||||
public void testFlushAndCompactRoutes() {
|
||||
enqueue("flush_memwal", 200, "");
|
||||
enqueue("compact_memwal", 200, "");
|
||||
enqueue("flush_rte", 200, "");
|
||||
enqueue("compact_rte", 200, "");
|
||||
|
||||
memWal.flushMemWal();
|
||||
memWal.compactMemWal();
|
||||
rte.flushRTE();
|
||||
rte.compactRTE();
|
||||
|
||||
assertEquals("/v1/table/my_table/flush_memwal/", requestPaths.get(0));
|
||||
assertEquals("/v1/table/my_table/compact_memwal/", requestPaths.get(1));
|
||||
assertEquals("/v1/table/my_table/flush_rte/", requestPaths.get(0));
|
||||
assertEquals("/v1/table/my_table/compact_rte/", requestPaths.get(1));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHttpErrorCarriesStatus() {
|
||||
enqueue("flush_memwal", 404, "no such table");
|
||||
enqueue("flush_rte", 404, "no such table");
|
||||
|
||||
LanceDbRestClient.HttpException e =
|
||||
assertThrows(LanceDbRestClient.HttpException.class, () -> memWal.flushMemWal());
|
||||
assertThrows(LanceDbRestClient.HttpException.class, () -> rte.flushRTE());
|
||||
assertEquals(404, e.statusCode());
|
||||
}
|
||||
|
||||
@@ -377,98 +375,97 @@ public class LanceDbTableMemWalTest {
|
||||
// ===========================================================================
|
||||
|
||||
@Test
|
||||
public void testCheckpointReturnsWhenMemWalDisabled() {
|
||||
enqueue("flush_memwal", 200, "");
|
||||
enqueue("get_memwal_stats", 200, "{\"memwal_stats\":null}");
|
||||
public void testCheckpointReturnsWhenRTEDisabled() {
|
||||
enqueue("flush_rte", 200, "");
|
||||
enqueue("get_rte_stats", 200, "{\"rte_stats\":null}");
|
||||
|
||||
memWal.checkpointMemWal();
|
||||
rte.checkpointRTE();
|
||||
|
||||
assertEquals(0, countCalls("compact_memwal"), "nothing to compact when the LSM path is off");
|
||||
assertEquals(0, countCalls("compact_rte"), "nothing to compact when the LSM path is off");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCheckpointReturnsWhenNoGenerationsOutstanding() {
|
||||
enqueue("flush_memwal", 200, "");
|
||||
enqueue("flush_rte", 200, "");
|
||||
// A table shard with no SSTables yields no target, so the drain never starts.
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", false)));
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", false)));
|
||||
|
||||
memWal.checkpointMemWal();
|
||||
rte.checkpointRTE();
|
||||
|
||||
assertEquals(0, countCalls("compact_memwal"));
|
||||
assertEquals(0, countCalls("compact_rte"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCheckpointConvergesOnceTargetGenerationsAreGone() {
|
||||
enqueue("flush_memwal", 200, "");
|
||||
enqueue("flush_rte", 200, "");
|
||||
// Watermark read: shard-0 holds sstables 7 and 8, so target = 8.
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
|
||||
// First drain poll: both still outstanding, nothing compacting -> dispatch a pass.
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", false, 7L, 8L)));
|
||||
// Second drain poll: drained past the target -> done.
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", false, 9L)));
|
||||
enqueue("compact_memwal", 200, "");
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", false, 9L)));
|
||||
enqueue("compact_rte", 200, "");
|
||||
|
||||
memWal.checkpointMemWal();
|
||||
rte.checkpointRTE();
|
||||
|
||||
assertEquals(1, countCalls("compact_memwal"), "one pass dispatched");
|
||||
assertEquals(3, countCalls("get_memwal_stats"), "watermark read plus two drain polls");
|
||||
assertEquals(1, countCalls("compact_rte"), "one pass dispatched");
|
||||
assertEquals(3, countCalls("get_rte_stats"), "watermark read plus two drain polls");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCheckpointDoesNotPileOnWhileEveryTargetBucketIsCompacting() {
|
||||
enqueue("flush_memwal", 200, "");
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", true, 4L)));
|
||||
enqueue("flush_rte", 200, "");
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", true, 4L)));
|
||||
// Still compacting on the first poll, so no pass is dispatched; then it drains.
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", true, 4L)));
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", false, 5L)));
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", true, 4L)));
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", false, 5L)));
|
||||
|
||||
memWal.checkpointMemWal();
|
||||
rte.checkpointRTE();
|
||||
|
||||
assertEquals(0, countCalls("compact_memwal"), "a latched table shard is left alone");
|
||||
assertEquals(0, countCalls("compact_rte"), "a latched table shard 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_memwal", 200, "");
|
||||
enqueue("get_memwal_stats", 421, "no claim");
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", false)));
|
||||
enqueue("flush_rte", 200, "");
|
||||
enqueue("get_rte_stats", 421, "no claim");
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", false)));
|
||||
|
||||
memWal.checkpointMemWal();
|
||||
rte.checkpointRTE();
|
||||
|
||||
assertEquals(2, countCalls("flush_memwal"), "re-issued from flush");
|
||||
assertEquals(2, countCalls("flush_rte"), "re-issued from flush");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCheckpointRetriesRetryableStatusInPlace() {
|
||||
enqueue("flush_memwal", 429, "latch held");
|
||||
enqueue("flush_memwal", 200, "");
|
||||
enqueue("get_memwal_stats", 200, stats(tableShard("shard-0", false)));
|
||||
enqueue("flush_rte", 429, "latch held");
|
||||
enqueue("flush_rte", 200, "");
|
||||
enqueue("get_rte_stats", 200, stats(tableShard("shard-0", false)));
|
||||
|
||||
memWal.checkpointMemWal();
|
||||
rte.checkpointRTE();
|
||||
|
||||
assertEquals(2, countCalls("flush_memwal"), "429 retried in place, not re-issued");
|
||||
assertEquals(2, countCalls("flush_rte"), "429 retried in place, not re-issued");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCheckpointPropagatesTerminalStatus() {
|
||||
enqueue("flush_memwal", 400, "bad request");
|
||||
enqueue("flush_rte", 400, "bad request");
|
||||
|
||||
LanceDbRestClient.HttpException e =
|
||||
assertThrows(LanceDbRestClient.HttpException.class, () -> memWal.checkpointMemWal());
|
||||
assertThrows(LanceDbRestClient.HttpException.class, () -> rte.checkpointRTE());
|
||||
assertEquals(400, e.statusCode());
|
||||
assertEquals(1, countCalls("flush_memwal"), "a terminal status is not retried");
|
||||
assertEquals(1, countCalls("flush_rte"), "a terminal status is not retried");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCheckpointGivesUpAfterRepeatedLostClaims() {
|
||||
enqueue("flush_memwal", 421, "no claim");
|
||||
enqueue("flush_rte", 421, "no claim");
|
||||
|
||||
IllegalStateException e =
|
||||
assertThrows(IllegalStateException.class, () -> memWal.checkpointMemWal());
|
||||
IllegalStateException e = assertThrows(IllegalStateException.class, () -> rte.checkpointRTE());
|
||||
assertTrue(e.getMessage().contains("kept losing its claim"), e.getMessage());
|
||||
assertEquals(4, countCalls("flush_memwal"), "the initial attempt plus MAX_REISSUES");
|
||||
assertEquals(4, countCalls("flush_rte"), "the initial attempt plus MAX_REISSUES");
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
@@ -478,24 +475,24 @@ public class LanceDbTableMemWalTest {
|
||||
/**
|
||||
* A stats payload that does not decode must fail closed. Every one of these bodies used to be
|
||||
* read as "no table shards", which is indistinguishable from a drained table, so {@code
|
||||
* checkpointMemWal} reported convergence for a checkpoint that never ran.
|
||||
* checkpointRTE} reported convergence for a checkpoint that never ran.
|
||||
*/
|
||||
@Test
|
||||
public void testCheckpointRejectsMalformedStats() {
|
||||
Map<String, String> malformed = new LinkedHashMap<String, String>();
|
||||
malformed.put("no response body at all", "");
|
||||
malformed.put("stats object with no table shards", "{\"memwal_stats\":{}}");
|
||||
malformed.put("stats object with no table shards", "{\"rte_stats\":{}}");
|
||||
malformed.put(
|
||||
"table shard missing its required fields", "{\"memwal_stats\":{\"table_shards\":[{}]}}");
|
||||
"table shard missing its required fields", "{\"rte_stats\":{\"table_shards\":[{}]}}");
|
||||
malformed.put(
|
||||
"table shard missing sstables",
|
||||
"{\"memwal_stats\":{\"table_shards\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
|
||||
"{\"rte_stats\":{\"table_shards\":[{\"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",
|
||||
"{\"memwal_stats\":{\"table_shards\":[{\"shard_id\":\"shard-0\",\"status\":\"Active\","
|
||||
"{\"rte_stats\":{\"table_shards\":[{\"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,"
|
||||
+ "\"sstables\":[{\"generation\":\"7\",\"bytes\":1024}],"
|
||||
@@ -503,25 +500,25 @@ public class LanceDbTableMemWalTest {
|
||||
|
||||
for (Map.Entry<String, String> each : malformed.entrySet()) {
|
||||
setUpFresh();
|
||||
enqueue("flush_memwal", 200, "");
|
||||
enqueue("get_memwal_stats", 200, each.getValue());
|
||||
enqueue("flush_rte", 200, "");
|
||||
enqueue("get_rte_stats", 200, each.getValue());
|
||||
|
||||
assertThrows(
|
||||
IllegalStateException.class,
|
||||
() -> memWal.checkpointMemWal(),
|
||||
() -> rte.checkpointRTE(),
|
||||
each.getKey() + " must not report convergence");
|
||||
}
|
||||
}
|
||||
|
||||
/** The one shape that legitimately means "this table has no MemWAL write path". */
|
||||
/** The one shape that legitimately means "this table has no RTE write path". */
|
||||
@Test
|
||||
public void testCheckpointTreatsNullStatsAsNotWalBacked() {
|
||||
enqueue("flush_memwal", 200, "");
|
||||
enqueue("get_memwal_stats", 200, "{\"memwal_stats\":null}");
|
||||
enqueue("flush_rte", 200, "");
|
||||
enqueue("get_rte_stats", 200, "{\"rte_stats\":null}");
|
||||
|
||||
memWal.checkpointMemWal();
|
||||
rte.checkpointRTE();
|
||||
|
||||
assertEquals(1, countCalls("get_memwal_stats"));
|
||||
assertEquals(1, countCalls("get_rte_stats"));
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
@@ -531,19 +528,18 @@ public class LanceDbTableMemWalTest {
|
||||
/**
|
||||
* 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_memwal} in place, where the loop is
|
||||
* doubled every budget here and also retried {@code compact_rte} in place, where the loop is
|
||||
* built to fall through to a fresh stats poll instead.
|
||||
*/
|
||||
@Test
|
||||
public void testCheckpointRetryBudgetIsNotDoubledByTheTransport() {
|
||||
enqueue("flush_memwal", 429, "latch held");
|
||||
enqueue("flush_rte", 429, "latch held");
|
||||
|
||||
LanceDbRestClient.HttpException e =
|
||||
assertThrows(LanceDbRestClient.HttpException.class, () -> memWal.checkpointMemWal());
|
||||
assertThrows(LanceDbRestClient.HttpException.class, () -> rte.checkpointRTE());
|
||||
|
||||
assertEquals(429, e.statusCode(), "the exhausted budget propagates the last error as itself");
|
||||
assertEquals(
|
||||
9, countCalls("flush_memwal"), "the initial request plus MAX_RETRIES, and no more");
|
||||
assertEquals(9, countCalls("flush_rte"), "the initial request plus MAX_RETRIES, and no more");
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
@@ -558,9 +554,9 @@ public class LanceDbTableMemWalTest {
|
||||
return numbers;
|
||||
}
|
||||
|
||||
/** Build a {@code memwal_stats} response body from table shard fragments. */
|
||||
/** Build a {@code rte_stats} response body from table shard fragments. */
|
||||
private static String stats(String... tableShards) {
|
||||
return "{\"memwal_stats\":{\"table_shards\":[" + String.join(",", tableShards) + "]}}";
|
||||
return "{\"rte_stats\":{\"table_shards\":[" + String.join(",", tableShards) + "]}}";
|
||||
}
|
||||
|
||||
private static String tableShard(String shardId, boolean compacting, Long... sstables) {
|
||||
@@ -600,7 +596,7 @@ public class LanceDbTableMemWalTest {
|
||||
return requestPaths.stream().filter(path -> operationOf(path).equals(operation)).count();
|
||||
}
|
||||
|
||||
/** {@code /v1/table/my_table/flush_memwal/} -> {@code flush_memwal}. */
|
||||
/** {@code /v1/table/my_table/flush_rte/} -> {@code flush_rte}. */
|
||||
private static String operationOf(String path) {
|
||||
String[] segments = path.split("/");
|
||||
return segments.length == 0 ? "" : segments[segments.length - 1];
|
||||
Reference in New Issue
Block a user