fix(node): preserve auto query state provenance

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<void>((resolve) => {
+        markVectorStarted = resolve;
+      });
+      let releaseVector!: () => void;
+      const vectorReleased = new Promise<void>((resolve) => {
+        releaseVector = resolve;
+      });
+
+      @register("stale-fts-race")
+      class RaceEmbedding extends EmbeddingFunction<string> {
+        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<Buffer>;
+      };
+      type NativeWithSnapshot = {
+        querySnapshot: () => Promise<Snapshot>;
+      };
+      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<void>((resolve) => {
+        markStaleSchemaStarted = resolve;
+      });
+      let releaseStaleSchema!: () => void;
+      const staleSchemaReleased = new Promise<void>((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 = [

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" };

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<Dataset>) -> 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(),

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();