From f1ce09b4be79f20e099ecec09c8826594d679bff Mon Sep 17 00:00:00 2001 From: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Date: Sat, 8 Aug 2026 14:19:29 +0000 Subject: [PATCH] fix(node): preserve auto query state provenance --- nodejs/__test__/table.test.ts | 97 +++++++++++++++++++++++++++++++ nodejs/lancedb/query.ts | 1 - rust/lancedb/src/table/dataset.rs | 8 ++- rust/lancedb/src/table/merge.rs | 20 +++++++ 4 files changed, 124 insertions(+), 2 deletions(-) diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts index 5658029b2..47f80c3d9 100644 --- a/nodejs/__test__/table.test.ts +++ b/nodejs/__test__/table.test.ts @@ -2569,6 +2569,103 @@ describe.each([arrow15, arrow16, arrow17, arrow18])( expect(bCalls).toBe(1); }); + test("stale FTS routing keeps newer vector preparation", async () => { + let vectorCalls = 0; + let markVectorStarted!: () => void; + const vectorStarted = new Promise((resolve) => { + markVectorStarted = resolve; + }); + let releaseVector!: () => void; + const vectorReleased = new Promise((resolve) => { + releaseVector = resolve; + }); + + @register("stale-fts-race") + class RaceEmbedding extends EmbeddingFunction { + ndims() { + return 1; + } + embeddingDataType() { + return new arrow.Float32(); + } + async computeQueryEmbeddings() { + vectorCalls += 1; + markVectorStarted(); + await vectorReleased; + return [0.1]; + } + async computeSourceEmbeddings(values: string[]) { + return values.map(() => [0.1]); + } + } + + const writer = await connect(tmpDir.name); + const ftsTable = await writer.createTable("stale_fts", [ + { text: "hello", vector: [0.0] }, + ]); + await ftsTable.createIndex("text", { config: Index.fts() }); + + const reader = await connect(tmpDir.name, { + readConsistencyInterval: 0, + }); + const tracked = await reader.openTable("stale_fts"); + type Snapshot = { + schema: () => Promise; + }; + type NativeWithSnapshot = { + querySnapshot: () => Promise; + }; + const native = (tracked as unknown as { inner: NativeWithSnapshot }) + .inner; + const querySnapshot = native.querySnapshot.bind(native); + let snapshotCalls = 0; + let markStaleSchemaStarted!: () => void; + const staleSchemaStarted = new Promise((resolve) => { + markStaleSchemaStarted = resolve; + }); + let releaseStaleSchema!: () => void; + const staleSchemaReleased = new Promise((resolve) => { + releaseStaleSchema = resolve; + }); + native.querySnapshot = async () => { + const snapshot = await querySnapshot(); + snapshotCalls += 1; + if (snapshotCalls === 1) { + const schema = snapshot.schema.bind(snapshot); + snapshot.schema = async () => { + markStaleSchemaStarted(); + await staleSchemaReleased; + return await schema(); + }; + } + return snapshot; + }; + + const query = tracked.search("hello"); + const staleFtsExecution = query.toArray(); + await staleSchemaStarted; + + const embedding = new RaceEmbedding(); + const vectorSchema = LanceSchema({ + text: embedding.sourceField(new arrow.Utf8()), + vector: embedding.vectorField(), + }); + await writer.createTable("stale_fts", [{ text: "hello" }], { + mode: "overwrite", + schema: vectorSchema, + }); + + const vectorExecution = query.toArray(); + await vectorStarted; + releaseStaleSchema(); + await staleFtsExecution; + releaseVector(); + await vectorExecution; + + await query.toArray(); + expect(vectorCalls).toBe(1); + }); + test("tokenizes FTS queries by column or index name", async () => { const db = await connect(tmpDir.name); const data = [ diff --git a/nodejs/lancedb/query.ts b/nodejs/lancedb/query.ts index a51471ef4..eff712d9a 100644 --- a/nodejs/lancedb/query.ts +++ b/nodejs/lancedb/query.ts @@ -929,7 +929,6 @@ export function createAutoQuery( while (true) { const initial = await snapshotRoute(); if (initial.embeddingMetadata === undefined) { - cachedPreparation = undefined; const inner = initial.table.query(); inner.fullTextSearch({ query, columns }); return { inner, route: "fts" }; diff --git a/rust/lancedb/src/table/dataset.rs b/rust/lancedb/src/table/dataset.rs index 643fc5ec0..3fde6ef62 100644 --- a/rust/lancedb/src/table/dataset.rs +++ b/rust/lancedb/src/table/dataset.rs @@ -102,11 +102,17 @@ impl DatasetConsistencyWrapper { /// retaining this wrapper's live MemWAL read context. pub fn new_query_snapshot(&self, dataset: Arc) -> Self { let version = dataset.version().version; + let query_snapshot = { + let state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + // Preserve user time travel so the MemWAL safety guard still sees + // it. Latest and already-internal snapshots remain internal pins. + state.query_snapshot || state.pinned_version.is_none() + }; Self { state: Arc::new(Mutex::new(DatasetState { dataset, pinned_version: Some(version), - query_snapshot: true, + query_snapshot, })), consistency: ConsistencyMode::Lazy, shard_writer: self.shard_writer.clone(), diff --git a/rust/lancedb/src/table/merge.rs b/rust/lancedb/src/table/merge.rs index d03ef0d50..480527bc0 100644 --- a/rust/lancedb/src/table/merge.rs +++ b/rust/lancedb/src/table/merge.rs @@ -932,6 +932,26 @@ mod lsm_tests { ); } + #[tokio::test] + async fn query_snapshot_preserves_time_travel_lsm_guard() { + let dir = tempdir().unwrap(); + let table = id_value_table(&dir).await; + table + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap(); + lsm_upsert(&table, vec![4]).await; + + let version = table.version().await.unwrap(); + table.checkout(version).await.unwrap(); + let direct_error = table.query().execute().await.err().unwrap(); + assert!(matches!(direct_error, Error::NotSupported { .. })); + + let snapshot = table.query_snapshot().await.unwrap(); + let snapshot_error = snapshot.query().execute().await.err().unwrap(); + assert!(matches!(snapshot_error, Error::NotSupported { .. })); + } + #[tokio::test] async fn lsm_read_dedup_newest_wins() { let dir = tempdir().unwrap();