rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-10-13 07:12:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(lsm): finish SSTable and table shard naming

Browse Source 
Applies the SSTable / table shard / freeze vocabulary from the previous
commit to code that came in from main, and to spots that commit missed:

- `get_lsm_stats` test fixtures in Rust and Python still sent the old
  `buckets` / `generations` keys, so the Rust test target no longer
  compiled and the Python tests would fail against `table_shards`.
- The Java javadoc example called a nonexistent `LsmWriteSpec.tableShard`,
  an over-rename of `LsmWriteSpec.bucket`; the README used the old
  `buckets()` / `generations()` accessors.
- Prose that still said bucket (for a shard), seal, L0, or flushed
  generations, and identifier-style "table_shard" / "tableShard" in prose.

`LsmWriteSpec::bucket`, `num_buckets`, the hash-bucket values in the
merge-insert router, and the `"Sealed"` status wire value are unchanged:
those name the sharding transform or a server value, not the shard.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Daniel Rammer
2026-10-12 15:30:06 -05:00
co-authored by Claude Opus 5.5
parent d5544cd5d9
commit 06d7f82fa2
23 changed files with 125 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/web/maintenance/background-jobs.mdx
+1 -1
View File
@@ -25,7 +25,7 @@ log from growing without bound:
| Job type | What it does |
| --- | --- |
| `sstable_compaction` | Folds a WAL bucket's SSTables into the base table by primary key. Distinct from `compact`, which merges base-table fragments -- the two are unrelated. |
| `sstable_compaction` | Folds a table shard's SSTables into the base table by primary key. Distinct from `compact`, which merges base-table fragments -- the two are unrelated. |
| `wal_index_catchup` | Extends a base-table index over the rows `sstable_compaction` just folded in, and records how far the index has caught up. Nothing else advances that record. |
## Index maintenance {#index}
java/README.md
+3 -3
View File
@@ -55,9 +55,9 @@ lsm.setLsmWriteSpec(LsmWriteSpec.bucket("id", 16));
// Converge the fresh tier into the base table.
lsm.checkpointLsm();
// Inspect live per-bucket state.
lsm.getLsmStats().ifPresent(stats -> stats.buckets().forEach(bucket ->
System.out.println(bucket.shardId() + ": " + bucket.generations().size() + " L0 generations")));
// Inspect live per-table-shard state.
lsm.getLsmStats().ifPresent(stats -> stats.tableShards().forEach(tableShard ->
System.out.println(tableShard.shardId() + ": " + tableShard.sstables().size() + " SSTables")));
client.close();
```
java/lancedb-core/src/main/java/com/lancedb/JsonFields.java
+1 -1
View File
@@ -25,7 +25,7 @@ import com.fasterxml.jackson.databind.JsonNode;
*
* <p>The alternative — Jackson's {@code path()}, which yields a missing node that reads as an empty
* array or a zero — is unsafe here because {@link LanceDbTableLsm#checkpointLsm()} decides
* convergence from these numbers. A defaulted {@code generations} array is indistinguishable from a
* 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.
*/
final class JsonFields {
java/lancedb-core/src/main/java/com/lancedb/LanceDbTableLsm.java
+12 -12
View File
@@ -38,7 +38,7 @@ import java.util.OptionalLong;
* .buildRestClient();
*
* LanceDbTableLsm lsm = new LanceDbTableLsm(client, "my_table");
* lsm.setLsmWriteSpec(LsmWriteSpec.tableShard("id", 16));
* lsm.setLsmWriteSpec(LsmWriteSpec.bucket("id", 16));
* // ... merge_insert traffic ...
* lsm.checkpointLsm();
* }</pre>
@@ -94,7 +94,7 @@ public class LanceDbTableLsm {
* Install an {@link LsmWriteSpec} on this table, selecting the MemWAL LSM write path for future
* {@code mergeInsert} calls.
*
* <p>All variants require the table to have an unenforced primary key; tableShard sharding
* <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 setLsmWriteSpec(LsmWriteSpec spec) {
@@ -136,7 +136,7 @@ public class LanceDbTableLsm {
/**
* Freeze every table shard's active memtable into a new SSTable.
*
* <p>Returns once the seal is committed. Sealing an empty memtable is a no-op, so this is safe to
* <p>Returns once the freeze is committed. Freezing an empty memtable is a no-op, so this is safe to
* call repeatedly.
*/
public void flushLsm() {
@@ -154,9 +154,9 @@ public class LanceDbTableLsm {
}
/**
* Read live per-tableShard LSM state.
* Read live per-table-shard LSM state.
*
* <p>Answers "how far behind is my fresh tier", "which tableShard is hot", and "why is my fresh-tier
* <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 LSM write path is not enabled — that is, when the server sends an absent
@@ -206,8 +206,8 @@ public class LanceDbTableLsm {
*/
public void checkpointLsm() {
for (int reissue = 0; reissue <= MAX_REISSUES; reissue++) {
// The seal turns everything written before this call into a generation, so the
// watermark has to be read after it. Idempotent: sealing an empty memtable is a
// 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::flushLsm)) {
backoff(reissue);
@@ -240,7 +240,7 @@ public class LanceDbTableLsm {
}
/**
* Trigger and poll until no tableShard holds a generation at or below its target.
* Trigger and poll until no table shard holds an SSTable generation at or below its target.
*
* @return true when the drain finished, false when the table needs re-claiming from flush.
*/
@@ -254,9 +254,9 @@ public class LanceDbTableLsm {
return true;
}
// `compacting` is the tableShard's compaction latch, held from dispatch until the pass
// `compacting` is the table shard's compaction latch, held from dispatch until the pass
// ends — including while it waits on a pod-wide permit. So it answers one question
// only: do not pile on. Buckets with nothing outstanding are skipped, not counted
// only: do not pile on. Table shards with nothing outstanding are skipped, not counted
// as idle.
long outstanding = 0;
boolean allCompacting = true;
@@ -285,7 +285,7 @@ public class LanceDbTableLsm {
if (!isRetryable(e)) {
throw e;
}
// A 429 here means the server could latch no tableShard at all, which the poll
// A 429 here means the server could latch no table shard at all, which the poll
// above already handles. Not retried in place: the latch it would contend for
// is the one doing the work, so fall through and re-read — POLL_INTERVAL_MS is
// the backoff.
@@ -295,7 +295,7 @@ public class LanceDbTableLsm {
}
}
/** The newest generation held by each tableShard, skipping tableShards holding none. */
/** The newest SSTable generation held by each table shard, skipping table shards holding none. */
private static Map<String, Long> newestSstableGenerations(LsmStats stats) {
Map<String, Long> targets = new HashMap<String, Long>();
for (TableShardStats tableShard : stats.tableShards()) {
java/lancedb-core/src/main/java/com/lancedb/LsmStats.java
+2 -2
View File
@@ -20,7 +20,7 @@ import java.util.Collections;
import java.util.List;
/**
* Live per-tableShard LSM state, as returned by {@link LanceDbTableLsm#getLsmStats()}.
* Live per-table-shard LSM state, as returned by {@link LanceDbTableLsm#getLsmStats()}.
*
* <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},
@@ -35,7 +35,7 @@ public class LsmStats {
this.tableShards = Collections.unmodifiableList(tableShards);
}
/** One entry per tableShard. */
/** One entry per table shard. */
public List<TableShardStats> tableShards() {
return tableShards;
}
java/lancedb-core/src/main/java/com/lancedb/MemtableStats.java
+1 -1
View File
@@ -37,7 +37,7 @@ public class MemtableStats {
this.indexes = Collections.unmodifiableList(indexes);
}
/** The generation this memtable will become once sealed. */
/** The generation this memtable will become once frozen. */
public long generation() {
return generation;
}
java/lancedb-core/src/main/java/com/lancedb/TableShardStats.java
+5 -5
View File
@@ -22,8 +22,8 @@ import java.util.Optional;
import java.util.OptionalLong;
/**
* Live state of one tableShard. A table is N tableShards on one node; flattening to a single number hides
* the one hot tableShard that is usually why someone opened this endpoint.
* Live state of one table shard. A table is N table shards on one node; flattening to a single number hides
* the one hot table shard that is usually why someone opened this endpoint.
*/
public class TableShardStats {
private static final String CONTEXT = "tableShard stats";
@@ -62,7 +62,7 @@ public class TableShardStats {
this.memtables = memtables == null ? null : Collections.unmodifiableList(memtables);
}
/** The shard this tableShard writes. */
/** The id of this table shard. */
public String shardId() {
return shardId;
}
@@ -106,7 +106,7 @@ public class TableShardStats {
}
/**
* Whether a pass owns this tableShard's compaction latch right now. Says <em>a</em> driver is
* Whether a pass owns this table shard's compaction latch right now. Says <em>a</em> driver is
* running, not <em>whose</em>, and the latch is held from dispatch — including while the pass
* queues for a pod-wide compactor permit. Read it as "do not pile on", never as "mine is
* progressing".
@@ -115,7 +115,7 @@ public class TableShardStats {
return compacting;
}
/** Oldest first, active last. Empty for a {@code "Sealed"} tableShard, whose state is torn down. */
/** Oldest first, active last. Empty for a {@code "Sealed"} table shard, whose state is torn down. */
public Optional<List<MemtableStats>> memtables() {
return Optional.ofNullable(memtables);
}
nodejs/__test__/table.test.ts
+1 -1
View File
@@ -4908,7 +4908,7 @@ describe("LSM convergence and stats", () => {
const conn = await connect(tmpDir.name);
const table = await lsmTable(conn);
// checkpointLsm seals first, so it surfaces flushLsm's rejection.
// checkpointLsm freezes first, so it surfaces flushLsm's rejection.
await expect(table.checkpointLsm()).rejects.toThrow(/not supported/i);
});
});
nodejs/lancedb/query.ts
+1 -1
View File
@@ -467,7 +467,7 @@ export class StandardQueryBase<
* {@link Table#setLsmWriteSpec}), 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 flushed generations), deduplicated by primary key; a table without a spec
* 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
nodejs/lancedb/table.ts
+3 -3
View File
@@ -815,7 +815,7 @@ export abstract class Table {
/**
* Freeze every table shard's active memtable into a new SSTable.
*
* Returns once the seal is committed. Sealing an empty memtable is a no-op,
* Returns once the freeze is committed. Freezing an empty memtable is a no-op,
* so this is safe to call repeatedly.
* @returns {Promise<void>}
*/
@@ -852,9 +852,9 @@ export abstract class Table {
*/
abstract checkpointLsm(): Promise<void>;
/**
* Read live per-bucket LSM state.
* Read live per-table-shard LSM state.
*
* Answers "how far behind is my fresh tier", "which bucket is hot", and
* 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 LSM write path is not enabled.
nodejs/src/table.rs
+1 -1
View File
@@ -1106,7 +1106,7 @@ impl From<lancedb::table::MemtableStats> for MemtableStats {
#[napi(object)]
#[derive(Clone, Debug)]
pub struct TableShardStats {
/// The shard this table shard writes.
/// The id of this table shard.
pub shard_id: String,
/// `"Active"` or `"Sealed"` (drop-table 2PC in flight).
pub status: String,
python/python/lancedb/query.py
+2 -2
View File
@@ -1386,7 +1386,7 @@ class LanceQueryBuilder(ABC):
By default (unset), a query against a table with an LSM write spec is
routed through the LSM scanner so it also returns data written via the
``merge_insert`` LSM path that has not yet been compacted into the base
table (active/frozen memtables + flushed generations); a table without a
table (active/frozen memtables + SSTables); a table without a
spec reads the base table.
Parameters
@@ -3461,7 +3461,7 @@ class AsyncStandardQuery(AsyncQueryBase):
[AsyncTable.set_lsm_write_spec][lancedb.table.AsyncTable.set_lsm_write_spec])
is routed through the LSM scanner so it also returns data written via the
``merge_insert`` LSM path that has not yet been compacted into the base
table (the active/frozen in-memory memtables and the flushed generations),
table (the active/frozen in-memory memtables and the SSTables),
deduplicated by primary key; a table without a spec reads the base table.
Parameters
python/python/lancedb/table.py
+3 -3
View File
@@ -5477,7 +5477,7 @@ class AsyncTable:
"""Converge this table's LSM write path into its base table.
One flush, freezing every memtable into an SSTable, then compaction triggers
until every generation that existed at that moment has reached base.
until every SSTable that existed at that moment has reached base.
The loop runs client-side, reading progress from ``get_lsm_stats``.
Best-effort: SSTables created *while* it runs are deliberately not
@@ -5516,9 +5516,9 @@ class AsyncTable:
async def get_lsm_stats(
self, *, include_sstable_rows: bool = False
) -> Optional[dict]:
"""Read live per-bucket LSM state.
"""Read live per-table-shard LSM state.
Answers "how far behind is my fresh tier", "which bucket is hot", and
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, though on a node that has not claimed this table it claims it,
exactly as a read would.
python/python/tests/test_remote_db.py
+6 -6
View File
@@ -1882,7 +1882,7 @@ def send_json(request, payload, status=200):
def test_get_lsm_stats_sync():
"""The sync wrapper round-trips the server payload into a dict."""
bucket = {
table_shard = {
"shard_id": "b0",
"status": "Active",
"writer_epoch": 3,
@@ -1890,7 +1890,7 @@ def test_get_lsm_stats_sync():
"current_generation": 6,
"replay_after_wal_entry_position": 40,
"wal_entry_position_last_seen": 42,
"generations": [{"generation": 5, "bytes": 1024, "rows": 7}],
"sstables": [{"generation": 5, "bytes": 1024, "rows": 7}],
"compacting": False,
"memtables": [
{
@@ -1907,10 +1907,10 @@ def test_get_lsm_stats_sync():
def lsm_handler(request, route):
assert route == "get_lsm_stats"
seen_bodies.append(read_json_body(request))
send_json(request, {"lsm_stats": {"buckets": [bucket]}})
send_json(request, {"lsm_stats": {"table_shards": [table_shard]}})
with lsm_test_table(lsm_handler) as table:
assert table.get_lsm_stats() == {"buckets": [bucket]}
assert table.get_lsm_stats() == {"table_shards": [table_shard]}
# Off by default, and forwarded when asked for.
assert seen_bodies == [{"include_sstable_rows": False}]
table.get_lsm_stats(include_sstable_rows=True)
@@ -1954,8 +1954,8 @@ def test_checkpoint_lsm_sync():
called.append(route)
if route == "get_lsm_stats":
# An empty SSTable tier yields no target watermark, so the loop is done
# after the seal without ever polling compaction.
send_json(request, {"lsm_stats": {"buckets": []}})
# after the freeze without ever polling compaction.
send_json(request, {"lsm_stats": {"table_shards": []}})
else:
request.send_response(202)
request.end_headers()
rust/lancedb/src/query.rs
+1 -1
View File
@@ -550,7 +550,7 @@ pub trait QueryBase {
/// [`crate::Table::set_lsm_write_spec`]), reads are routed through the LSM
/// scanner so they also return data written via the `merge_insert` LSM path
/// that has not yet been compacted into the base table (active/frozen
/// memtables and flushed generations); a table without a spec reads the base
/// memtables and SSTables); a table without a spec reads the base
/// table.
///
/// - `use_lsm(true)` forces LSM routing and errors if the table has no
rust/lancedb/src/remote/table.rs
+20 -20
View File
@@ -9968,19 +9968,19 @@ mod tests {
assert!(table.get_lsm_write_spec().await.unwrap().is_none());
}
/// Build a `get_lsm_stats` body for one bucket holding `generations`.
fn stats_body(generations: &[u64], compacting: bool) -> String {
/// Build a `get_lsm_stats` body for one table shard holding `sstables`.
fn stats_body(sstables: &[u64], compacting: bool) -> String {
serde_json::json!({
"lsm_stats": {
"buckets": [{
"table_shards": [{
"shard_id": "b0",
"status": "Active",
"writer_epoch": 1,
"manifest_version": 1,
"current_generation": generations.iter().max().copied().unwrap_or(0) + 1,
"current_generation": sstables.iter().max().copied().unwrap_or(0) + 1,
"replay_after_wal_entry_position": 0,
"wal_entry_position_last_seen": 0,
"generations": generations.iter()
"sstables": sstables.iter()
.map(|g| serde_json::json!({ "generation": g, "bytes": 1 }))
.collect::<Vec<_>>(),
"compacting": compacting,
@@ -10086,7 +10086,7 @@ mod tests {
assert_eq!(
compacts.load(std::sync::atomic::Ordering::SeqCst),
1,
"the loop must not chase generations written after it started"
"the loop must not chase SSTables written after it started"
);
}
@@ -10103,7 +10103,7 @@ mod tests {
return accepted();
}
if path.contains("compact_lsm") {
// First two triggers: every bucket already latched.
// First two triggers: every table shard already latched.
if seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst) < 2 {
return http::Response::builder()
.status(429)
@@ -10201,7 +10201,7 @@ mod tests {
assert_eq!(
flushes.load(std::sync::atomic::Ordering::SeqCst),
6,
"five retries against one seal, then it lands"
"five retries against one freeze, then it lands"
);
}
@@ -10270,7 +10270,7 @@ mod tests {
/// A long stall with nothing compacting must keep waiting, not fail. The
/// client cannot judge this: a checkpoint queued behind unrelated tables
/// on the pod-wide compactor pool reports exactly these numbers — flat
/// generations, an idle latch — as one whose merges are failing. The
/// SSTable counts, an idle latch — as one whose merges are failing. The
/// deadline is the caller's.
#[tokio::test(start_paused = true)]
async fn test_checkpoint_waits_out_a_long_stall_rather_than_failing() {
@@ -10297,7 +10297,7 @@ mod tests {
);
}
/// A pass already owns the latch on every outstanding bucket, so the loop
/// A pass already owns the latch on every outstanding table shard, so the loop
/// waits rather than piling on triggers it would only refuse. This is the
/// sole thing `compacting` is read for.
#[tokio::test(start_paused = true)]
@@ -10331,7 +10331,7 @@ mod tests {
assert_eq!(
compacts.load(std::sync::atomic::Ordering::SeqCst),
0,
"never trigger against a bucket already compacting"
"never trigger against a table shard already compacting"
);
}
@@ -10350,7 +10350,7 @@ mod tests {
);
let response = serde_json::json!({
"lsm_stats": {
"buckets": [{
"table_shards": [{
"shard_id": "b0",
"status": "Active",
"writer_epoch": 3,
@@ -10358,7 +10358,7 @@ mod tests {
"current_generation": 9,
"replay_after_wal_entry_position": 100,
"wal_entry_position_last_seen": 140,
"generations": [{ "generation": 8, "bytes": 4096, "rows": 30 }],
"sstables": [{ "generation": 8, "bytes": 4096, "rows": 30 }],
"compacting": false,
"memtables": [
{ "generation": 9, "rows": 12, "bytes": 900, "batches": 2,
@@ -10378,15 +10378,15 @@ mod tests {
.await
.unwrap()
.expect("a WAL-backed table reports Some");
let bucket = &stats.buckets[0];
assert_eq!(bucket.replay_after_wal_entry_position, 100);
assert_eq!(bucket.wal_entry_position_last_seen, 140);
assert!(!bucket.compacting);
assert_eq!(bucket.generations[0].generation, 8);
assert_eq!(bucket.generations[0].rows, Some(30));
let table_shard = &stats.table_shards[0];
assert_eq!(table_shard.replay_after_wal_entry_position, 100);
assert_eq!(table_shard.wal_entry_position_last_seen, 140);
assert!(!table_shard.compacting);
assert_eq!(table_shard.sstables[0].generation, 8);
assert_eq!(table_shard.sstables[0].rows, Some(30));
// The line that answers "why is my fresh-tier vector search
// brute-force" — an absent index name is the whole explanation.
let memtables = bucket.memtables.as_ref().unwrap();
let memtables = table_shard.memtables.as_ref().unwrap();
assert_eq!(memtables[0].indexes, vec!["vec_idx".to_string()]);
}
rust/lancedb/src/table.rs
+5 -5
View File
@@ -401,7 +401,7 @@ impl LsmWriteSpec {
/// `column` must be a deterministic function of the unenforced primary
/// key: every row with a given primary key must always produce the same
/// `column` value. MemWAL dedups upserts by primary key but tracks
/// generations per shard, so if the same key is written with two
/// SSTables per shard, so if the same key is written with two
/// different `column` values its versions land in different shards and a
/// stale value can win. Typically `column` is the primary key itself, or
/// a stable attribute of it (e.g. a tenant id).
@@ -2008,7 +2008,7 @@ impl Table {
/// - [`LsmWriteSpec::unsharded`] — route every write to a single shard.
///
/// A table carries one spec: this fails while one is installed, since the
/// generations already written were homed under it.
/// SSTables already written were homed under it.
/// [`Table::unset_lsm_write_spec`] removes one.
///
/// # Example
@@ -2078,7 +2078,7 @@ impl Table {
/// already-converged table costs two round trips and triggers nothing.
///
/// **No deadline, and the caller owns that.** It returns when the target
/// generations are gone, propagates a terminal server fault, and
/// SSTables are gone, propagates a terminal server fault, and
/// otherwise waits however long the server takes. A slow table and a
/// stuck one are the same picture from here: the compactor pool is shared
/// across every table on the node, so a checkpoint queued behind
@@ -2121,9 +2121,9 @@ impl Table {
self.inner.compact_lsm().await
}
/// Read live per-bucket LSM state.
/// Read live per-table-shard LSM state.
///
/// Answers "how far behind is my fresh tier", "which bucket is hot", and
/// 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, though on a node that has not claimed this table it claims it,
/// exactly as a read would.
rust/lancedb/src/table/checkpoint.rs
+13 -13
View File
@@ -3,7 +3,7 @@
//! Converging a table's LSM write path into its base table.
//!
//! `checkpoint_lsm` seals once, then triggers compaction and watches
//! `checkpoint_lsm` freezes once, then triggers compaction and watches
//! generation numbers until the SSTables that existed at the start are gone.
//!
//! The loop runs in the client, not the server: `compact_lsm` dispatches a
@@ -12,7 +12,7 @@
//! generation numbers in the shard manifest — durable state, unlike a count
//! in a compact response, which a concurrent write invalidates.
//!
//! The target set is fixed at the start, so generations created *during* the
//! The target set is fixed at the start, so SSTables created *during* the
//! checkpoint are ignored. That is what lets it terminate under write load,
//! and what makes it best-effort: it converges the fresh tier as of some
//! instant. Idempotent, abandonable at any point, safe on a cadence.
@@ -149,14 +149,14 @@ where
}
}
/// Drive [`Table::checkpoint_lsm`]: seal once, fix the target watermark
/// Drive [`Table::checkpoint_lsm`]: freeze once, fix the target watermark
/// from the resulting SSTables, then trigger and poll until they drain.
pub(crate) async fn checkpoint_lsm(table: &Table) -> Result<()> {
for reissue in 0..=MAX_REISSUES {
// The seal turns everything written before this call into a
// generation, so the watermark has to be read after it. Idempotent:
// sealing an empty memtable is a no-op, so a re-issue does not churn
// empty generations.
// 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.
match issue(|| table.flush_lsm()).await? {
Attempt::Ok(()) => {}
Attempt::ReissueFromFlush => {
@@ -200,12 +200,12 @@ pub(crate) async fn checkpoint_lsm(table: &Table) -> Result<()> {
})
}
/// Trigger and poll until no bucket holds a generation at or below its
/// Trigger and poll until no table shard holds an SSTable generation at or below its
/// target.
///
/// No liveness bound, deliberately. The pod-wide compactor pool (a semaphore
/// of 2 by default, shared across every table on the node) is taken *inside*
/// the pass, after the bucket latch, so a checkpoint queued behind unrelated
/// the pass, after the table shard latch, so a checkpoint queued behind unrelated
/// tables is indistinguishable from one that is merging. An idle-poll counter
/// here could only ever have fired on a table that would have finished.
async fn drain_to_targets(
@@ -220,10 +220,10 @@ async fn drain_to_targets(
let Some(stats) = stats else {
return Ok(CheckpointOutcome::Done);
};
// `compacting` is the bucket's compaction latch, held from dispatch
// `compacting` is the table shard's compaction latch, held from dispatch
// until the pass ends — including while it waits on the pod-wide
// permit. So it answers one question only: do not pile on. Buckets
// with nothing outstanding are skipped, not counted as idle.
// permit. So it answers one question only: do not pile on. Table
// shards with nothing outstanding are skipped, not counted as idle.
let mut outstanding = 0;
let mut all_compacting = true;
for b in &stats.table_shards {
@@ -245,7 +245,7 @@ async fn drain_to_targets(
Ok(()) => {}
Err(e) if is_lost_claim(&e) => return Ok(CheckpointOutcome::ReissueFromFlush),
Err(e) if !is_retryable(&e) => return Err(e),
// A 429 here means the server could latch no bucket at all,
// A 429 here means the server could latch no table shard at all,
// which the poll above already handles. Not retried in place:
// the latch it would contend for is the one doing the work, so
// fall through and re-read — `POLL_INTERVAL` is the backoff.
rust/lancedb/src/table/lsm_stats.rs
+7 -7
View File
@@ -1,7 +1,7 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
//! Live per-table_shard LSM state — the shape [`crate::Table::get_lsm_stats`]
//! Live per-table-shard LSM state — the shape [`crate::Table::get_lsm_stats`]
//! returns and [`super::checkpoint`] polls.
//!
//! Nothing here is derived: sums and differences (total SSTable bytes, WAL lag)
@@ -34,8 +34,8 @@ pub struct MemtableStats {
pub indexes: Vec<String>,
}
/// Live state of one table_shard. A table is N table_shards on one node; flattening to
/// a single number hides the one hot table_shard that is usually why someone
/// Live state of one table shard. A table is N table shards on one node; flattening to
/// a single number hides the one hot table shard that is usually why someone
/// opened this endpoint.
#[derive(Debug, Clone, Deserialize)]
pub struct TableShardStats {
@@ -48,12 +48,12 @@ pub struct TableShardStats {
pub replay_after_wal_entry_position: u64,
pub wal_entry_position_last_seen: u64,
pub sstables: Vec<SsTableStats>,
/// Whether a pass owns this table_shard's compaction latch right now. Says *a*
/// Whether a pass owns this table shard's compaction latch right now. Says *a*
/// driver is running, not *whose*, and the latch is held from dispatch —
/// including while the pass queues for a pod-wide compactor permit. Read
/// it as "do not pile on", never as "mine is progressing".
pub compacting: bool,
/// Oldest first, active last. Absent for a `Sealed` table_shard, whose
/// Oldest first, active last. Absent for a `Sealed` table shard, whose
/// in-memory state is torn down.
#[serde(default)]
pub memtables: Option<Vec<MemtableStats>>,
@@ -79,7 +79,7 @@ impl TableShardStats {
}
}
/// Live LSM state, one entry per table_shard.
/// Live LSM state, one entry per table shard.
#[derive(Debug, Clone, Deserialize)]
pub struct LsmStats {
pub table_shards: Vec<TableShardStats>,
@@ -146,7 +146,7 @@ mod tests {
}
/// The metric counts SSTables, not table shards: a pass drains a bounded
/// prefix, so one table_shard going 3 → 2 → 1 → 0 is three steps.
/// prefix, so one table shard going 3 → 2 → 1 → 0 is three steps.
#[test]
fn progress_is_measured_in_sstables() {
let target = 3;
rust/lancedb/src/table/merge.rs
+26 -26
View File
@@ -1429,7 +1429,7 @@ mod lsm_tests {
.await
.unwrap();
// Two single-row upserts that route to (likely) different buckets; each
// Two single-row upserts that route to (likely) different shards; each
// closes the writer so the next opens a fresh shard.
lsm_upsert(&table, vec![10]).await;
table.close_lsm_writers().await.unwrap();
@@ -1681,7 +1681,7 @@ mod lsm_tests {
/// A row written before the index exists is searchable once it does.
///
/// The MemTable holding it was built without the index. Creating the index
/// replaces the writer's configs, which seals that MemTable and waits for
/// replaces the writer's configs, which freezes that MemTable and waits for
/// its flush, so no resident MemTable is left that cannot answer the read.
#[tokio::test]
async fn lsm_a_row_written_before_the_index_is_searchable_after_it() {
@@ -2041,8 +2041,8 @@ mod lsm_tests {
}
/// A kind Lance has no memtable index for is refused when named until the
/// table's registry maintains it; then it is maintained, flushed with each
/// generation, and answers searches on the flushed generations.
/// table's registry maintains it; then it is maintained, flushed into each
/// SSTable, and answers searches on the SSTables.
#[tokio::test]
async fn lsm_maintains_an_index_kind_from_the_tables_registry() {
let dir = tempdir().unwrap();
@@ -2138,11 +2138,11 @@ mod lsm_tests {
table.close_lsm_writers().await.unwrap();
assert_eq!(fresh().await, vec![10, 11]);
assert_generations_answer_through_region_bitmap(&table, vec![10, 11]).await;
assert_sstables_answer_through_region_bitmap(&table, vec![10, 11]).await;
}
/// The generations each shard's manifest publishes.
async fn published_generations(table: &Table) -> Vec<lance::Dataset> {
/// The SSTables each shard's manifest publishes.
async fn published_sstables(table: &Table) -> Vec<lance::Dataset> {
use lance::dataset::mem_wal::ShardManifestStore;
let uri = table.uri().await.unwrap();
@@ -2150,7 +2150,7 @@ mod lsm_tests {
.await
.unwrap();
let mem_wal = std::path::Path::new(&uri).join("_mem_wal");
let mut generations = Vec::new();
let mut sstables = Vec::new();
for shard in std::fs::read_dir(&mem_wal).unwrap() {
let Ok(shard_id) = shard.unwrap().file_name().to_string_lossy().parse() else {
continue;
@@ -2162,30 +2162,30 @@ mod lsm_tests {
.expect("a shard directory has a manifest");
for sstable in manifest.sstables {
let path = mem_wal.join(shard_id.to_string()).join(&sstable.path);
generations.push(lance::Dataset::open(path.to_str().unwrap()).await.unwrap());
sstables.push(lance::Dataset::open(path.to_str().unwrap()).await.unwrap());
}
}
generations
sstables
}
/// Every published generation carries `region_bitmap` as a bitmap index on
/// Every published SSTable carries `region_bitmap` as a bitmap index on
/// `region`, and a search for `region = 'fresh'` on it uses that index.
async fn assert_generations_answer_through_region_bitmap(table: &Table, fresh: Vec<i64>) {
async fn assert_sstables_answer_through_region_bitmap(table: &Table, fresh: Vec<i64>) {
use lance::index::DatasetIndexExt;
let generations = published_generations(table).await;
let sstables = published_sstables(table).await;
assert!(
!generations.is_empty(),
"closing the writer flushed a generation"
!sstables.is_empty(),
"closing the writer flushed an SSTable"
);
let mut found = Vec::new();
for generation in generations {
let region = generation.schema().field("region").unwrap().id;
let indices = generation.load_indices().await.unwrap();
for sstable in sstables {
let region = sstable.schema().field("region").unwrap().id;
let indices = sstable.load_indices().await.unwrap();
let index = indices
.iter()
.find(|index| index.name == "region_bitmap")
.expect("the generation carries the index the plugin trained");
.expect("the SSTable carries the index the plugin trained");
assert_eq!(index.fields, vec![region]);
assert!(
index
@@ -2196,7 +2196,7 @@ mod lsm_tests {
index.index_details
);
let mut scan = generation.scan();
let mut scan = sstable.scan();
scan.filter("region = 'fresh'").unwrap();
let plan = scan.explain_plan(false).await.unwrap();
assert!(plan.contains("ScalarIndexQuery"), "{plan}");
@@ -2288,12 +2288,12 @@ mod lsm_tests {
);
table.close_lsm_writers().await.unwrap();
// Only the generation written after the index exists carries it.
let generations = published_generations(&table).await;
// Only the SSTable written after the index exists carries it.
let sstables = published_sstables(&table).await;
let mut carrying = 0;
for generation in &generations {
for sstable in &sstables {
use lance::index::DatasetIndexExt;
if generation
if sstable
.load_indices()
.await
.unwrap()
@@ -2305,8 +2305,8 @@ mod lsm_tests {
}
assert!(
carrying >= 1,
"{} generations, none carry the index",
generations.len()
"{} SSTables, none carry the index",
sstables.len()
);
}
}
rust/lancedb/src/table/merge/lsm.rs
+7 -7
View File
@@ -370,7 +370,7 @@ impl ShardWriterEntry {
}
/// The cached writer's latest in-memory manifest (current generation +
/// flushed generations). `Ok(None)` if the writer was already closed.
/// SSTables). `Ok(None)` if the writer was already closed.
/// Used by the LSM read path to snapshot this shard authoritatively
/// without re-reading the on-disk manifest.
async fn manifest(&self) -> Result<Option<lance_index::mem_wal::ShardManifest>> {
@@ -449,7 +449,7 @@ impl ShardWriterCache {
/// writer opened afterwards reads the committed set.
///
/// Holds the entry's read lock across the refresh, so a concurrent close
/// waits rather than retiring the writer mid-seal. Writes take the same
/// waits rather than retiring the writer mid-freeze. Writes take the same
/// lock first, so the order matches theirs.
pub(crate) async fn refresh_maintained_indexes(&self, dataset: &Dataset) {
let entry = {
@@ -463,13 +463,13 @@ impl ShardWriterCache {
let Some(writer) = guard.as_ref() else {
return;
};
// Replacing the configs seals the active memtable so the next one carries
// Replacing the configs freezes the active memtable so the next one carries
// the new indexes. The outgoing one keeps the set it was built with, so
// its flush has to land before a read: while it is still frozen, an
// indexed read sees a resident memtable that cannot answer and is
// refused.
let sealed = match dataset.refresh_mem_wal_index_configs(writer).await {
Ok(sealed) => sealed,
let frozen = match dataset.refresh_mem_wal_index_configs(writer).await {
Ok(frozen) => frozen,
Err(error) => {
log::warn!(
"the maintained index set changed but the open shard writer could \
@@ -478,12 +478,12 @@ impl ShardWriterCache {
return;
}
};
if let Some(fence) = sealed
if let Some(fence) = frozen
&& let Err(error) = fence.wait().await
{
log::warn!(
"the shard writer took the new maintained index set, but the memtable \
it sealed was not flushed, so an indexed read stays refused until \
it froze was not flushed, so an indexed read stays refused until \
it is: {error}"
);
}
rust/lancedb/src/table/query.rs
+1 -1
View File
@@ -172,7 +172,7 @@ pub async fn create_plan(
// * Some(true) — force LSM routing; error if the table has no write spec
// * Some(false) — read the base table only, bypassing the MemWAL
// The LSM scanner surfaces in-flight `merge_insert` data (active/frozen
// memtables + flushed generations); validation and dispatch live in `lsm`.
// memtables + SSTables); validation and dispatch live in `lsm`.
let has_spec = ds_ref.mem_wal_index_details().await?.is_some();
let use_lsm = match query.base.use_lsm {
Some(true) if !has_spec => {
rust/lancedb/src/table/query/lsm.rs
+3 -3
View File
@@ -8,7 +8,7 @@
//! base-table scan unless the query sets
//! [`use_lsm(false)`](crate::query::QueryBase::use_lsm). This makes data
//! written via the LSM `merge_insert` path — which lives in the active/frozen
//! in-memory memtables and the flushed SSTable generations until an external
//! in-memory memtables and the SSTables until an external
//! compaction merges it into the base table — visible to queries, deduplicated by
//! primary key (newest generation wins).
//!
@@ -275,7 +275,7 @@ fn exclusion_watermarks(
exclude
}
/// Assemble the per-shard snapshots (flushed SSTable generations) and the
/// Assemble the per-shard snapshots (SSTables) and the
/// in-memory memtables (active + frozen) for the table.
///
/// Snapshots for all shards come from their on-disk manifests; for the shard
@@ -855,7 +855,7 @@ mod tests {
fn exclusion_watermark_gates_on_lagging_index_catchup() {
let shard = Uuid::from_u128(1);
let details = MemWalIndexDetails {
// Compaction has drained generations through 5 into the base table...
// Compaction has drained SSTables through generation 5 into the base table...
compacted_sstables: vec![CompactedSsTable::new(shard, 5)],
// ...but the FTS index has only caught up through generation 2.
index_catchup: vec![IndexCatchupProgress::new(