mirror of
https://github.com/lancedb/lancedb.git
synced 2026-08-18 12:08:35 +00:00
Heng Ge
f655f62e09
## What
MemWAL LSM **read** support. When a table has an LSM write spec
(`set_lsm_write_spec`), `merge_insert` upserts live in the MemWAL
active/frozen memtables and flushed SSTables until an external
compaction merges them into the base table, so a normal scan returns
**stale** data. This routes reads through Lance's `LsmScanner` so
queries also surface that in-flight data, deduplicated by primary key
(newest generation wins).
## How
- Adds a **`use_lsm: Option<bool>`** query flag, symmetric with the
`merge_insert` flag:
- **unset** — auto-route through the LSM scanner when the table carries
a write spec
- **`use_lsm(true)`** — force the LSM path; error if there is no spec
- **`use_lsm(false)`** — read the base table only (the escape hatch)
- Plain scan, single-column full-text search, and single-vector ANN all
run through one `LsmScanner` (assembled from on-disk shard manifests
plus the cached writer's in-memory memtables), so a `where` predicate is
honored as a **prefilter** uniformly — including for vector search.
- **Compaction-aware snapshots:** an SSTable generation is dropped only
once it is both compacted into the base table and covered by the arm's
base-index catch-up (`index_catchup`); plain scans use the compaction
watermark alone.
- Query shapes the scanner cannot honor hard-error with guidance to set
`use_lsm(false)`: hybrid, multi/binary vectors, `with_row_id`,
reranking, `order_by`, dynamic/Substrait projection or filters,
`distance_range`, `use_index(false)`, postfilter, take-by-row-id/offset,
reads from a time-traveled version, and an unmaintained or ambiguous
(multiple) FTS/vector index. Namespace-pushdown queries fall back to
local execution when a spec is present; WAL-only writers are handled.
- Exposed across the Rust core and the Python (`use_lsm`) and TypeScript
(`useLsm`) bindings, including `TakeQuery`.
Rebased from Lance `7.2.0-beta.3` to `10.0.0-beta.3`.
90 lines
2.4 KiB
Rust
90 lines
2.4 KiB
Rust
// SPDX-License-Identifier: Apache-2.0
|
|
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
|
|
|
use std::time::Duration;
|
|
|
|
use lancedb::{ipc::ipc_file_to_batches, table::merge::MergeInsertBuilder};
|
|
use napi::bindgen_prelude::*;
|
|
use napi_derive::napi;
|
|
|
|
use crate::{error::convert_error, table::MergeResult};
|
|
|
|
#[napi]
|
|
#[derive(Clone)]
|
|
/// A builder used to create and run a merge insert operation
|
|
pub struct NativeMergeInsertBuilder {
|
|
pub(crate) inner: MergeInsertBuilder,
|
|
}
|
|
|
|
#[napi]
|
|
impl NativeMergeInsertBuilder {
|
|
#[napi]
|
|
pub fn when_matched_update_all(&self, condition: Option<String>) -> Self {
|
|
let mut this = self.clone();
|
|
this.inner.when_matched_update_all(condition);
|
|
this
|
|
}
|
|
|
|
#[napi]
|
|
pub fn when_not_matched_insert_all(&self) -> Self {
|
|
let mut this = self.clone();
|
|
this.inner.when_not_matched_insert_all();
|
|
this
|
|
}
|
|
#[napi]
|
|
pub fn when_not_matched_by_source_delete(&self, filter: Option<String>) -> Self {
|
|
let mut this = self.clone();
|
|
this.inner.when_not_matched_by_source_delete(filter);
|
|
this
|
|
}
|
|
|
|
#[napi]
|
|
pub fn set_timeout(&mut self, timeout: u32) {
|
|
self.inner.timeout(Duration::from_millis(timeout as u64));
|
|
}
|
|
|
|
#[napi]
|
|
pub fn use_index(&self, use_index: bool) -> Self {
|
|
let mut this = self.clone();
|
|
this.inner.use_index(use_index);
|
|
this
|
|
}
|
|
|
|
#[napi]
|
|
pub fn use_lsm(&self, enable: bool) -> Self {
|
|
let mut this = self.clone();
|
|
this.inner.use_lsm(enable);
|
|
this
|
|
}
|
|
|
|
#[napi]
|
|
pub fn validate_single_shard(&self, validate_single_shard: bool) -> Self {
|
|
let mut this = self.clone();
|
|
this.inner.validate_single_shard(validate_single_shard);
|
|
this
|
|
}
|
|
|
|
#[napi(catch_unwind)]
|
|
pub async fn execute(&self, buf: Buffer) -> napi::Result<MergeResult> {
|
|
let data = ipc_file_to_batches(buf.to_vec()).map_err(|e| {
|
|
napi::Error::from_reason(format!("Failed to read IPC file: {}", convert_error(&e)))
|
|
})?;
|
|
|
|
let this = self.clone();
|
|
|
|
let res = this.inner.execute(data).await.map_err(|e| {
|
|
napi::Error::from_reason(format!(
|
|
"Failed to execute merge insert: {}",
|
|
convert_error(&e)
|
|
))
|
|
})?;
|
|
Ok(res.into())
|
|
}
|
|
}
|
|
|
|
impl From<MergeInsertBuilder> for NativeMergeInsertBuilder {
|
|
fn from(inner: MergeInsertBuilder) -> Self {
|
|
Self { inner }
|
|
}
|
|
}
|