feat(datafusion): let BaseTableAdapter set the MemWAL read routing

`BaseTableAdapter::scan` built its `QueryRequest` with `use_lsm` unset, so a
SQL scan of a table with a MemWAL write spec always planned lancedb's own LSM
read. A caller that serves the fresh tier another way had no way to ask for a
base-only scan.

Add `with_use_lsm(Option<bool>)`, carried through `with_fts_query` and into
every scan's `QueryRequest`. The default stays `None`, so existing callers are
unchanged.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Daniel Rammer
2026-10-08 15:42:19 -05:00
co-authored by Claude Opus 5.5
parent d3390a9ca7
commit d479a3f207
2 changed files with 45 additions and 0 deletions
+10
View File
@@ -166,6 +166,7 @@ pub struct BaseTableAdapter {
table: Arc<dyn BaseTable>,
schema: Arc<ArrowSchema>,
fts_query: Option<FullTextSearchQuery>,
use_lsm: Option<bool>,
}
impl BaseTableAdapter {
@@ -181,9 +182,16 @@ impl BaseTableAdapter {
table,
schema: Arc::new(schema),
fts_query: None,
use_lsm: None,
})
}
/// Set the MemWAL read routing of every scan; see [`QueryRequest::use_lsm`].
pub fn with_use_lsm(mut self, use_lsm: Option<bool>) -> Self {
self.use_lsm = use_lsm;
self
}
/// Create a new adapter with an FTS query applied.
pub fn with_fts_query(&self, fts_query: FullTextSearchQuery) -> Self {
// Add _score column to the schema
@@ -196,6 +204,7 @@ impl BaseTableAdapter {
table: self.table.clone(),
schema,
fts_query: Some(fts_query),
use_lsm: self.use_lsm,
}
}
}
@@ -223,6 +232,7 @@ impl TableProvider for BaseTableAdapter {
let mut query = QueryRequest {
full_text_search: self.fts_query.clone(),
disable_scoring_autoprojection: disable_scoring,
use_lsm: self.use_lsm,
..Default::default()
};
+35
View File
@@ -1334,6 +1334,41 @@ mod lsm_tests {
);
}
#[tokio::test]
async fn base_table_adapter_use_lsm_false_scans_the_base_only() {
use crate::table::datafusion::BaseTableAdapter;
use datafusion::prelude::SessionContext;
let dir = tempdir().unwrap();
let table = id_value_table(&dir).await; // base: ids 1,2,3
table
.set_lsm_write_spec(LsmWriteSpec::unsharded())
.await
.unwrap();
// Ids 4,5 land in the active memtable, not the base table.
lsm_upsert(&table, vec![4, 5]).await;
let adapter = BaseTableAdapter::try_new(table.base_table().clone())
.await
.unwrap()
.with_use_lsm(Some(false));
let ctx = SessionContext::new();
ctx.register_table("base_only", Arc::new(adapter)).unwrap();
let batches = ctx
.sql("SELECT id FROM base_only")
.await
.unwrap()
.collect()
.await
.unwrap();
let mut ids: Vec<i64> = batches
.iter()
.flat_map(|b| b.column(0).as_primitive::<Int64Type>().values().to_vec())
.collect();
ids.sort();
assert_eq!(ids, vec![1, 2, 3]);
}
#[tokio::test]
async fn query_snapshot_preserves_lsm_read_semantics() {
let dir = tempdir().unwrap();