rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-11-02 08:46:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(query): drop what the fusion fallback does not need

Browse Source 
Four things the optimistic rework left behind:

`BackgroundCache` was the wrong type for remembering a refusal. Only
`try_get`/`seed`/`invalidate` were ever called, so its background-refresh
machinery was unreachable and the refresh window served only to shorten the
TTL it was paired with -- 30s of intent, 25s of behaviour. A
`Mutex<Option<Instant>>` says what this is.

Invalidating the memory after *installing* a write spec cleared a bit that is
necessarily unset already; the refusal that follows is what sets it. The
`unset` side is real and stays.

`LanceTable` overrode the learned-bit accessors with the same body as the
remote table, but a local table is never refused, so the base class default
already answers for it.

`drop_columns` was ten lines for one call site that `RecordBatch::drop_column`
covers in five, with the same tolerance for a reranker that reshaped its
output.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Daniel Rammer
2026-09-23 13:15:27 -05:00
co-authored by Claude Opus 5
parent a2d4597689
commit 90b99af20f
3 changed files with 26 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/python/lancedb/table.py
-7
View File
@@ -4565,13 +4565,6 @@ class LanceTable(Table):
[`AsyncTable.get_lsm_write_spec`][lancedb.AsyncTable.get_lsm_write_spec]."""
return LOOP.run(self._table.get_lsm_write_spec())
def _hybrid_pk_fusion_learned(self) -> bool:
# `self._table` is the async table, which owns the Rust handle.
return self._table._hybrid_pk_fusion_learned()
def _note_hybrid_pk_fusion(self) -> None:
self._table._note_hybrid_pk_fusion()
def checkpoint_lsm(self) -> None:
"""Synchronous version of
[`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
rust/lancedb/src/query.rs
+7 -14
View File
@@ -1258,19 +1258,6 @@ fn inject_pk_columns(select: &mut Select, pk_columns: &[String]) {
}
}
/// Drop `columns` from `batch`, ignoring any that are not present.
fn drop_columns(batch: RecordBatch, columns: &[String]) -> Result<RecordBatch> {
let keep: Vec<usize> = batch
.schema()
.fields()
.iter()
.enumerate()
.filter(|(_, field)| !columns.iter().any(|name| name == field.name()))
.map(|(idx, _)| idx)
.collect();
Ok(batch.project(&keep)?)
}
impl VectorQuery {
fn new(base: Query) -> Self {
Self {
@@ -1633,7 +1620,13 @@ impl VectorQuery {
// has already refused the query if they did.
FusionKey::PrimaryKey { injected, .. } => {
results = results.drop_column(ROW_ID)?;
results = drop_columns(results, injected)?;
for column in injected {
// A reranker is free to reshape its output, so only drop
// what actually came back.
if results.schema().column_with_name(column).is_some() {
results = results.drop_column(column)?;
}
}
}
}
rust/lancedb/src/remote/table.rs
+19 -31
View File
@@ -79,7 +79,7 @@ use std::collections::{HashMap, HashSet};
use std::io::Cursor;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};
use std::time::{Duration, Instant, SystemTime};
use tokio::sync::RwLock;
const REQUEST_TIMEOUT_HEADER: HeaderName = HeaderName::from_static("x-request-timeout-ms");
@@ -91,12 +91,13 @@ const METRIC_TYPE_KEY: &str = "metric_type";
const INDEX_TYPE_KEY: &str = "index_type";
const SCHEMA_CACHE_TTL: Duration = Duration::from_secs(30);
const SCHEMA_CACHE_REFRESH_WINDOW: Duration = Duration::from_secs(5);
/// How long a hybrid query remembers that this table refused `_rowid`. Nothing
/// fetches this — it is seeded by the refusal itself — so the TTL only bounds
/// how long the memory outlives a write spec someone removed, and being late to
/// notice costs a base-only read that is still correct.
/// How long a hybrid query remembers that this table refused `_rowid`.
///
/// Nothing fetches this — the refusal seeds it — so the bound exists only to
/// stop the memory outliving a write spec someone removed through another
/// handle. Being late to notice keeps the query on the primary key, which on a
/// base table is still a valid read; the window is what stops that lasting.
const WAL_FUSION_MEMORY_TTL: Duration = Duration::from_secs(30);
const WAL_FUSION_MEMORY_REFRESH_WINDOW: Duration = Duration::from_secs(5);
const SCHEMA_SELECTOR_CHANGED: &str = "table selector changed while fetching schema";
fn fts_query_requires_document_granularity_support(query: &FtsQuery) -> bool {
@@ -503,7 +504,7 @@ pub struct RemoteTable<S: HttpSend = Sender> {
version: Arc<RwLock<Option<u64>>>,
location: RwLock<Option<String>>,
schema_cache: BackgroundCache<SchemaRef, Error>,
wal_pk_fusion: BackgroundCache<bool, Error>,
wal_pk_fusion: Mutex<Option<Instant>>,
freshness: Arc<Mutex<FreshnessState>>,
/// The branch this handle is scoped to, or `None` for the main branch.
/// Stamped onto every branch-accepting request so reads and writes resolve
@@ -672,10 +673,7 @@ impl<S: HttpSend> RemoteTable<S> {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: BackgroundCache::new(
WAL_FUSION_MEMORY_TTL,
WAL_FUSION_MEMORY_REFRESH_WINDOW,
),
wal_pk_fusion: Mutex::new(None),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch: None,
}
@@ -709,10 +707,7 @@ impl<S: HttpSend> RemoteTable<S> {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: BackgroundCache::new(
WAL_FUSION_MEMORY_TTL,
WAL_FUSION_MEMORY_REFRESH_WINDOW,
),
wal_pk_fusion: Mutex::new(None),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch,
}
@@ -1448,7 +1443,7 @@ impl<S: HttpSend> RemoteTable<S> {
}
fn invalidate_wal_pk_fusion(&self) {
self.wal_pk_fusion.invalidate();
*self.wal_pk_fusion.lock().unwrap() = None;
}
fn handle_error_invalidation(&self, error: &Error) {
@@ -1634,10 +1629,7 @@ mod test_utils {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: BackgroundCache::new(
WAL_FUSION_MEMORY_TTL,
WAL_FUSION_MEMORY_REFRESH_WINDOW,
),
wal_pk_fusion: Mutex::new(None),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch: None,
}
@@ -1662,10 +1654,7 @@ mod test_utils {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: BackgroundCache::new(
WAL_FUSION_MEMORY_TTL,
WAL_FUSION_MEMORY_REFRESH_WINDOW,
),
wal_pk_fusion: Mutex::new(None),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch: None,
}
@@ -1699,10 +1688,7 @@ mod test_utils {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: BackgroundCache::new(
WAL_FUSION_MEMORY_TTL,
WAL_FUSION_MEMORY_REFRESH_WINDOW,
),
wal_pk_fusion: Mutex::new(None),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch: None,
}
@@ -3253,7 +3239,6 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
.json(&body);
let (request_id, response) = self.send(request, true).await?;
self.check_table_response(&request_id, response).await?;
self.invalidate_wal_pk_fusion();
Ok(())
}
@@ -3333,11 +3318,14 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
}
fn hybrid_pk_fusion_learned(&self) -> bool {
self.wal_pk_fusion.try_get().unwrap_or(false)
self.wal_pk_fusion
.lock()
.unwrap()
.is_some_and(|learned| learned.elapsed() < WAL_FUSION_MEMORY_TTL)
}
fn note_hybrid_pk_fusion(&self) {
self.wal_pk_fusion.seed(true);
*self.wal_pk_fusion.lock().unwrap() = Some(Instant::now());
}
async fn tags(&self) -> Result<Box<dyn Tags + '_>> {