refactor(mem_wal): align LsmScanner with the dataset Scanner interface

LsmScanner gains a Scanner-aligned (owned) builder so an LSM read reads like a
normal scan. `nearest()` and `full_text_search()` become state setters and
`create_plan()` dispatches to the vector, FTS, point-lookup, or plain planner,
mirroring `Scanner::create_plan` / `MemTableScanner::create_plan`.

- Add `nearest`, `nprobes`, `refine`, `distance_metric` (vector search folds the
  LsmVectorSearchPlanner behind the builder; honors the builder filter).
- `full_text_search(FullTextSearchQuery)` is now a setter (column from the query,
  k from `limit`) instead of returning a plan directly.
- Align `project` (`<T: AsRef<str>>` + `Result`) and `limit`
  (`Option<i64>, Option<i64>` + `Result`) with `Scanner`.

Knobs the LSM planner cannot yet honor (ef, distance_range, maximum_nprobes,
with_row_id) are intentionally not exposed to avoid silently ignoring them.

Author: touch-of-grey

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/src/dataset/mem_wal/scanner.rs

@@ -24,9 +24,9 @@
 //! use lance::dataset::mem_wal::scanner::LsmScanner;
 //!
 //! let scanner = LsmScanner::new(base_table, shard_snapshots, vec!["pk".to_string()])
-//!     .project(&["id", "name"])
+//!     .project(&["id", "name"])?
 //!     .filter("id > 10")?
-//!     .limit(100, None);
+//!     .limit(Some(100), None)?;
 //!
 //! let stream = scanner.try_into_stream().await?;
 //! ```