feat(mem_wal): support prefilters in LSM vector and full-text search

java/lance-jni/src/mem_wal.rs

@@ -361,7 +361,7 @@ fn inner_scanner_project(env: &mut JNIEnv, this: JObject, columns: JObject) -> R
     let columns = env.get_strings(&columns)?;
     with_lsm_scanner(env, &this, |scanner| {
         let cols: Vec<&str> = columns.iter().map(String::as_str).collect();
-        Ok(scanner.project(&cols))
+        Ok(scanner.project(&cols)?)
     })
 }
 
@@ -395,10 +395,12 @@ fn inner_scanner_limit(
     limit: jlong,
     offset: JObject,
 ) -> Result<()> {
-    let offset = env.get_u64_opt(&offset)?.map(|v| v as usize);
-    with_lsm_scanner(env, &this, |scanner| {
-        Ok(scanner.limit(limit as usize, offset))
-    })
+    let offset = env.get_u64_opt(&offset)?.map(|v| v as i64);
+    with_lsm_scanner(
+        env,
+        &this,
+        |scanner| Ok(scanner.limit(Some(limit), offset)?),
+    )
 }
 
 #[unsafe(no_mangle)]

python/src/mem_wal.rs

@@ -520,7 +520,11 @@ impl PyLsmScanner {
             .take()
             .ok_or_else(|| PyRuntimeError::new_err("Scanner has already been consumed"))?;
         let cols: Vec<&str> = columns.iter().map(|s| s.as_str()).collect();
-        slf.inner = Some(scanner.project(&cols));
+        slf.inner = Some(
+            scanner
+                .project(&cols)
+                .map_err(|e| PyValueError::new_err(e.to_string()))?,
+        );
         Ok(slf)
     }
 
@@ -549,7 +553,11 @@ impl PyLsmScanner {
             .inner
             .take()
             .ok_or_else(|| PyRuntimeError::new_err("Scanner has already been consumed"))?;
-        slf.inner = Some(scanner.limit(n, offset));
+        slf.inner = Some(
+            scanner
+                .limit(Some(n as i64), offset.map(|o| o as i64))
+                .map_err(|e| PyValueError::new_err(e.to_string()))?,
+        );
         Ok(slf)
     }
 

rust/lance/benches/mem_wal/fts/mem_wal_fineweb_fts.rs

@@ -706,10 +706,14 @@ async fn run_read(args: &Args, uri: &str, corpus: &[String]) -> Result<serde_jso
         if *is_phrase {
             scanner.full_text_phrase(TEXT_COL, q, 0);
         } else {
-            scanner.full_text_search(TEXT_COL, q);
+            scanner.full_text_search(
+                FullTextSearchQuery::new(q.to_string())
+                    .with_column(TEXT_COL.to_string())
+                    .unwrap(),
+            )?;
         }
-        scanner.project(&["id"]);
-        scanner.limit(args.top_k, None);
+        scanner.project(&["id"])?;
+        scanner.limit(Some(args.top_k as i64), None)?;
         let t0 = Instant::now();
         let stream = scanner.try_into_stream().await?;
         let batches: Vec<RecordBatch> = stream.try_collect().await?;

rust/lance/benches/mem_wal/read/memtable_read.rs

@@ -773,8 +773,14 @@ fn bench_fts(c: &mut Criterion) {
                 async move {
                     let mut total_found = 0usize;
                     for (mut scanner, term) in scanners.into_iter().zip(terms.iter()) {
+                        scanner
+                            .full_text_search(
+                                FullTextSearchQuery::new(term.to_string())
+                                    .with_column("text".to_string())
+                                    .unwrap(),
+                            )
+                            .unwrap();
                         let batches: Vec<RecordBatch> = scanner
-                            .full_text_search("text", term)
                             .try_into_stream()
                             .await
                             .unwrap()
@@ -947,11 +953,14 @@ fn bench_vector_search(c: &mut Criterion) {
             let query_array: Arc<dyn arrow_array::Array> = Arc::new(Float32Array::from(q.clone()));
             b.to_async(&rt).iter(|| {
                 let query_array = query_array.clone();
-                async {
+                let memtable = &memtable;
+                async move {
                     let mut scanner = memtable.scan();
+                    scanner
+                        .nearest("vector", query_array.as_ref(), k)
+                        .unwrap()
+                        .nprobes(8);
                     let batches: Vec<RecordBatch> = scanner
-                        .nearest("vector", query_array, k)
-                        .nprobes(8)
                         .try_into_stream()
                         .await
                         .unwrap()

rust/lance/benches/mem_wal/vector/hnsw/mem_wal_recall_hnsw.rs

@@ -665,7 +665,7 @@ async fn run_checkpoint(
             active.schema.clone(),
         );
         let q_arr: ArrayRef = Arc::new(q_fsl);
-        scanner.nearest(VECTOR_COL, q_arr, k);
+        scanner.nearest(VECTOR_COL, q_arr.as_ref(), k)?;
         let h_t = Instant::now();
         let stream = scanner.try_into_stream().await?;
         let batches: Vec<RecordBatch> = stream.try_collect().await?;

rust/lance/benches/mem_wal/vector/mem_wal_index_micro.rs

@@ -298,7 +298,7 @@ async fn main() -> lance_core::Result<()> {
                     active.schema.clone(),
                 );
                 let q_arr: ArrayRef = Arc::new(q_fsl);
-                scanner.nearest(VECTOR_COL, q_arr, 10);
+                scanner.nearest(VECTOR_COL, q_arr.as_ref(), 10)?;
                 let t = Instant::now();
                 let stream = scanner.try_into_stream().await?;
                 let _: Vec<RecordBatch> = stream.try_collect().await?;