mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-14 14:46:53 +00:00
refactor(table): shrink LsmStats to measured per-bucket fields
`get_lsm_stats` reported four table-level rollups, three per-bucket derived fields, and two pieces of in-process state. All of it is either arithmetic the caller can do on the per-bucket list (`bucket_count`, `generations_total`, `l0_bytes_total`, `memtable_rows_total`, `l0_bytes`, `wal_lag`) or a value that is true only for the microsecond it was read (`fenced`, `compaction_in_progress`). Drop them. What is left is measured: manifest metadata, one entry per flushed L0 generation, and one entry per in-memory memtable. Three shape changes follow: * `active_memtable` + `frozen_memtables` collapse into `memtables`, ordered oldest first with the active one last. Absent — not empty — for a `Sealed` bucket, whose writer is torn down. * Generation `rows` is unconditional, so `LsmStatsOptions` and its `include_generation_rows` flag are gone. A row count that had to be asked for was routinely absent from the one report that needed it. * `memtable_indexes` becomes a list of names on each memtable, since each carries its own index store and a frozen one predates an index created after it froze. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
d26139417e
commit
c09e808dab
@@ -292,7 +292,7 @@ class Table:
|
||||
) -> dict: ...
|
||||
async def flush_lsm(self) -> dict: ...
|
||||
async def compact_lsm(self, max_generations_per_bucket: Optional[int]) -> dict: ...
|
||||
async def get_lsm_stats(self, include_generation_rows: bool) -> Optional[dict]: ...
|
||||
async def get_lsm_stats(self) -> Optional[dict]: ...
|
||||
async def close_lsm_writers(self) -> None: ...
|
||||
@property
|
||||
def tags(self) -> Tags: ...
|
||||
|
||||
@@ -3903,12 +3903,10 @@ class LanceTable(Table):
|
||||
)
|
||||
)
|
||||
|
||||
def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
|
||||
def get_lsm_stats(self) -> Optional[dict]:
|
||||
"""Synchronous version of
|
||||
[`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
|
||||
return LOOP.run(
|
||||
self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
|
||||
)
|
||||
return LOOP.run(self._table.get_lsm_stats())
|
||||
|
||||
def close_lsm_writers(self) -> None:
|
||||
"""Close cached MemWAL shard writers. See
|
||||
@@ -4676,10 +4674,8 @@ class AsyncTable:
|
||||
"""
|
||||
return await self._inner.compact_lsm(max_generations_per_bucket)
|
||||
|
||||
async def get_lsm_stats(
|
||||
self, *, include_generation_rows: bool = False
|
||||
) -> Optional[dict]:
|
||||
"""Read live LSM state: per-bucket detail plus a table-level aggregate.
|
||||
async def get_lsm_stats(self) -> Optional[dict]:
|
||||
"""Read live per-bucket LSM state.
|
||||
|
||||
Answers "how far behind is my fresh tier", "which bucket is hot", and
|
||||
"why is my fresh-tier vector search brute-force". Mutates no table
|
||||
@@ -4688,14 +4684,8 @@ class AsyncTable:
|
||||
|
||||
Returns ``None`` — and only — when the LSM write path is not enabled
|
||||
for the table.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
include_generation_rows
|
||||
Report row counts per L0 generation. Off by default: each costs
|
||||
one Lance manifest read.
|
||||
"""
|
||||
return await self._inner.get_lsm_stats(include_generation_rows)
|
||||
return await self._inner.get_lsm_stats()
|
||||
|
||||
async def close_lsm_writers(self) -> None:
|
||||
"""Drain and close any cached MemWAL shard writers for this table.
|
||||
|
||||
+11
-50
@@ -57,58 +57,30 @@ fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult
|
||||
"wal_entry_position_last_seen",
|
||||
b.wal_entry_position_last_seen,
|
||||
)?;
|
||||
e.set_item("wal_lag", b.wal_lag)?;
|
||||
e.set_item("l0_bytes", b.l0_bytes)?;
|
||||
e.set_item("fenced", b.fenced)?;
|
||||
e.set_item("compaction_in_progress", b.compaction_in_progress)?;
|
||||
|
||||
let generations = PyList::empty(py);
|
||||
for g in &b.generations {
|
||||
let ge = PyDict::new(py);
|
||||
ge.set_item("generation", g.generation)?;
|
||||
ge.set_item("path", &g.path)?;
|
||||
ge.set_item("bytes", g.bytes)?;
|
||||
ge.set_item("rows", g.rows)?;
|
||||
generations.append(ge)?;
|
||||
}
|
||||
e.set_item("generations", generations)?;
|
||||
|
||||
let memtable = |m: &lancedb::table::MemtableStats| -> PyResult<Py<PyDict>> {
|
||||
let d = PyDict::new(py);
|
||||
d.set_item("generation", m.generation)?;
|
||||
d.set_item("rows", m.rows)?;
|
||||
d.set_item("bytes", m.bytes)?;
|
||||
d.set_item("batches", m.batches)?;
|
||||
Ok(d.unbind())
|
||||
};
|
||||
e.set_item(
|
||||
"active_memtable",
|
||||
b.active_memtable.as_ref().map(memtable).transpose()?,
|
||||
)?;
|
||||
e.set_item(
|
||||
"frozen_memtables",
|
||||
b.frozen_memtables
|
||||
"memtables",
|
||||
b.memtables
|
||||
.as_ref()
|
||||
.map(|fs| {
|
||||
.map(|ms| {
|
||||
let l = PyList::empty(py);
|
||||
for m in fs {
|
||||
l.append(memtable(m)?)?;
|
||||
}
|
||||
PyResult::Ok(l.unbind())
|
||||
})
|
||||
.transpose()?,
|
||||
)?;
|
||||
e.set_item(
|
||||
"memtable_indexes",
|
||||
b.memtable_indexes
|
||||
.as_ref()
|
||||
.map(|ixs| {
|
||||
let l = PyList::empty(py);
|
||||
for i in ixs {
|
||||
for m in ms {
|
||||
let d = PyDict::new(py);
|
||||
d.set_item("name", &i.name)?;
|
||||
d.set_item("kind", &i.kind)?;
|
||||
d.set_item("column", &i.column)?;
|
||||
d.set_item("generation", m.generation)?;
|
||||
d.set_item("rows", m.rows)?;
|
||||
d.set_item("bytes", m.bytes)?;
|
||||
d.set_item("batches", m.batches)?;
|
||||
d.set_item("indexes", m.indexes.clone())?;
|
||||
l.append(d)?;
|
||||
}
|
||||
PyResult::Ok(l.unbind())
|
||||
@@ -118,10 +90,6 @@ fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult
|
||||
buckets.append(e)?;
|
||||
}
|
||||
out.set_item("buckets", buckets)?;
|
||||
out.set_item("bucket_count", stats.bucket_count)?;
|
||||
out.set_item("generations_total", stats.generations_total)?;
|
||||
out.set_item("l0_bytes_total", stats.l0_bytes_total)?;
|
||||
out.set_item("memtable_rows_total", stats.memtable_rows_total)?;
|
||||
Ok(out.unbind())
|
||||
}
|
||||
|
||||
@@ -1473,17 +1441,10 @@ impl Table {
|
||||
}
|
||||
|
||||
/// Live LSM state, or `None` when the LSM write path is not enabled.
|
||||
#[pyo3(signature = (include_generation_rows=false))]
|
||||
pub fn get_lsm_stats(
|
||||
self_: PyRef<'_, Self>,
|
||||
include_generation_rows: bool,
|
||||
) -> PyResult<Bound<'_, PyAny>> {
|
||||
pub fn get_lsm_stats(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.inner_ref()?.clone();
|
||||
let opts = lancedb::table::LsmStatsOptions {
|
||||
include_generation_rows,
|
||||
};
|
||||
future_into_py(self_.py(), async move {
|
||||
let stats = inner.get_lsm_stats(opts).await.infer_error()?;
|
||||
let stats = inner.get_lsm_stats().await.infer_error()?;
|
||||
Python::attach(|py| stats.map(|s| lsm_stats_to_py(py, &s)).transpose())
|
||||
})
|
||||
}
|
||||
|
||||
@@ -31,7 +31,7 @@ use crate::table::merge::MergeFilter;
|
||||
use crate::table::query::create_multi_vector_plan;
|
||||
use crate::table::{AlterColumnsResult, FieldMetadataUpdate, UpdateFieldMetadataResult};
|
||||
use crate::table::{AnyQuery, Filter, Predicate, PreprocessingOutput, TableStatistics};
|
||||
use crate::table::{CompactOptions, CompactReport, FlushReport, LsmStats, LsmStatsOptions};
|
||||
use crate::table::{CompactOptions, CompactReport, FlushReport, LsmStats};
|
||||
use crate::utils::background_cache::BackgroundCache;
|
||||
use crate::utils::{
|
||||
resolve_arrow_field_path, supported_btree_data_type, supported_vector_data_type,
|
||||
@@ -2418,13 +2418,9 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
|
||||
self.send_lsm_route(request, "compact_lsm").await
|
||||
}
|
||||
|
||||
async fn get_lsm_stats(&self, opts: LsmStatsOptions) -> Result<Option<LsmStats>> {
|
||||
async fn get_lsm_stats(&self) -> Result<Option<LsmStats>> {
|
||||
// Read-semantics POST, like `get_lsm_write_spec`.
|
||||
let request = self
|
||||
.post_read(&format!("/v1/table/{}/get_lsm_stats/", self.identifier))
|
||||
.json(&serde_json::json!({
|
||||
"include_generation_rows": opts.include_generation_rows,
|
||||
}));
|
||||
let request = self.post_read(&format!("/v1/table/{}/get_lsm_stats/", self.identifier));
|
||||
let parsed: GetLsmStatsResponse = self.send_lsm_route(request, "get_lsm_stats").await?;
|
||||
// `null` — and only — when the table has no LSM write path.
|
||||
Ok(parsed.lsm_stats)
|
||||
@@ -6009,18 +6005,11 @@ mod tests {
|
||||
}
|
||||
|
||||
/// WAL off ⇒ `None`; WAL on ⇒ a fully populated `Some` with no field
|
||||
/// defaulting to a zero it did not measure. `include_generation_rows`
|
||||
/// is off by default.
|
||||
/// defaulting to a zero it did not measure.
|
||||
#[tokio::test]
|
||||
async fn test_get_lsm_stats_round_trip() {
|
||||
let table = Table::new_with_handler("my_table", |request| {
|
||||
assert_eq!(request.url().path(), "/v1/table/my_table/get_lsm_stats/");
|
||||
let body = request.body().unwrap().as_bytes().unwrap();
|
||||
let body: serde_json::Value = serde_json::from_slice(body).unwrap();
|
||||
assert_eq!(
|
||||
body["include_generation_rows"], false,
|
||||
"generation rows cost a manifest read each and must be opt-in"
|
||||
);
|
||||
let response = serde_json::json!({
|
||||
"lsm_stats": {
|
||||
"buckets": [{
|
||||
@@ -6031,21 +6020,12 @@ mod tests {
|
||||
"current_generation": 9,
|
||||
"replay_after_wal_entry_position": 100,
|
||||
"wal_entry_position_last_seen": 140,
|
||||
"wal_lag": 40,
|
||||
"generations": [{ "generation": 8, "path": "ab_gen_8", "bytes": 4096 }],
|
||||
"l0_bytes": 4096,
|
||||
"fenced": false,
|
||||
"compaction_in_progress": false,
|
||||
"active_memtable": { "generation": 9, "rows": 12, "bytes": 900, "batches": 2 },
|
||||
"frozen_memtables": [],
|
||||
"memtable_indexes": [
|
||||
{ "name": "vec_idx", "kind": "hnsw", "column": "vector" }
|
||||
"generations": [{ "generation": 8, "bytes": 4096, "rows": 30 }],
|
||||
"memtables": [
|
||||
{ "generation": 9, "rows": 12, "bytes": 900, "batches": 2,
|
||||
"indexes": ["vec_idx"] }
|
||||
],
|
||||
}],
|
||||
"bucket_count": 1,
|
||||
"generations_total": 1,
|
||||
"l0_bytes_total": 4096,
|
||||
"memtable_rows_total": 12,
|
||||
}
|
||||
});
|
||||
http::Response::builder()
|
||||
@@ -6055,25 +6035,19 @@ mod tests {
|
||||
});
|
||||
|
||||
let stats = table
|
||||
.get_lsm_stats(crate::table::LsmStatsOptions::default())
|
||||
.get_lsm_stats()
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("a WAL-backed table reports Some");
|
||||
assert_eq!(stats.bucket_count, 1);
|
||||
assert_eq!(stats.l0_bytes_total, 4096);
|
||||
assert_eq!(stats.memtable_rows_total, 12);
|
||||
let bucket = &stats.buckets[0];
|
||||
assert_eq!(bucket.wal_lag, 40);
|
||||
assert_eq!(bucket.replay_after_wal_entry_position, 100);
|
||||
assert_eq!(bucket.wal_entry_position_last_seen, 140);
|
||||
assert_eq!(bucket.generations[0].generation, 8);
|
||||
assert!(
|
||||
bucket.generations[0].rows.is_none(),
|
||||
"rows are absent unless asked for"
|
||||
);
|
||||
// The row that answers "why is my fresh-tier vector search
|
||||
// brute-force" — an absent hnsw entry is the whole explanation.
|
||||
let indexes = bucket.memtable_indexes.as_ref().unwrap();
|
||||
assert_eq!(indexes[0].kind, "hnsw");
|
||||
assert_eq!(indexes[0].column, "vector");
|
||||
assert_eq!(bucket.generations[0].rows, 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();
|
||||
assert_eq!(memtables[0].indexes, vec!["vec_idx".to_string()]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -6084,13 +6058,7 @@ mod tests {
|
||||
.body(serde_json::json!({ "lsm_stats": null }).to_string())
|
||||
.unwrap()
|
||||
});
|
||||
assert!(
|
||||
table
|
||||
.get_lsm_stats(crate::table::LsmStatsOptions::default())
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
assert!(table.get_lsm_stats().await.unwrap().is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -86,7 +86,7 @@ pub use branch_merge::{
|
||||
pub use checkpoint::{
|
||||
BucketStats, CheckpointOptions, CheckpointReport, CheckpointStopReason, CompactBucketReport,
|
||||
CompactOptions, CompactReport, FlushBucketReport, FlushReport, GenerationStats, LsmFault,
|
||||
LsmStats, LsmStatsOptions, MemIndexStats, MemtableStats,
|
||||
LsmStats, MemtableStats,
|
||||
};
|
||||
use checkpoint::{CheckpointControl, checkpoint_fault};
|
||||
pub use chrono::Duration;
|
||||
@@ -654,7 +654,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
|
||||
/// enabled for this table.
|
||||
///
|
||||
/// The default implementation returns `NotSupported`.
|
||||
async fn get_lsm_stats(&self, _opts: LsmStatsOptions) -> Result<Option<LsmStats>> {
|
||||
async fn get_lsm_stats(&self) -> Result<Option<LsmStats>> {
|
||||
Err(Error::NotSupported {
|
||||
message: "get_lsm_stats is not supported on this table type".into(),
|
||||
})
|
||||
@@ -1760,7 +1760,7 @@ impl Table {
|
||||
}
|
||||
|
||||
if opts.include_stats {
|
||||
report.stats = self.get_lsm_stats(LsmStatsOptions::default()).await?;
|
||||
report.stats = self.get_lsm_stats().await?;
|
||||
}
|
||||
Ok(report)
|
||||
}
|
||||
@@ -1787,8 +1787,7 @@ impl Table {
|
||||
self.inner.compact_lsm(opts).await
|
||||
}
|
||||
|
||||
/// Read live LSM state: per-bucket fresh-tier detail plus a table-level
|
||||
/// aggregate.
|
||||
/// Read live per-bucket LSM state.
|
||||
///
|
||||
/// Answers "how far behind is my fresh tier", "which bucket is hot",
|
||||
/// and "why is my fresh-tier vector search brute-force". Mutates no
|
||||
@@ -1797,14 +1796,13 @@ impl Table {
|
||||
///
|
||||
/// `Ok(None)` — and only — when the LSM write path is not enabled for
|
||||
/// the table, matching [`Table::get_lsm_write_spec`]. Stats is
|
||||
/// fresh-tier only, so with the WAL off there is no bucket count, no
|
||||
/// manifest, and no watermark; a struct of zeros would read as
|
||||
/// measurements.
|
||||
/// fresh-tier only, so with the WAL off there is no manifest and no
|
||||
/// watermark; a struct of zeros would read as measurements.
|
||||
///
|
||||
/// Do not build a checkpoint's termination on this: the completion
|
||||
/// predicate lives in the `flush` and `compact` responses.
|
||||
pub async fn get_lsm_stats(&self, opts: LsmStatsOptions) -> Result<Option<LsmStats>> {
|
||||
self.inner.get_lsm_stats(opts).await
|
||||
pub async fn get_lsm_stats(&self) -> Result<Option<LsmStats>> {
|
||||
self.inner.get_lsm_stats().await
|
||||
}
|
||||
|
||||
/// Drain and close any cached MemWAL shard writers held for this table.
|
||||
|
||||
@@ -44,7 +44,7 @@ pub struct CheckpointOptions {
|
||||
/// Per-request, per-bucket bound on generations merged. `None` uses the
|
||||
/// server's `compact_prefix_max`. Note this is **per bucket**: an
|
||||
/// N-bucket table admits `N × max_generations_per_bucket` merges per
|
||||
/// request. [`LsmStats::bucket_count`] reports N.
|
||||
/// request. [`LsmStats::buckets`] has one entry per bucket.
|
||||
pub max_generations_per_bucket: Option<usize>,
|
||||
/// Collect a [`LsmStats`] block into the report when the loop ends.
|
||||
pub include_stats: bool,
|
||||
@@ -165,24 +165,12 @@ pub struct CompactReport {
|
||||
pub highest_wal_entry_position: u64,
|
||||
}
|
||||
|
||||
/// Options for [`crate::Table::get_lsm_stats`].
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct LsmStatsOptions {
|
||||
/// Report `rows` per L0 generation. Off by default: the shard manifest
|
||||
/// stores only `{generation, path}`, so rows cost one Lance manifest
|
||||
/// read per generation.
|
||||
pub include_generation_rows: bool,
|
||||
}
|
||||
|
||||
/// One flushed L0 generation.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct GenerationStats {
|
||||
pub generation: u64,
|
||||
pub path: String,
|
||||
pub bytes: u64,
|
||||
/// Present only when [`LsmStatsOptions::include_generation_rows`] was set.
|
||||
#[serde(default)]
|
||||
pub rows: Option<u64>,
|
||||
pub rows: u64,
|
||||
}
|
||||
|
||||
/// One in-memory memtable.
|
||||
@@ -192,22 +180,18 @@ pub struct MemtableStats {
|
||||
pub rows: u64,
|
||||
pub bytes: u64,
|
||||
pub batches: u64,
|
||||
}
|
||||
|
||||
/// An index the memtable carries. An absent `hnsw` entry on a vector
|
||||
/// column is the whole answer to "why is my fresh-tier vector search
|
||||
/// brute-force".
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct MemIndexStats {
|
||||
pub name: String,
|
||||
/// `btree` | `hnsw` | `fts`.
|
||||
pub kind: String,
|
||||
pub column: String,
|
||||
/// Names of the indexes this memtable carries. An absent name is the
|
||||
/// whole answer to "why is my fresh-tier search on that column
|
||||
/// brute-force"; cross-reference `list_indices` for its kind and column.
|
||||
pub indexes: Vec<String>,
|
||||
}
|
||||
|
||||
/// Live state of one bucket. A table is N buckets on one node; flattening
|
||||
/// to a single number hides the one hot bucket that is usually why someone
|
||||
/// opened this endpoint.
|
||||
///
|
||||
/// Nothing here is derived — sums and differences (total L0 bytes, WAL lag)
|
||||
/// are the caller's to compute from measured fields.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct BucketStats {
|
||||
pub shard_id: String,
|
||||
@@ -218,22 +202,14 @@ pub struct BucketStats {
|
||||
pub current_generation: u64,
|
||||
pub replay_after_wal_entry_position: u64,
|
||||
pub wal_entry_position_last_seen: u64,
|
||||
/// Accepted WAL entries not yet covered by a flush.
|
||||
pub wal_lag: u64,
|
||||
pub generations: Vec<GenerationStats>,
|
||||
pub l0_bytes: u64,
|
||||
pub fenced: bool,
|
||||
pub compaction_in_progress: bool,
|
||||
/// Absent for a `Sealed` bucket, whose in-memory state is torn down.
|
||||
/// Oldest first, active last. Absent for a `Sealed` bucket, whose
|
||||
/// in-memory state is torn down.
|
||||
#[serde(default)]
|
||||
pub active_memtable: Option<MemtableStats>,
|
||||
#[serde(default)]
|
||||
pub frozen_memtables: Option<Vec<MemtableStats>>,
|
||||
#[serde(default)]
|
||||
pub memtable_indexes: Option<Vec<MemIndexStats>>,
|
||||
pub memtables: Option<Vec<MemtableStats>>,
|
||||
}
|
||||
|
||||
/// Live LSM state: per-bucket detail plus a table-level aggregate.
|
||||
/// Live LSM state, one entry per bucket.
|
||||
///
|
||||
/// Every field is measured. There is no "WAL is off" shape here — that
|
||||
/// case is `None` from [`crate::Table::get_lsm_stats`], because a struct of
|
||||
@@ -241,10 +217,6 @@ pub struct BucketStats {
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct LsmStats {
|
||||
pub buckets: Vec<BucketStats>,
|
||||
pub bucket_count: usize,
|
||||
pub generations_total: usize,
|
||||
pub l0_bytes_total: u64,
|
||||
pub memtable_rows_total: u64,
|
||||
}
|
||||
|
||||
/// Server-side JSON envelope for `get_lsm_stats`. `lsm_stats` is null when
|
||||
|
||||
Reference in New Issue
Block a user