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

WAL queries have the ability to bypass WAL by using `use_lsm(bool)` to
determine whether to explicitly use or not use it. This is exposed on
many of the entry-points but not on the `BaseTableAdapter`. This PR adds
that function to allow WAL use configuration.

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

rust/lancedb/src/table/datafusion.rs

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

rust/lancedb/src/table/merge.rs

@@ -1349,6 +1349,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();