mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-22 03:32:28 +00:00
Merge branch 'merge-insert-source-stats' into merge-insert-source-stats-bindings
This commit is contained in:
@@ -642,11 +642,11 @@ describe.each(arrowVersions)(
|
||||
);
|
||||
});
|
||||
|
||||
it("should expose useMemWal on takeRowIds as the base-only escape hatch", async () => {
|
||||
it("should expose useRTE on takeRowIds as the base-only escape hatch", async () => {
|
||||
await table.add([{ id: 1 }, { id: 2 }, { id: 3 }]);
|
||||
// useMemWal(false) is reachable on TakeQuery (the escape hatch for MemWAL tables,
|
||||
// useRTE(false) is reachable on TakeQuery (the escape hatch for RTE tables,
|
||||
// where take-by-row-id auto-routes to the LSM scanner and is rejected).
|
||||
const res = await table.takeRowIds([0, 2]).useMemWal(false).toArray();
|
||||
const res = await table.takeRowIds([0, 2]).useRTE(false).toArray();
|
||||
expect(res.map((r) => r.id)).toEqual([1, 3]);
|
||||
});
|
||||
|
||||
@@ -5008,7 +5008,7 @@ describe("setUnenforcedPrimaryKey", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("enableMemWal / disableMemWal", () => {
|
||||
describe("enableRTE / disableRTE", () => {
|
||||
let tmpDir: tmp.DirResult;
|
||||
|
||||
beforeEach(() => {
|
||||
@@ -5028,16 +5028,16 @@ describe("enableMemWal / disableMemWal", () => {
|
||||
const table = await makeTable(conn);
|
||||
|
||||
await table.setUnenforcedPrimaryKey("id");
|
||||
await table.enableMemWal({
|
||||
await table.enableRTE({
|
||||
specType: "bucket",
|
||||
column: "id",
|
||||
numBuckets: 4,
|
||||
});
|
||||
await table.disableMemWal();
|
||||
await table.disableRTE();
|
||||
// A second unset errors — there is no spec left to remove.
|
||||
await expect(table.disableMemWal()).rejects.toThrow();
|
||||
await expect(table.disableRTE()).rejects.toThrow();
|
||||
// A fresh spec can be installed after unset.
|
||||
await table.enableMemWal({
|
||||
await table.enableRTE({
|
||||
specType: "bucket",
|
||||
column: "id",
|
||||
numBuckets: 8,
|
||||
@@ -5049,8 +5049,8 @@ describe("enableMemWal / disableMemWal", () => {
|
||||
const table = await makeTable(conn);
|
||||
|
||||
await table.setUnenforcedPrimaryKey("id");
|
||||
await table.enableMemWal({ specType: "unsharded" });
|
||||
await table.disableMemWal();
|
||||
await table.enableRTE({ specType: "unsharded" });
|
||||
await table.disableRTE();
|
||||
});
|
||||
|
||||
it("installs an identity spec", async () => {
|
||||
@@ -5058,8 +5058,8 @@ describe("enableMemWal / disableMemWal", () => {
|
||||
const table = await makeTable(conn);
|
||||
|
||||
await table.setUnenforcedPrimaryKey("id");
|
||||
await table.enableMemWal({ specType: "identity", column: "id" });
|
||||
await table.disableMemWal();
|
||||
await table.enableRTE({ specType: "identity", column: "id" });
|
||||
await table.disableRTE();
|
||||
});
|
||||
|
||||
it("rejects an invalid spec", async () => {
|
||||
@@ -5069,7 +5069,7 @@ describe("enableMemWal / disableMemWal", () => {
|
||||
await table.setUnenforcedPrimaryKey("id");
|
||||
// num_buckets out of range.
|
||||
await expect(
|
||||
table.enableMemWal({
|
||||
table.enableRTE({
|
||||
specType: "bucket",
|
||||
column: "id",
|
||||
numBuckets: 0,
|
||||
@@ -5077,7 +5077,7 @@ describe("enableMemWal / disableMemWal", () => {
|
||||
).rejects.toThrow();
|
||||
// Column mismatch.
|
||||
await expect(
|
||||
table.enableMemWal({
|
||||
table.enableRTE({
|
||||
specType: "bucket",
|
||||
column: "missing",
|
||||
numBuckets: 4,
|
||||
@@ -5085,13 +5085,13 @@ describe("enableMemWal / disableMemWal", () => {
|
||||
).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("reads back the installed spec via getMemWalSpec", async () => {
|
||||
it("reads back the installed spec via getRTESpec", async () => {
|
||||
const conn = await connect(tmpDir.name);
|
||||
const table = await makeTable(conn);
|
||||
await table.setUnenforcedPrimaryKey("id");
|
||||
|
||||
// Nothing installed yet.
|
||||
expect(await table.getMemWalSpec()).toBeUndefined();
|
||||
expect(await table.getRTESpec()).toBeUndefined();
|
||||
|
||||
// A real scalar index is needed to name it as a maintained index.
|
||||
await table.add([{ id: 1 }, { id: 2 }, { id: 3 }]);
|
||||
@@ -5102,14 +5102,14 @@ describe("enableMemWal / disableMemWal", () => {
|
||||
// defaults. Lance writer-config keys are canonically snake_case.
|
||||
// biome-ignore lint/style/useNamingConvention: Lance writer-config keys are snake_case
|
||||
const writerConfigDefaults = { durable_write: "false" };
|
||||
await table.enableMemWal({
|
||||
await table.enableRTE({
|
||||
specType: "bucket",
|
||||
column: "id",
|
||||
numBuckets: 4,
|
||||
maintainedIndexes: [indexName],
|
||||
writerConfigDefaults,
|
||||
});
|
||||
const spec = await table.getMemWalSpec();
|
||||
const spec = await table.getRTESpec();
|
||||
expect(spec).toBeDefined();
|
||||
expect(spec?.specType).toBe("bucket");
|
||||
expect(spec?.column).toBe("id");
|
||||
@@ -5118,19 +5118,19 @@ describe("enableMemWal / disableMemWal", () => {
|
||||
expect(spec?.writerConfigDefaults).toEqual(writerConfigDefaults);
|
||||
|
||||
// After unset, undefined again.
|
||||
await table.disableMemWal();
|
||||
expect(await table.getMemWalSpec()).toBeUndefined();
|
||||
await table.disableRTE();
|
||||
expect(await table.getRTESpec()).toBeUndefined();
|
||||
|
||||
// Identity round-trips (column recovered from the schema).
|
||||
await table.enableMemWal({ specType: "identity", column: "id" });
|
||||
const identity = await table.getMemWalSpec();
|
||||
await table.enableRTE({ specType: "identity", column: "id" });
|
||||
const identity = await table.getRTESpec();
|
||||
expect(identity?.specType).toBe("identity");
|
||||
expect(identity?.column).toBe("id");
|
||||
await table.disableMemWal();
|
||||
await table.disableRTE();
|
||||
|
||||
// Unsharded round-trips (no routing column).
|
||||
await table.enableMemWal({ specType: "unsharded" });
|
||||
const unsharded = await table.getMemWalSpec();
|
||||
await table.enableRTE({ specType: "unsharded" });
|
||||
const unsharded = await table.getRTESpec();
|
||||
expect(unsharded?.specType).toBe("unsharded");
|
||||
expect(unsharded?.column).toBeFalsy();
|
||||
});
|
||||
@@ -5159,7 +5159,7 @@ describe("LSM merge insert", () => {
|
||||
]);
|
||||
await table.setUnenforcedPrimaryKey("id");
|
||||
// numBuckets = 1: every row routes to the single bucket.
|
||||
await table.enableMemWal({
|
||||
await table.enableRTE({
|
||||
specType: "bucket",
|
||||
column: "id",
|
||||
numBuckets: 1,
|
||||
@@ -5179,22 +5179,22 @@ describe("LSM merge insert", () => {
|
||||
{ id: "c", value: 3 },
|
||||
{ id: "d", value: 4 },
|
||||
]);
|
||||
// LSM path: rows go to the MemWAL, so only numRows is populated.
|
||||
// LSM path: rows go to the RTE, so only numRows is populated.
|
||||
expect(res.numRows).toBe(2);
|
||||
expect(res.version).toBe(0);
|
||||
expect(res.numInsertedRows).toBe(0);
|
||||
|
||||
await table.closeMemWalWriters();
|
||||
await table.closeRTEWriters();
|
||||
});
|
||||
|
||||
it("falls back to the standard path with useMemWal(false)", async () => {
|
||||
it("falls back to the standard path with useRTE(false)", async () => {
|
||||
const conn = await connect(tmpDir.name);
|
||||
const table = await bucketTable(conn);
|
||||
|
||||
const res = await table
|
||||
.mergeInsert("id")
|
||||
.whenNotMatchedInsertAll()
|
||||
.useMemWal(false)
|
||||
.useRTE(false)
|
||||
.execute([
|
||||
{ id: "b", value: 9 },
|
||||
{ id: "e", value: 5 },
|
||||
@@ -5217,7 +5217,7 @@ describe("LSM merge insert", () => {
|
||||
expect(res.numRows).toBe(1);
|
||||
});
|
||||
|
||||
it("rejects a non-upsert merge under a MemWAL spec", async () => {
|
||||
it("rejects a non-upsert merge under an RTE spec", async () => {
|
||||
const conn = await connect(tmpDir.name);
|
||||
const table = await bucketTable(conn);
|
||||
|
||||
@@ -5229,7 +5229,7 @@ describe("LSM merge insert", () => {
|
||||
).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("auto-routes reads through the MemWAL scanner", async () => {
|
||||
it("auto-routes reads through the RTE scanner", async () => {
|
||||
const conn = await connect(tmpDir.name);
|
||||
const table = await bucketTable(conn); // base ids "a", "b"
|
||||
|
||||
@@ -5243,24 +5243,22 @@ describe("LSM merge insert", () => {
|
||||
const lsm = await table.query().toArray();
|
||||
expect(lsm.map((r) => r.id).sort()).toEqual(["a", "b", "c"]);
|
||||
|
||||
// useMemWal(false) bypasses the MemWAL and reads the base table only.
|
||||
const baseOnly = await table.query().useMemWal(false).toArray();
|
||||
// useRTE(false) bypasses the RTE and reads the base table only.
|
||||
const baseOnly = await table.query().useRTE(false).toArray();
|
||||
expect(baseOnly.map((r) => r.id).sort()).toEqual(["a", "b"]);
|
||||
});
|
||||
|
||||
it("reads the base table when no MemWAL spec is installed", async () => {
|
||||
it("reads the base table when no RTE spec is installed", async () => {
|
||||
const conn = await connect(tmpDir.name);
|
||||
const table = await conn.createEmptyTable(
|
||||
"plain",
|
||||
new arrow.Schema([new arrow.Field("id", new arrow.Utf8(), false)]),
|
||||
);
|
||||
// No spec: default read and useMemWal(false) both succeed against the base table.
|
||||
// No spec: default read and useRTE(false) both succeed against the base table.
|
||||
await expect(table.query().toArray()).resolves.toBeDefined();
|
||||
await expect(
|
||||
table.query().useMemWal(false).toArray(),
|
||||
).resolves.toBeDefined();
|
||||
// useMemWal(true) demands MemWAL routing; without a spec it errors.
|
||||
await expect(table.query().useMemWal(true).toArray()).rejects.toThrow();
|
||||
await expect(table.query().useRTE(false).toArray()).resolves.toBeDefined();
|
||||
// useRTE(true) demands RTE routing; without a spec it errors.
|
||||
await expect(table.query().useRTE(true).toArray()).rejects.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -5278,42 +5276,42 @@ describe("LSM convergence and stats", () => {
|
||||
new arrow.Schema([new arrow.Field("id", new arrow.Utf8(), false)]),
|
||||
);
|
||||
await table.setUnenforcedPrimaryKey("id");
|
||||
await table.enableMemWal({ specType: "unsharded" });
|
||||
await table.enableRTE({ specType: "unsharded" });
|
||||
return table;
|
||||
}
|
||||
|
||||
// These four route through the server that owns the MemWAL, so a local table
|
||||
// These four route through the server that owns the RTE, 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 flushMemWal on a local table", async () => {
|
||||
it("rejects flushRTE on a local table", async () => {
|
||||
const conn = await connect(tmpDir.name);
|
||||
const table = await lsmTable(conn);
|
||||
|
||||
await expect(table.flushMemWal()).rejects.toThrow(/not supported/i);
|
||||
await expect(table.flushRTE()).rejects.toThrow(/not supported/i);
|
||||
});
|
||||
|
||||
it("rejects compactMemWal on a local table", async () => {
|
||||
it("rejects compactRTE on a local table", async () => {
|
||||
const conn = await connect(tmpDir.name);
|
||||
const table = await lsmTable(conn);
|
||||
|
||||
await expect(table.compactMemWal()).rejects.toThrow(/not supported/i);
|
||||
await expect(table.compactRTE()).rejects.toThrow(/not supported/i);
|
||||
});
|
||||
|
||||
it("rejects getMemWalStats on a local table", async () => {
|
||||
it("rejects getRTEStats on a local table", async () => {
|
||||
const conn = await connect(tmpDir.name);
|
||||
const table = await lsmTable(conn);
|
||||
|
||||
await expect(table.getMemWalStats()).rejects.toThrow(/not supported/i);
|
||||
await expect(table.getMemWalStats(true)).rejects.toThrow(/not supported/i);
|
||||
await expect(table.getRTEStats()).rejects.toThrow(/not supported/i);
|
||||
await expect(table.getRTEStats(true)).rejects.toThrow(/not supported/i);
|
||||
});
|
||||
|
||||
it("rejects checkpointMemWal on a local table", async () => {
|
||||
it("rejects checkpointRTE on a local table", async () => {
|
||||
const conn = await connect(tmpDir.name);
|
||||
const table = await lsmTable(conn);
|
||||
|
||||
// checkpointMemWal freezes first, so it surfaces flushMemWal's rejection.
|
||||
await expect(table.checkpointMemWal()).rejects.toThrow(/not supported/i);
|
||||
// checkpointRTE freezes first, so it surfaces flushRTE's rejection.
|
||||
await expect(table.checkpointRTE()).rejects.toThrow(/not supported/i);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -168,8 +168,8 @@ export {
|
||||
WriteProgress,
|
||||
FtsToken,
|
||||
TokenizeTableOptions,
|
||||
MemWalWriteSpec,
|
||||
MemWalStats,
|
||||
RTEWriteSpec,
|
||||
RTEStats,
|
||||
TableShardStats,
|
||||
SsTableStats,
|
||||
MemtableStats,
|
||||
|
||||
@@ -88,27 +88,27 @@ export class MergeInsertBuilder {
|
||||
);
|
||||
}
|
||||
/**
|
||||
* Control MemWAL routing for this merge.
|
||||
* Control RTE routing for this merge.
|
||||
*
|
||||
* By default (unset), a `mergeInsert` on a table with a MemWAL write spec is
|
||||
* By default (unset), a `mergeInsert` on a table with an RTE write spec is
|
||||
* routed through Lance's MemWAL shard writer, and a table without one uses the
|
||||
* standard path.
|
||||
*
|
||||
* @param enable - `true` forces MemWAL routing and errors if the table has no
|
||||
* MemWAL write spec. `false` forces the standard write path even when a spec is set.
|
||||
* @param enable - `true` forces RTE routing and errors if the table has no
|
||||
* RTE write spec. `false` forces the standard write path even when a spec is set.
|
||||
*/
|
||||
useMemWal(enable: boolean): MergeInsertBuilder {
|
||||
return new MergeInsertBuilder(this.#native.useMemWal(enable), this.#schema);
|
||||
useRTE(enable: boolean): MergeInsertBuilder {
|
||||
return new MergeInsertBuilder(this.#native.useRTE(enable), this.#schema);
|
||||
}
|
||||
/**
|
||||
* Controls how an LSM merge checks that its input targets a single shard.
|
||||
*
|
||||
* When a table has a MemWAL write spec, every row in a `mergeInsert` call must
|
||||
* When a table has an RTE write spec, every row in a `mergeInsert` call must
|
||||
* route to the same shard. When `true` (the default), every row is inspected
|
||||
* to verify this. When `false`, only the first row is inspected and the
|
||||
* shard it routes to is used for the whole input — a faster path for callers
|
||||
* that have already pre-sharded their input. Has no effect on tables without
|
||||
* a MemWAL write spec.
|
||||
* an RTE write spec.
|
||||
*
|
||||
* @param validateSingleShard - Whether to check every row routes to one shard. Defaults to `true`.
|
||||
*/
|
||||
|
||||
+15
-15
@@ -677,26 +677,26 @@ export class StandardQueryBase<
|
||||
}
|
||||
|
||||
/**
|
||||
* Control MemWAL read routing for this query.
|
||||
* Control RTE read routing for this query.
|
||||
*
|
||||
* By default (unset), when the table carries a MemWAL write spec (see
|
||||
* {@link Table#enableMemWal}), reads are routed through the LSM scanner so
|
||||
* By default (unset), when the table carries an RTE write spec (see
|
||||
* {@link Table#enableRTE}), reads are routed through the LSM scanner so
|
||||
* they also return data written via the `mergeInsert` LSM path that has not yet
|
||||
* been compacted into the base table (the active/frozen in-memory memtables and
|
||||
* the SSTables), deduplicated by primary key; a table without a spec
|
||||
* reads the base table.
|
||||
*
|
||||
* @param enable - `true` forces the LSM scanner and errors if the table has no
|
||||
* MemWAL write spec. `false` bypasses the MemWAL and reads the base table only,
|
||||
* RTE write spec. `false` bypasses the RTE and reads the base table only,
|
||||
* even when a spec is present.
|
||||
*
|
||||
* Note: the LSM scanner does not support every query shape (e.g. reranking,
|
||||
* hybrid search, `orderBy`). On a MemWAL table those shapes error unless
|
||||
* `useMemWal(false)` is set, because a base-only read would silently exclude
|
||||
* un-compacted MemWAL data.
|
||||
* hybrid search, `orderBy`). On an RTE table those shapes error unless
|
||||
* `useRTE(false)` is set, because a base-only read would silently exclude
|
||||
* un-compacted RTE data.
|
||||
*/
|
||||
useMemWal(enable: boolean): this {
|
||||
this.doCall((inner: NativeQueryType) => inner.useMemWal(enable));
|
||||
useRTE(enable: boolean): this {
|
||||
this.doCall((inner: NativeQueryType) => inner.useRTE(enable));
|
||||
return this;
|
||||
}
|
||||
}
|
||||
@@ -776,7 +776,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
|
||||
* probing may continue when too few candidates are found.
|
||||
* Any explicit effort, including 0.5, overrides nprobes in either call order.
|
||||
* Explicit minimum and maximum bounds still apply. This is not a recall
|
||||
* guarantee and does not change refinement or HNSW ef. The MemWAL LSM
|
||||
* guarantee and does not change refinement or HNSW ef. The RTE LSM
|
||||
* scanner does not yet support this option.
|
||||
*
|
||||
* @example
|
||||
@@ -1082,16 +1082,16 @@ export class TakeQuery extends QueryBase<NativeTakeQuery> {
|
||||
}
|
||||
|
||||
/**
|
||||
* Control MemWAL read routing for this take query.
|
||||
* Control RTE read routing for this take query.
|
||||
*
|
||||
* `false` bypasses the MemWAL and reads the base table only — the escape hatch,
|
||||
* `false` bypasses the RTE and reads the base table only — the escape hatch,
|
||||
* since take-by-row-id/offset is not supported on the LSM scanner and, on a
|
||||
* MemWAL table, auto-routes to it and errors otherwise.
|
||||
* RTE table, auto-routes to it and errors otherwise.
|
||||
*
|
||||
* @param enable - `false` reads the base table only.
|
||||
*/
|
||||
useMemWal(enable: boolean): this {
|
||||
this.doCall((inner: NativeTakeQuery) => inner.useMemWal(enable));
|
||||
useRTE(enable: boolean): this {
|
||||
this.doCall((inner: NativeTakeQuery) => inner.useRTE(enable));
|
||||
return this;
|
||||
}
|
||||
}
|
||||
|
||||
+54
-54
@@ -35,9 +35,9 @@ import {
|
||||
FunctionErrorsOptions,
|
||||
IndexConfig,
|
||||
IndexStatistics,
|
||||
MemWalStats,
|
||||
Branches as NativeBranches,
|
||||
OptimizeStats,
|
||||
RTEStats,
|
||||
RefreshColumnResult,
|
||||
RefreshMaterializedViewResult,
|
||||
TableStatistics,
|
||||
@@ -61,7 +61,7 @@ export { IndexConfig } from "./native";
|
||||
export {
|
||||
TableShardStats,
|
||||
SsTableStats,
|
||||
MemWalStats,
|
||||
RTEStats,
|
||||
MemtableStats,
|
||||
} from "./native";
|
||||
|
||||
@@ -207,8 +207,8 @@ export type TokenizeTableOptions =
|
||||
};
|
||||
|
||||
/**
|
||||
* Specification selecting Lance's MemWAL LSM-style write path for
|
||||
* `mergeInsert`.
|
||||
* Specification enabling the realtime engine (RTE), which routes
|
||||
* `mergeInsert` through Lance's MemWAL LSM-style write path.
|
||||
*
|
||||
* `specType` is `"bucket"`, `"identity"`, or `"unsharded"`. For `"bucket"`,
|
||||
* `column` and `numBuckets` are required; for `"identity"`, `column` is
|
||||
@@ -217,7 +217,7 @@ export type TokenizeTableOptions =
|
||||
* `column` value, or upserts of that key can land in different shards and a
|
||||
* stale version can win).
|
||||
*/
|
||||
export interface MemWalWriteSpec {
|
||||
export interface RTEWriteSpec {
|
||||
/** One of `"bucket"`, `"identity"`, or `"unsharded"`. */
|
||||
specType: "bucket" | "identity" | "unsharded";
|
||||
/** Bucket and identity variants: the sharding column. */
|
||||
@@ -225,7 +225,7 @@ export interface MemWalWriteSpec {
|
||||
/** Bucket variant: the number of buckets, in `[1, 1024]`. */
|
||||
numBuckets?: number;
|
||||
/**
|
||||
* Indexes the MemWAL keeps up to date. Omit for every index the table has,
|
||||
* Indexes the RTE keeps up to date. Omit for every index the table has,
|
||||
* including ones created later. Pass `[]` for none.
|
||||
*/
|
||||
maintainedIndexes?: string[];
|
||||
@@ -758,10 +758,10 @@ export abstract class Table {
|
||||
*/
|
||||
abstract setUnenforcedPrimaryKey(columns: string | string[]): Promise<void>;
|
||||
/**
|
||||
* Install an {@link MemWalWriteSpec} on this table, selecting Lance's MemWAL
|
||||
* Install an {@link RTEWriteSpec} on this table, selecting Lance's MemWAL
|
||||
* LSM-style write path for future `mergeInsert` calls.
|
||||
*
|
||||
* `MemWalWriteSpec` chooses one of three sharding strategies via `specType`:
|
||||
* `RTEWriteSpec` chooses one of three sharding strategies via `specType`:
|
||||
*
|
||||
* - `"bucket"` — hash-bucket writes by the single-column unenforced primary
|
||||
* key (`column` and `numBuckets` required).
|
||||
@@ -773,15 +773,15 @@ export abstract class Table {
|
||||
* requires it to be the single column being bucketed.
|
||||
*
|
||||
* Omitting `maintainedIndexes` maintains every index the table has,
|
||||
* including ones created later, and skips a kind the MemWAL cannot maintain.
|
||||
* including ones created later, and skips a kind the RTE cannot maintain.
|
||||
* Naming them pins an exact set, and a still-building index is rejected
|
||||
* rather than quietly omitted.
|
||||
* @param {MemWalWriteSpec} spec The sharding spec to install.
|
||||
* @param {RTEWriteSpec} spec The sharding spec to install.
|
||||
* @returns {Promise<void>}
|
||||
* @example
|
||||
* ```ts
|
||||
* await table.setUnenforcedPrimaryKey("id");
|
||||
* await table.enableMemWal({
|
||||
* await table.enableRTE({
|
||||
* specType: "bucket",
|
||||
* column: "id",
|
||||
* numBuckets: 16,
|
||||
@@ -789,36 +789,36 @@ export abstract class Table {
|
||||
* });
|
||||
* ```
|
||||
*/
|
||||
abstract enableMemWal(spec: MemWalWriteSpec): Promise<void>;
|
||||
abstract enableRTE(spec: RTEWriteSpec): Promise<void>;
|
||||
/**
|
||||
* Remove the {@link MemWalWriteSpec} from this table, reverting to the standard
|
||||
* Remove the {@link RTEWriteSpec} from this table, reverting to the standard
|
||||
* `mergeInsert` write path.
|
||||
*
|
||||
* Errors if no spec is currently set.
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
abstract disableMemWal(): Promise<void>;
|
||||
abstract disableRTE(): Promise<void>;
|
||||
/**
|
||||
* Read the {@link MemWalWriteSpec} currently installed on this table.
|
||||
* Read the {@link RTEWriteSpec} currently installed on this table.
|
||||
*
|
||||
* Resolves to `undefined` when the MemWAL write path is not enabled (no
|
||||
* spec has been set, or it was removed with {@link Table#disableMemWal}).
|
||||
* Resolves to `undefined` when the RTE write path is not enabled (no
|
||||
* spec has been set, or it was removed with {@link Table#disableRTE}).
|
||||
* The spec is the one installed, including its maintained-index selection:
|
||||
* an absent `maintainedIndexes` for every index the table has, an empty
|
||||
* array for none.
|
||||
* @returns {Promise<MemWalWriteSpec | undefined>}
|
||||
* @returns {Promise<RTEWriteSpec | undefined>}
|
||||
*/
|
||||
abstract getMemWalSpec(): Promise<MemWalWriteSpec | undefined>;
|
||||
abstract getRTESpec(): Promise<RTEWriteSpec | undefined>;
|
||||
/**
|
||||
* Drain and close any cached MemWAL shard writers held for this table.
|
||||
* Drain and close any cached RTE shard writers held for this table.
|
||||
*
|
||||
* When an {@link MemWalWriteSpec} is installed, `mergeInsert` opens MemWAL
|
||||
* When an {@link RTEWriteSpec} is installed, `mergeInsert` opens RTE
|
||||
* shard writers and caches them for reuse across calls. This closes them,
|
||||
* flushing pending data; writers reopen lazily on the next `mergeInsert`.
|
||||
* It is a no-op when no writers are cached.
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
abstract closeMemWalWriters(): Promise<void>;
|
||||
abstract closeRTEWriters(): Promise<void>;
|
||||
/**
|
||||
* Freeze every table shard's active memtable into a new SSTable.
|
||||
*
|
||||
@@ -826,18 +826,18 @@ export abstract class Table {
|
||||
* so this is safe to call repeatedly.
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
abstract flushMemWal(): Promise<void>;
|
||||
abstract flushRTE(): Promise<void>;
|
||||
/**
|
||||
* Trigger a background SSTable compaction pass per table shard.
|
||||
*
|
||||
* Returns once the passes are *dispatched*, not once they finish — watch
|
||||
* {@link Table#getMemWalStats} for progress, or use
|
||||
* {@link Table#checkpointMemWal} to wait for convergence.
|
||||
* {@link Table#getRTEStats} for progress, or use
|
||||
* {@link Table#checkpointRTE} to wait for convergence.
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
abstract compactMemWal(): Promise<void>;
|
||||
abstract compactRTE(): Promise<void>;
|
||||
/**
|
||||
* Converge this table's MemWAL write path into its base table.
|
||||
* Converge this table's RTE write path into its base table.
|
||||
*
|
||||
* Freezes once, then triggers compaction and polls until the SSTables that existed
|
||||
* at the start is gone. The target set is fixed at the start, so
|
||||
@@ -852,26 +852,26 @@ export abstract class Table {
|
||||
* @returns {Promise<void>}
|
||||
* @example
|
||||
* ```ts
|
||||
* const before = await table.getMemWalStats();
|
||||
* await table.checkpointMemWal();
|
||||
* const after = await table.getMemWalStats();
|
||||
* const before = await table.getRTEStats();
|
||||
* await table.checkpointRTE();
|
||||
* const after = await table.getRTEStats();
|
||||
* ```
|
||||
*/
|
||||
abstract checkpointMemWal(): Promise<void>;
|
||||
abstract checkpointRTE(): Promise<void>;
|
||||
/**
|
||||
* Read live per-table-shard MemWAL state.
|
||||
* Read live per-table-shard RTE state.
|
||||
*
|
||||
* 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.
|
||||
*
|
||||
* Resolves to `undefined` only when the MemWAL write path is not enabled.
|
||||
* Resolves to `undefined` only when the RTE write path is not enabled.
|
||||
* @param {boolean} includeSstableRows Also count rows per SSTable.
|
||||
* Off by default because each count opens an uncached Lance dataset.
|
||||
* @returns {Promise<MemWalStats | undefined>}
|
||||
* @returns {Promise<RTEStats | undefined>}
|
||||
*/
|
||||
abstract getMemWalStats(
|
||||
abstract getRTEStats(
|
||||
includeSstableRows?: boolean,
|
||||
): Promise<MemWalStats | undefined>;
|
||||
): Promise<RTEStats | undefined>;
|
||||
/** Retrieve the version of the table */
|
||||
|
||||
abstract version(): Promise<number>;
|
||||
@@ -1480,43 +1480,43 @@ export class LocalTable extends Table {
|
||||
return await this.inner.setUnenforcedPrimaryKey(cols);
|
||||
}
|
||||
|
||||
async enableMemWal(spec: MemWalWriteSpec): Promise<void> {
|
||||
return await this.inner.enableMemWal(spec);
|
||||
async enableRTE(spec: RTEWriteSpec): Promise<void> {
|
||||
return await this.inner.enableRTE(spec);
|
||||
}
|
||||
|
||||
async disableMemWal(): Promise<void> {
|
||||
return await this.inner.disableMemWal();
|
||||
async disableRTE(): Promise<void> {
|
||||
return await this.inner.disableRTE();
|
||||
}
|
||||
|
||||
async getMemWalSpec(): Promise<MemWalWriteSpec | undefined> {
|
||||
async getRTESpec(): Promise<RTEWriteSpec | undefined> {
|
||||
// The native binding types `specType` as a plain `string`; narrow it back
|
||||
// to the public union. The Rust `From` impl only ever emits one of the
|
||||
// three valid values, so the cast is safe.
|
||||
return ((await this.inner.getMemWalSpec()) ?? undefined) as
|
||||
| MemWalWriteSpec
|
||||
return ((await this.inner.getRTESpec()) ?? undefined) as
|
||||
| RTEWriteSpec
|
||||
| undefined;
|
||||
}
|
||||
|
||||
async closeMemWalWriters(): Promise<void> {
|
||||
return await this.inner.closeMemWalWriters();
|
||||
async closeRTEWriters(): Promise<void> {
|
||||
return await this.inner.closeRTEWriters();
|
||||
}
|
||||
|
||||
async flushMemWal(): Promise<void> {
|
||||
return await this.inner.flushMemWal();
|
||||
async flushRTE(): Promise<void> {
|
||||
return await this.inner.flushRTE();
|
||||
}
|
||||
|
||||
async compactMemWal(): Promise<void> {
|
||||
return await this.inner.compactMemWal();
|
||||
async compactRTE(): Promise<void> {
|
||||
return await this.inner.compactRTE();
|
||||
}
|
||||
|
||||
async checkpointMemWal(): Promise<void> {
|
||||
return await this.inner.checkpointMemWal();
|
||||
async checkpointRTE(): Promise<void> {
|
||||
return await this.inner.checkpointRTE();
|
||||
}
|
||||
|
||||
async getMemWalStats(
|
||||
async getRTEStats(
|
||||
includeSstableRows: boolean = false,
|
||||
): Promise<MemWalStats | undefined> {
|
||||
return (await this.inner.getMemWalStats(includeSstableRows)) ?? undefined;
|
||||
): Promise<RTEStats | undefined> {
|
||||
return (await this.inner.getRTEStats(includeSstableRows)) ?? undefined;
|
||||
}
|
||||
|
||||
async version(): Promise<number> {
|
||||
|
||||
+3
-3
@@ -51,10 +51,10 @@ impl NativeMergeInsertBuilder {
|
||||
this
|
||||
}
|
||||
|
||||
#[napi(js_name = "useMemWal")]
|
||||
pub fn use_memwal(&self, enable: bool) -> Self {
|
||||
#[napi(js_name = "useRTE")]
|
||||
pub fn use_rte(&self, enable: bool) -> Self {
|
||||
let mut this = self.clone();
|
||||
this.inner.use_memwal(enable);
|
||||
this.inner.use_rte(enable);
|
||||
this
|
||||
}
|
||||
|
||||
|
||||
+9
-9
@@ -168,9 +168,9 @@ impl Query {
|
||||
self.inner = self.inner.clone().with_row_id();
|
||||
}
|
||||
|
||||
#[napi(js_name = "useMemWal")]
|
||||
pub fn use_memwal(&mut self, enable: bool) {
|
||||
self.inner = self.inner.clone().use_memwal(enable);
|
||||
#[napi(js_name = "useRTE")]
|
||||
pub fn use_rte(&mut self, enable: bool) {
|
||||
self.inner = self.inner.clone().use_rte(enable);
|
||||
}
|
||||
|
||||
#[napi]
|
||||
@@ -399,9 +399,9 @@ impl VectorQuery {
|
||||
self.inner = self.inner.clone().with_row_id();
|
||||
}
|
||||
|
||||
#[napi(js_name = "useMemWal")]
|
||||
pub fn use_memwal(&mut self, enable: bool) {
|
||||
self.inner = self.inner.clone().use_memwal(enable);
|
||||
#[napi(js_name = "useRTE")]
|
||||
pub fn use_rte(&mut self, enable: bool) {
|
||||
self.inner = self.inner.clone().use_rte(enable);
|
||||
}
|
||||
|
||||
#[napi]
|
||||
@@ -514,9 +514,9 @@ impl TakeQuery {
|
||||
self.inner = self.inner.clone().with_row_id();
|
||||
}
|
||||
|
||||
#[napi(js_name = "useMemWal")]
|
||||
pub fn use_memwal(&mut self, enable: bool) {
|
||||
self.inner = self.inner.clone().use_memwal(enable);
|
||||
#[napi(js_name = "useRTE")]
|
||||
pub fn use_rte(&mut self, enable: bool) {
|
||||
self.inner = self.inner.clone().use_rte(enable);
|
||||
}
|
||||
|
||||
#[napi(catch_unwind)]
|
||||
|
||||
+47
-50
@@ -580,60 +580,57 @@ impl Table {
|
||||
.default_error()
|
||||
}
|
||||
|
||||
#[napi(catch_unwind, js_name = "enableMemWal")]
|
||||
pub async fn enable_memwal(&self, spec: MemWalWriteSpec) -> napi::Result<()> {
|
||||
let native_spec = lancedb::table::MemWalWriteSpec::try_from(spec)?;
|
||||
#[napi(catch_unwind, js_name = "enableRTE")]
|
||||
pub async fn enable_rte(&self, spec: RTEWriteSpec) -> napi::Result<()> {
|
||||
let native_spec = lancedb::table::RTEWriteSpec::try_from(spec)?;
|
||||
self.inner_ref()?
|
||||
.enable_memwal(native_spec)
|
||||
.enable_rte(native_spec)
|
||||
.await
|
||||
.default_error()
|
||||
}
|
||||
|
||||
#[napi(catch_unwind, js_name = "disableMemWal")]
|
||||
pub async fn disable_memwal(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?.disable_memwal().await.default_error()
|
||||
#[napi(catch_unwind, js_name = "disableRTE")]
|
||||
pub async fn disable_rte(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?.disable_rte().await.default_error()
|
||||
}
|
||||
|
||||
#[napi(catch_unwind, js_name = "getMemWalSpec")]
|
||||
pub async fn get_memwal_spec(&self) -> napi::Result<Option<MemWalWriteSpec>> {
|
||||
let spec = self.inner_ref()?.get_memwal_spec().await.default_error()?;
|
||||
Ok(spec.map(MemWalWriteSpec::from))
|
||||
#[napi(catch_unwind, js_name = "getRTESpec")]
|
||||
pub async fn get_rte_spec(&self) -> napi::Result<Option<RTEWriteSpec>> {
|
||||
let spec = self.inner_ref()?.get_rte_spec().await.default_error()?;
|
||||
Ok(spec.map(RTEWriteSpec::from))
|
||||
}
|
||||
|
||||
#[napi(catch_unwind, js_name = "closeMemWalWriters")]
|
||||
pub async fn close_memwal_writers(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?
|
||||
.close_memwal_writers()
|
||||
.await
|
||||
.default_error()
|
||||
#[napi(catch_unwind, js_name = "closeRTEWriters")]
|
||||
pub async fn close_rte_writers(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?.close_rte_writers().await.default_error()
|
||||
}
|
||||
|
||||
#[napi(catch_unwind, js_name = "flushMemWal")]
|
||||
pub async fn flush_memwal(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?.flush_memwal().await.default_error()
|
||||
#[napi(catch_unwind, js_name = "flushRTE")]
|
||||
pub async fn flush_rte(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?.flush_rte().await.default_error()
|
||||
}
|
||||
|
||||
#[napi(catch_unwind, js_name = "compactMemWal")]
|
||||
pub async fn compact_memwal(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?.compact_memwal().await.default_error()
|
||||
#[napi(catch_unwind, js_name = "compactRTE")]
|
||||
pub async fn compact_rte(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?.compact_rte().await.default_error()
|
||||
}
|
||||
|
||||
#[napi(catch_unwind, js_name = "checkpointMemWal")]
|
||||
pub async fn checkpoint_memwal(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?.checkpoint_memwal().await.default_error()
|
||||
#[napi(catch_unwind, js_name = "checkpointRTE")]
|
||||
pub async fn checkpoint_rte(&self) -> napi::Result<()> {
|
||||
self.inner_ref()?.checkpoint_rte().await.default_error()
|
||||
}
|
||||
|
||||
#[napi(catch_unwind, js_name = "getMemWalStats")]
|
||||
pub async fn get_memwal_stats(
|
||||
#[napi(catch_unwind, js_name = "getRTEStats")]
|
||||
pub async fn get_rte_stats(
|
||||
&self,
|
||||
include_sstable_rows: bool,
|
||||
) -> napi::Result<Option<MemWalStats>> {
|
||||
) -> napi::Result<Option<RTEStats>> {
|
||||
let stats = self
|
||||
.inner_ref()?
|
||||
.get_memwal_stats(include_sstable_rows)
|
||||
.get_rte_stats(include_sstable_rows)
|
||||
.await
|
||||
.default_error()?;
|
||||
Ok(stats.map(MemWalStats::from))
|
||||
Ok(stats.map(RTEStats::from))
|
||||
}
|
||||
|
||||
#[napi(catch_unwind)]
|
||||
@@ -946,53 +943,53 @@ impl From<LanceDbFtsToken> for FtsToken {
|
||||
}
|
||||
}
|
||||
|
||||
/// Specification selecting Lance's MemWAL LSM-style write path for
|
||||
/// `mergeInsert`.
|
||||
/// Specification enabling the realtime engine (RTE), which routes
|
||||
/// `mergeInsert` through Lance's MemWAL LSM-style write path.
|
||||
///
|
||||
/// `specType` must be `"bucket"`, `"identity"`, or `"unsharded"`. For
|
||||
/// `"bucket"`, `column` and `numBuckets` are required; for `"identity"`,
|
||||
/// `column` is required.
|
||||
#[napi(object)]
|
||||
#[napi(object, js_name = "RTEWriteSpec")]
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct MemWalWriteSpec {
|
||||
pub struct RTEWriteSpec {
|
||||
/// One of `"bucket"`, `"identity"`, or `"unsharded"`.
|
||||
pub spec_type: String,
|
||||
/// Bucket and identity variants: the sharding column.
|
||||
pub column: Option<String>,
|
||||
/// Bucket variant: the number of buckets, in `[1, 1024]`.
|
||||
pub num_buckets: Option<u32>,
|
||||
/// Indexes the MemWAL keeps up to date. Omitted resolves every
|
||||
/// Indexes the RTE keeps up to date. Omitted resolves every
|
||||
/// maintainable index on install; an empty array means none.
|
||||
pub maintained_indexes: Option<Vec<String>>,
|
||||
/// Default `ShardWriter` configuration recorded in the MemWAL index.
|
||||
pub writer_config_defaults: Option<HashMap<String, String>>,
|
||||
}
|
||||
|
||||
impl TryFrom<MemWalWriteSpec> for lancedb::table::MemWalWriteSpec {
|
||||
impl TryFrom<RTEWriteSpec> for lancedb::table::RTEWriteSpec {
|
||||
type Error = napi::Error;
|
||||
|
||||
fn try_from(value: MemWalWriteSpec) -> napi::Result<Self> {
|
||||
fn try_from(value: RTEWriteSpec) -> napi::Result<Self> {
|
||||
let writer_config_defaults = value.writer_config_defaults.unwrap_or_default();
|
||||
let spec = match value.spec_type.as_str() {
|
||||
"bucket" => {
|
||||
let column = value.column.ok_or_else(|| {
|
||||
napi::Error::from_reason("MemWalWriteSpec bucket requires `column`")
|
||||
napi::Error::from_reason("RTEWriteSpec bucket requires `column`")
|
||||
})?;
|
||||
let num_buckets = value.num_buckets.ok_or_else(|| {
|
||||
napi::Error::from_reason("MemWalWriteSpec bucket requires `numBuckets`")
|
||||
napi::Error::from_reason("RTEWriteSpec bucket requires `numBuckets`")
|
||||
})?;
|
||||
Self::bucket(column, num_buckets)
|
||||
}
|
||||
"identity" => {
|
||||
let column = value.column.ok_or_else(|| {
|
||||
napi::Error::from_reason("MemWalWriteSpec identity requires `column`")
|
||||
napi::Error::from_reason("RTEWriteSpec identity requires `column`")
|
||||
})?;
|
||||
Self::identity(column)
|
||||
}
|
||||
"unsharded" => Self::unsharded(),
|
||||
other => {
|
||||
return Err(napi::Error::from_reason(format!(
|
||||
"MemWalWriteSpec `specType` must be 'bucket', 'identity', or 'unsharded', got '{}'",
|
||||
"RTEWriteSpec `specType` must be 'bucket', 'identity', or 'unsharded', got '{}'",
|
||||
other
|
||||
)));
|
||||
}
|
||||
@@ -1003,9 +1000,9 @@ impl TryFrom<MemWalWriteSpec> for lancedb::table::MemWalWriteSpec {
|
||||
}
|
||||
}
|
||||
|
||||
impl From<lancedb::table::MemWalWriteSpec> for MemWalWriteSpec {
|
||||
fn from(spec: lancedb::table::MemWalWriteSpec) -> Self {
|
||||
use lancedb::table::MemWalWriteSpec as Native;
|
||||
impl From<lancedb::table::RTEWriteSpec> for RTEWriteSpec {
|
||||
fn from(spec: lancedb::table::RTEWriteSpec) -> Self {
|
||||
use lancedb::table::RTEWriteSpec as Native;
|
||||
match spec {
|
||||
Native::Bucket {
|
||||
column,
|
||||
@@ -1148,19 +1145,19 @@ impl From<lancedb::table::TableShardStats> for TableShardStats {
|
||||
}
|
||||
}
|
||||
|
||||
/// Live per-table-shard MemWAL state, as returned by `Table#getMemWalStats`.
|
||||
/// Live per-table-shard RTE state, as returned by `Table#getRTEStats`.
|
||||
///
|
||||
/// Nothing here is derived: sums and differences (total SSTable bytes, WAL lag) are
|
||||
/// the caller's to compute.
|
||||
#[napi(object)]
|
||||
#[napi(object, js_name = "RTEStats")]
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct MemWalStats {
|
||||
pub struct RTEStats {
|
||||
/// One entry per table shard backing this table.
|
||||
pub table_shards: Vec<TableShardStats>,
|
||||
}
|
||||
|
||||
impl From<lancedb::table::MemWalStats> for MemWalStats {
|
||||
fn from(stats: lancedb::table::MemWalStats) -> Self {
|
||||
impl From<lancedb::table::RTEStats> for RTEStats {
|
||||
fn from(stats: lancedb::table::RTEStats) -> Self {
|
||||
Self {
|
||||
table_shards: stats.table_shards.into_iter().map(Into::into).collect(),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user