From 8c504101aaf8380ede4f07d253fc23e8d60ed9b7 Mon Sep 17 00:00:00 2001
From: Daniel Rammer <hamersaw@protonmail.com>
Date: Fri, 9 Oct 2026 14:12:16 -0500
Subject: [PATCH 1/3] fix(mem_wal): rank every vector search arm on one
 distance scale

The LSM vector planner merges its arms by `_distance`, but cosine reached
the merge on two scales: flushed generations index as IVF_HNSW_SQ, which
ranks cosine as L2 over normalized vectors (twice `1 - cos`), while the
memtable ranks it exactly. Flush also wrote unnormalized vectors into the
cosine index, so a generation's search ranked non-unit embeddings by L2
against a normalized query.

Normalize vectors at flush for a cosine index, and add two planner
builders: `with_refine_factor` refines the base and SSTable arms so every
arm is exact, and `with_distance_scale` puts unrefined arms on the scale a
caller's own base search reports, converting the distance range to match.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../lance/memwal/LsmVectorSearchPlanner.java  |   9 +-
 python/python/lance/mem_wal.py                |   8 +-
 rust/lance/src/dataset/mem_wal/index/hnsw.rs  |  11 +
 .../src/dataset/mem_wal/memtable/flush.rs     | 127 +++++++++
 .../src/dataset/mem_wal/scanner/builder.rs    |  21 +-
 .../src/dataset/mem_wal/scanner/projection.rs |  37 ++-
 .../dataset/mem_wal/scanner/vector_search.rs  | 269 ++++++++++++++++--
 7 files changed, 435 insertions(+), 47 deletions(-)

diff --git a/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java b/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java
index b91e5ae4a7d..34e424c49ba 100644
--- a/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java
+++ b/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java
@@ -117,9 +117,9 @@ private native void nativeCreate(
    * @param nprobes number of IVF partitions to probe
    * @param columns columns to project; pass {@code null} to return all columns plus {@code
    *     _distance}
-   * @param refineBaseTable when true, the base-table arm re-ranks candidates with exact distances
-   *     (refine factor 1). Useful when the base table uses an approximate index (IVF-PQ).
-   *     Auto-enabled whenever stale filtering is on (see {@code overfetchFactor}).
+   * @param refineBaseTable when true, the base-table and SSTable arms re-rank candidates with
+   *     exact distances (refine factor 1), so every arm reports exact distances. Auto-enabled
+   *     whenever stale filtering is on (see {@code overfetchFactor}).
    * @param overfetchFactor over-fetch multiple for sources with rows superseded by newer
    *     generations. Must be at least {@code 1.0}:
    *     <ul>
@@ -177,7 +177,8 @@ public ExecutionPlan planSearch(
    * @param nprobes number of IVF partitions to probe
    * @param columns columns to project; pass {@code null} to return all columns plus {@code
    *     _distance}
-   * @param refineBaseTable when true, the base-table arm re-ranks candidates with exact distances.
+   * @param refineBaseTable when true, the base-table and SSTable arms re-rank candidates with
+   *     exact distances.
    * @return an executable plan
    */
   public ExecutionPlan planSearch(
diff --git a/python/python/lance/mem_wal.py b/python/python/lance/mem_wal.py
index b9a8890853a..04056199798 100644
--- a/python/python/lance/mem_wal.py
+++ b/python/python/lance/mem_wal.py
@@ -582,10 +582,10 @@ def plan_search(
             Columns to project.  Returns all columns + ``_distance`` if
             omitted.
         refine_base_table : bool, optional
-            When true, the base-table arm re-ranks its candidates with exact
-            distances.  Useful when the base index is approximate (e.g.
-            IVF-PQ).  Memtable arms use exact HNSW and are unaffected.
-            Auto-enabled whenever stale filtering is on (see ``overfetch_factor``).
+            When true, the base-table and SSTable arms re-rank their candidates
+            with exact distances, so every arm reports exact distances.
+            Memtable arms are already exact. Auto-enabled whenever stale
+            filtering is on (see ``overfetch_factor``).
         overfetch_factor : float, optional
             Over-fetch multiple for sources with rows superseded by newer
             generations. Must be at least ``1.0``. At ``1.0`` each affected
diff --git a/rust/lance/src/dataset/mem_wal/index/hnsw.rs b/rust/lance/src/dataset/mem_wal/index/hnsw.rs
index 581814a2309..51fb7d45a1d 100644
--- a/rust/lance/src/dataset/mem_wal/index/hnsw.rs
+++ b/rust/lance/src/dataset/mem_wal/index/hnsw.rs
@@ -569,6 +569,17 @@ async fn write_sq_storage(
     };
     let row_ids = column(ROW_ID)?;
     let vectors = column(FLAT_COLUMN)?;
+    // A cosine index normalizes the query and ranks by L2, so it must store
+    // normalized vectors, as a regular cosine build does. The graph is
+    // unaffected: cosine over raw vectors orders neighbors the same as L2 over
+    // normalized ones.
+    let vectors: Arc<dyn Array> = if distance_type == DistanceType::Cosine {
+        Arc::new(lance_linalg::kernels::normalize_fsl(
+            vectors.as_fixed_size_list(),
+        )?)
+    } else {
+        vectors
+    };
     let vectors = vectors.as_fixed_size_list();
     let mut sq = ScalarQuantizer::new(8, dim);
     let bounds: Range<f64> = sq.update_bounds::<Float32Type>(vectors)?;
diff --git a/rust/lance/src/dataset/mem_wal/memtable/flush.rs b/rust/lance/src/dataset/mem_wal/memtable/flush.rs
index 0b52568c733..ae47cbd7f32 100644
--- a/rust/lance/src/dataset/mem_wal/memtable/flush.rs
+++ b/rust/lance/src/dataset/mem_wal/memtable/flush.rs
@@ -1878,6 +1878,133 @@ mod tests {
         );
     }
 
+    /// A flushed cosine index must hold normalized vectors: its search
+    /// normalizes the query and ranks by L2, which only tracks cosine when the
+    /// stored side is normalized too.
+    #[tokio::test]
+    async fn test_flusher_cosine_index_ranks_unnormalized_vectors() {
+        use super::super::super::index::IndexStore;
+        use arrow_array::{FixedSizeListArray, Float32Array};
+        use lance_linalg::distance::DistanceType;
+
+        let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
+        let shard_id = Uuid::new_v4();
+        let manifest_store = Arc::new(ShardManifestStore::new(
+            store.clone(),
+            &base_path,
+            shard_id,
+            2,
+        ));
+        let (epoch, _manifest) = manifest_store.claim_epoch(0).await.unwrap();
+
+        // Norms spread over 0.2..5.0, so ranking by L2 against raw vectors
+        // would disagree with cosine.
+        let (dim, num_vectors) = (32usize, 2000usize);
+        let mut seed: u64 = 42;
+        let mut next = || {
+            seed = seed
+                .wrapping_mul(6364136223846793005)
+                .wrapping_add(1442695040888963407);
+            ((seed >> 33) as f32 / (1u64 << 31) as f32) * 2.0 - 1.0
+        };
+        let mut vectors = Vec::with_capacity(num_vectors * dim);
+        for _ in 0..num_vectors {
+            let row: Vec<f32> = (0..dim).map(|_| next()).collect();
+            let norm = row.iter().map(|x| x * x).sum::<f32>().sqrt();
+            let target = 0.2 + 4.8 * (next() * 0.5 + 0.5);
+            vectors.extend(row.iter().map(|x| x * target / norm));
+        }
+        let query: Vec<f32> = (0..dim).map(|_| next()).collect();
+        let cosine_distance = |id: usize| {
+            lance_linalg::distance::cosine_distance(&query, &vectors[id * dim..][..dim])
+        };
+        let mut by_distance: Vec<usize> = (0..num_vectors).collect();
+        by_distance.sort_by(|a, b| cosine_distance(*a).total_cmp(&cosine_distance(*b)));
+        let nearest: std::collections::HashSet<i32> =
+            by_distance[..10].iter().map(|id| *id as i32).collect();
+
+        let item = Arc::new(Field::new("item", DataType::Float32, false));
+        let schema = Arc::new(ArrowSchema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new(
+                "vector",
+                DataType::FixedSizeList(item.clone(), dim as i32),
+                false,
+            ),
+        ]));
+        let index_specs = vec![MemIndexSpec::hnsw(
+            "vector_hnsw",
+            1,
+            "vector",
+            DistanceType::Cosine,
+        )];
+        let mut memtable = MemTable::new(schema.clone(), 1, vec![]).unwrap();
+        let lance_schema = lance_core::datatypes::Schema::try_from(schema.as_ref()).unwrap();
+        memtable.set_indexes(
+            IndexStore::from_specs(&index_specs, &lance_schema, num_vectors, 100).unwrap(),
+        );
+        let batch = RecordBatch::try_new(
+            schema,
+            vec![
+                Arc::new(Int32Array::from_iter_values(0..num_vectors as i32)),
+                Arc::new(
+                    FixedSizeListArray::try_new(
+                        item,
+                        dim as i32,
+                        Arc::new(Float32Array::from(vectors.clone())),
+                        None,
+                    )
+                    .unwrap(),
+                ),
+            ],
+        )
+        .unwrap();
+        let frag_id = memtable.insert(batch).await.unwrap();
+
+        let flusher = MemTableFlusher::new(
+            store.clone(),
+            base_path.clone(),
+            base_uri.clone(),
+            shard_id,
+            manifest_store.clone(),
+        );
+        let result = flusher
+            .flush_with_indexes(&memtable, epoch, &index_specs, 1, frag_id + 1)
+            .await
+            .unwrap();
+        let gen_uri = format!("{}/_mem_wal/{}/{}", base_uri, shard_id, result.sstable.path);
+        let dataset = Dataset::open(&gen_uri).await.unwrap();
+
+        let mut scan = dataset.scan();
+        scan.nearest("vector", &Float32Array::from(query.clone()), 10)
+            .unwrap();
+        scan.distance_metric(DistanceType::Cosine);
+        // A wide search list keeps graph recall out of what this measures.
+        scan.ef(500);
+        scan.fast_search();
+        let batch = scan.try_into_batch().await.unwrap();
+        let ids = batch["id"].as_any().downcast_ref::<Int32Array>().unwrap();
+        let distances = batch["_distance"]
+            .as_any()
+            .downcast_ref::<Float32Array>()
+            .unwrap();
+
+        let found = ids
+            .values()
+            .iter()
+            .filter(|id| nearest.contains(id))
+            .count();
+        assert!(found >= 9, "recall@10 was {found}/10");
+        // Lance's SQ index reports cosine as L2 over normalized vectors.
+        for (id, distance) in ids.values().iter().zip(distances.values()) {
+            let expected = 2.0 * cosine_distance(*id as usize);
+            assert!(
+                (distance - expected).abs() < 0.05,
+                "id={id} reported {distance}, expected {expected}"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn test_flusher_with_fts_index() {
         use super::super::super::index::IndexStore;
diff --git a/rust/lance/src/dataset/mem_wal/scanner/builder.rs b/rust/lance/src/dataset/mem_wal/scanner/builder.rs
index 82fbeb41ce9..261142e2c8a 100644
--- a/rust/lance/src/dataset/mem_wal/scanner/builder.rs
+++ b/rust/lance/src/dataset/mem_wal/scanner/builder.rs
@@ -48,9 +48,9 @@ struct LsmVectorQuery {
     k: usize,
     /// IVF partition probe bounds for indexed arms.
     probe_bounds: ProbeBounds,
-    /// Re-rank base candidates with exact distances when set (refine factor is
-    /// treated as a boolean; the LSM merge needs exact base distances).
-    refine: bool,
+    /// Refine factor for the approximate-index arms; `None` leaves them on the
+    /// index's native distance scale.
+    refine_factor: Option<u32>,
     /// Distance metric; `None` defaults to L2 (matching the unindexed memtable arm).
     metric_type: Option<DistanceType>,
 }
@@ -490,7 +490,7 @@ impl LsmScanner {
             key: key.slice(0, key.len()),
             k,
             probe_bounds: ProbeBounds::default(),
-            refine: false,
+            refine_factor: None,
             metric_type: None,
         });
         Ok(self)
@@ -528,12 +528,12 @@ impl LsmScanner {
         self
     }
 
-    /// Re-rank base-table candidates with exact distances so they are directly
-    /// comparable to the exact memtable distances in the cross-source merge.
-    /// The factor is treated as a boolean today. No-op unless `nearest` was set.
+    /// Re-rank the approximate-index arms' candidates by exact distance,
+    /// searching `refine_factor` times as many. `0` clears it. No-op unless
+    /// `nearest` was set.
     pub fn refine(mut self, refine_factor: u32) -> Self {
         if let Some(q) = self.nearest.as_mut() {
-            q.refine = refine_factor > 0;
+            q.refine_factor = (refine_factor > 0).then_some(refine_factor);
         }
         self
     }
@@ -633,7 +633,8 @@ impl LsmScanner {
             distance_type,
         )
         .with_identity_schema(Arc::clone(&self.identity_schema))
-        .with_filter(self.filter.clone());
+        .with_filter(self.filter.clone())
+        .with_refine_factor(nearest.refine_factor);
         if let BaseSource::Table(dataset) = &self.base {
             planner = planner.with_dataset(dataset.clone());
         }
@@ -660,7 +661,7 @@ impl LsmScanner {
                 per_source_k,
                 nearest.probe_bounds,
                 self.projection.as_deref(),
-                nearest.refine,
+                false,
                 overfetch_factor,
             )
             .await?;
diff --git a/rust/lance/src/dataset/mem_wal/scanner/projection.rs b/rust/lance/src/dataset/mem_wal/scanner/projection.rs
index e0f6e470712..ddc25c7210a 100644
--- a/rust/lance/src/dataset/mem_wal/scanner/projection.rs
+++ b/rust/lance/src/dataset/mem_wal/scanner/projection.rs
@@ -19,8 +19,9 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use arrow_schema::{DataType, Field, Schema, SchemaRef};
+use datafusion::logical_expr::Operator;
 use datafusion::physical_expr::PhysicalExpr;
-use datafusion::physical_expr::expressions::{CastExpr, Column, Literal};
+use datafusion::physical_expr::expressions::{BinaryExpr, CastExpr, Column, Literal};
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::physical_plan::projection::ProjectionExec;
 use datafusion::scalar::ScalarValue;
@@ -217,6 +218,40 @@ pub fn canonical_output_schema(
     Ok(Arc::new(Schema::new(fields)))
 }
 
+/// Wrap `plan` so `_distance` is multiplied by `factor`; all other columns are
+/// forwarded unchanged. A no-op when `factor` is `1.0`.
+pub fn scale_distance(plan: Arc<dyn ExecutionPlan>, factor: f32) -> Result<Arc<dyn ExecutionPlan>> {
+    if factor == 1.0 {
+        return Ok(plan);
+    }
+    let input_schema = plan.schema();
+    let project_exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = input_schema
+        .fields()
+        .iter()
+        .enumerate()
+        .map(|(idx, field)| {
+            let column: Arc<dyn PhysicalExpr> = Arc::new(Column::new(field.name(), idx));
+            let expr = if field.name() == DISTANCE_COLUMN {
+                Arc::new(BinaryExpr::new(
+                    column,
+                    Operator::Multiply,
+                    Arc::new(Literal::new(ScalarValue::Float32(Some(factor)))),
+                ))
+            } else {
+                column
+            };
+            (expr, field.name().clone())
+        })
+        .collect();
+    let projection_exec = ProjectionExec::try_new(project_exprs, plan).map_err(|e| {
+        lance_core::Error::internal(format!(
+            "Failed to build scale_distance ProjectionExec: {}",
+            e
+        ))
+    })?;
+    Ok(Arc::new(projection_exec))
+}
+
 /// Wrap `plan` so the named columns become typed NULL literals; all
 /// other columns are forwarded unchanged. Schema is preserved (same
 /// fields, same dtypes). Useful for stripping the *value* of an
diff --git a/rust/lance/src/dataset/mem_wal/scanner/vector_search.rs b/rust/lance/src/dataset/mem_wal/scanner/vector_search.rs
index 884c9b1067c..20f737eb30d 100644
--- a/rust/lance/src/dataset/mem_wal/scanner/vector_search.rs
+++ b/rust/lance/src/dataset/mem_wal/scanner/vector_search.rs
@@ -31,7 +31,7 @@ use super::data_source::LsmDataSource;
 use super::generation_read::{GenerationRead, memtable_matches_table};
 use super::projection::{
     DISTANCE_COLUMN, build_scanner_projection, canonical_output_schema, null_columns,
-    project_to_canonical, validate_projection_names, wants_row_id,
+    project_to_canonical, scale_distance, validate_projection_names, wants_row_id,
 };
 use super::sstable_cache::{DatasetCache, SsTableWarmer, open_sstable};
 use crate::session::Session;
@@ -71,12 +71,14 @@ impl ProbeBounds {
 ///     SortExec: order_by=[_distance ASC], fetch=k          (per partition, parallel)
 ///       UnionExec
 ///         ProjectionExec (canonical output schema)
-///           SortExec(_distance, fetch=k)
+///           ProjectionExec (_distance rescaled, unrefined only; see `with_distance_scale`)
+///             SortExec(_distance, fetch=k)
 ///               MemTableBruteForceVectorExec or VectorIndexExec: active memtable KNN
 ///         ProjectionExec (canonical output schema)
 ///           ProjectionExec (null_columns _rowid)
 ///             PkBlockFilterExec: block-list                   (SSTable)
-///               KNNExec: SSTable gen N, fetch=ceil(k*overfetch) (fast_search)
+///               ProjectionExec (_distance rescaled, unrefined only)
+///                 KNNExec: SSTable gen N, fetch=ceil(k*overfetch) (fast_search)[.refine()?]
 ///         … one per SSTable gen …
 ///         ProjectionExec (canonical output schema)
 ///           PkBlockFilterExec: block-list                     (base)
@@ -130,6 +132,22 @@ pub struct LsmVectorSearchPlanner {
     /// generation's `IVF_HNSW_SQ` — and is ignored by an arm whose index is not
     /// a graph. `None` leaves each arm at its own default.
     ef: Option<usize>,
+    /// Refine factor for every approximate-index arm; see
+    /// [`Self::with_refine_factor`].
+    refine_factor: Option<u32>,
+    /// Multiple of the exact distance every unrefined arm reports; see
+    /// [`Self::with_distance_scale`].
+    distance_scale: f32,
+}
+
+/// Multiple of the exact distance a flushed generation's index reports. It is
+/// written as `IVF_HNSW_SQ`, which ranks cosine as L2 over normalized vectors,
+/// `2 * (1 - cos)`; the other metrics it ranks on their own scale.
+fn sstable_distance_scale(distance_type: lance_linalg::distance::DistanceType) -> f32 {
+    match distance_type {
+        lance_linalg::distance::DistanceType::Cosine => 2.0,
+        _ => 1.0,
+    }
 }
 
 impl LsmVectorSearchPlanner {
@@ -164,6 +182,8 @@ impl LsmVectorSearchPlanner {
             filter: None,
             distance_range: (None, None),
             ef: None,
+            refine_factor: None,
+            distance_scale: 1.0,
         }
     }
 
@@ -194,6 +214,34 @@ impl LsmVectorSearchPlanner {
         self
     }
 
+    /// Set the refine factor. `Some(rf)` refines the base and SSTable arms by
+    /// `rf`, so every arm reports exact distances. `None` leaves them
+    /// unrefined; see [`Self::with_distance_scale`].
+    pub fn with_refine_factor(mut self, refine_factor: Option<u32>) -> Self {
+        self.refine_factor = refine_factor;
+        self
+    }
+
+    /// Report unrefined distances as `scale` times the exact distance, which
+    /// lines this plan up with a base search the caller merges it with: Lance
+    /// ranks cosine on SQ and PQ indexes as L2 over normalized vectors, twice
+    /// `1 - cos`. The memtable arm scales its exact distances, and the SSTable
+    /// arms convert from their own index's multiple. The distance range is
+    /// given on the same scale. Ignored when refining. Defaults to `1.0`.
+    pub fn with_distance_scale(mut self, scale: f32) -> Self {
+        self.distance_scale = scale;
+        self
+    }
+
+    /// The factor that brings an arm reporting `native` times the exact
+    /// distance to the plan's scale.
+    fn arm_scale(&self, refine_factor: Option<u32>, native: f32) -> f32 {
+        match refine_factor {
+            Some(_) => 1.0,
+            None => self.distance_scale / native,
+        }
+    }
+
     /// The table's schema carrying each field's id, which is what resolves a
     /// generation's stored columns to the table's.
     ///
@@ -249,12 +297,10 @@ impl LsmVectorSearchPlanner {
     /// * `k` - Number of nearest neighbors to return
     /// * `nprobes` - Exact number of IVF partitions to search (for IVF-based indexes)
     /// * `projection` - Columns to include in output (None = all columns)
-    /// * `refine_base_table` - When true, the base-table arm re-ranks its
-    ///   candidates with exact distances (refine factor 1). Useful when the base
-    ///   table uses an approximate index (IVF-PQ) so cross-source distance
-    ///   comparison is exact. Memtable arms already use exact distances and
-    ///   never need refine. Auto-enabled whenever stale filtering is on (see
-    ///   below).
+    /// * `refine_base_table` - When true and no refine factor is set, every
+    ///   approximate-index arm re-ranks its candidates with exact distances
+    ///   (refine factor 1), so the base table's are exact too. Auto-enabled
+    ///   whenever a newer generation blocks the base table (see below).
     /// * `overfetch_factor` - Controls how aggressively sources over-fetch to
     ///   backfill the rows dropped by stale-row filtering. Values below `1.0`
     ///   are rejected; stale filtering is always enabled.
@@ -353,11 +399,16 @@ impl LsmVectorSearchPlanner {
             true, // include _distance — KNN always produces it
         )?;
 
-        // Refine the base table when explicitly requested, or whenever the base
-        // is blocked (it then over-fetches its approximate-index candidates, so
-        // distances must be re-ranked to exact before the cross-source merge).
+        // The merge ranks every arm by `_distance`, so all of them must report
+        // on one scale: exact when refining, `distance_scale` when not. A
+        // blocked base over-fetches its approximate candidates, so it refines.
         // `block_lists` is non-empty exactly when a newer generation exists.
-        let refine_base = refine_base_table || !block_lists.is_empty();
+        let has_base = sources
+            .iter()
+            .any(|source| matches!(source, LsmDataSource::BaseTable { .. }));
+        let refine_factor = self
+            .refine_factor
+            .or_else(|| (refine_base_table || (has_base && !block_lists.is_empty())).then_some(1));
 
         // Stage per-source over-fetch decisions, then build every KNN plan
         // concurrently — the builds are independent and a sequential loop was
@@ -384,8 +435,8 @@ impl LsmVectorSearchPlanner {
         // Type-erased, not merely boxed: the `Send` proof recurses through a
         // boxed future's concrete type but stops at a trait object, and an arm
         // resolves a generation's schema before it searches.
-        let built = futures::future::try_join_all(arm_inputs.iter().map(
-            |(source, is_base, _, _, fetch_k)| {
+        let built =
+            futures::future::try_join_all(arm_inputs.iter().map(|(source, _, _, _, fetch_k)| {
                 let arm: futures::future::BoxFuture<'_, Result<Arc<dyn ExecutionPlan>>> =
                     Box::pin(self.build_knn_plan(
                         source,
@@ -393,12 +444,11 @@ impl LsmVectorSearchPlanner {
                         *fetch_k,
                         probe_bounds,
                         projection,
-                        *is_base && refine_base,
+                        refine_factor,
                     ));
                 arm
-            },
-        ))
-        .await?;
+            }))
+            .await?;
 
         let mut knn_plans = Vec::new();
         for ((_, is_base, _, blocked, _), knn) in arm_inputs.iter().zip(built) {
@@ -516,7 +566,7 @@ impl LsmVectorSearchPlanner {
         k: usize,
         probe_bounds: ProbeBounds,
         projection: Option<&[String]>,
-        refine: bool,
+        refine_factor: Option<u32>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         match source {
             LsmDataSource::BaseTable { dataset } => {
@@ -558,10 +608,8 @@ impl LsmVectorSearchPlanner {
                 }
                 // Memtables cover unindexed rows; only search indexed data here.
                 scanner.fast_search();
-                // Re-rank base candidates with exact distances so they're
-                // directly comparable to memtable distances in the merge.
-                if refine {
-                    scanner.refine(1);
+                if let Some(refine_factor) = refine_factor {
+                    scanner.refine(refine_factor);
                 }
                 scanner.create_plan().await
             }
@@ -635,7 +683,12 @@ impl LsmVectorSearchPlanner {
                     }
                 };
                 scanner.nearest(&vector_column, query_arr.as_ref(), k)?;
-                scanner.distance_range(self.distance_range.0, self.distance_range.1);
+                let scale =
+                    self.arm_scale(refine_factor, sstable_distance_scale(self.distance_type));
+                scanner.distance_range(
+                    self.distance_range.0.map(|lower| lower / scale),
+                    self.distance_range.1.map(|upper| upper / scale),
+                );
                 if let Some(minimum_nprobes) = probe_bounds.minimum_nprobes {
                     scanner.minimum_nprobes(minimum_nprobes);
                 }
@@ -647,10 +700,14 @@ impl LsmVectorSearchPlanner {
                     scanner.ef(ef);
                 }
                 scanner.fast_search();
+                if let Some(refine_factor) = refine_factor {
+                    scanner.refine(refine_factor);
+                }
                 // Boxed for the reason the scan planner's arm gives: a
                 // generation resolves its own schema before scanning, and
                 // the inlined future is too deep for the `Send` proof.
-                generation.reconcile_above(Box::pin(scanner.create_plan()).await?, &above)
+                let plan = scale_distance(Box::pin(scanner.create_plan()).await?, scale)?;
+                generation.reconcile_above(plan, &above)
             }
             LsmDataSource::ActiveMemTable {
                 batch_store,
@@ -717,7 +774,12 @@ impl LsmVectorSearchPlanner {
                     }
                 };
                 scanner.nearest(&vector_column, query_vector, k)?;
-                scanner.distance_range(self.distance_range.0, self.distance_range.1);
+                // This arm's distances are exact.
+                let scale = self.arm_scale(refine_factor, 1.0);
+                scanner.distance_range(
+                    self.distance_range.0.map(|lower| lower / scale),
+                    self.distance_range.1.map(|upper| upper / scale),
+                );
                 if let Some(minimum_nprobes) = probe_bounds.minimum_nprobes {
                     scanner.minimum_nprobes(minimum_nprobes);
                 }
@@ -728,7 +790,7 @@ impl LsmVectorSearchPlanner {
                 if let Some(ef) = self.ef {
                     scanner.ef(ef);
                 }
-                let plan = Box::pin(scanner.create_plan()).await?;
+                let plan = scale_distance(Box::pin(scanner.create_plan()).await?, scale)?;
                 match generation {
                     Some(generation) => generation.reconcile_above(plan, &above),
                     None => Ok(plan),
@@ -3384,6 +3446,157 @@ mod tests {
         );
     }
 
+    /// The SSTable arm's `IVF_HNSW_SQ` index ranks cosine as `2 * (1 - cos)` and
+    /// the memtable arm ranks it exactly, so the plan must put both on the
+    /// requested scale or the merge interleaves them wrongly.
+    #[tokio::test]
+    async fn test_vector_search_cosine_arms_share_one_scale() {
+        use crate::dataset::mem_wal::scanner::collector::{InMemoryMemTableRef, InMemoryMemTables};
+        use crate::dataset::mem_wal::scanner::data_source::ShardSnapshot;
+        use crate::dataset::mem_wal::write::{BatchStore, IndexStore};
+        use crate::index::DatasetIndexExt;
+        use crate::index::vector::VectorIndexParams;
+        use datafusion::prelude::SessionContext;
+        use futures::TryStreamExt;
+        use lance_index::IndexType;
+        use lance_index::vector::hnsw::builder::HnswBuildParams;
+        use lance_index::vector::ivf::IvfBuildParams;
+        use lance_index::vector::sq::builder::SQBuildParams;
+        use lance_linalg::distance::DistanceType;
+
+        // A vector of norm `norm` whose cosine distance to the query is `cos_distance`.
+        let at = |cos_distance: f32, norm: f32| {
+            let cos = 1.0 - cos_distance;
+            [norm * cos, norm * (1.0 - cos * cos).sqrt(), 0.0, 0.0]
+        };
+        let query = {
+            use arrow_array::builder::Float32Builder;
+            let mut builder = FixedSizeListBuilder::new(Float32Builder::new(), 4);
+            builder.values().append_slice(&[1.0, 0.0, 0.0, 0.0]);
+            builder.append(true);
+            builder.finish()
+        };
+        // Exact cosine distance by id; the arms alternate so a scale mismatch
+        // between them reorders the result.
+        let expected = [(1, 0.05), (3, 0.2), (2, 0.4), (4, 0.6)];
+
+        let schema = create_vector_schema();
+        let temp_dir = tempfile::tempdir().unwrap();
+        let base_uri = format!("{}/base", temp_dir.path().to_str().unwrap());
+        let shard_id = uuid::Uuid::new_v4();
+
+        // SSTable: ids 1 and 2, plus far rows so SQ has a spread to train on.
+        let mut sstable_rows = vec![(1, at(0.05, 3.0)), (2, at(0.4, 2.0))];
+        sstable_rows.extend((10..20).map(|id| (id, at(1.9, 1.0 + id as f32 * 0.1))));
+        let gen1_uri = format!("{}/_mem_wal/{}/gen_1", base_uri, shard_id);
+        let mut gen1 = create_dataset(&gen1_uri, vec![batch_rows(&schema, &sstable_rows)]).await;
+        let hnsw_sq = VectorIndexParams::with_ivf_hnsw_sq_params(
+            DistanceType::Cosine,
+            IvfBuildParams::new(1),
+            HnswBuildParams::default(),
+            SQBuildParams::default(),
+        );
+        gen1.create_index(&["vector"], IndexType::Vector, None, &hnsw_sq, true)
+            .await
+            .unwrap();
+
+        // Active memtable: ids 3 and 4.
+        let batch_store = Arc::new(BatchStore::with_capacity(16));
+        let mut index_store = IndexStore::new();
+        index_store.enable_pk_index(&[("id".to_string(), 0)]);
+        index_store.add_hnsw(
+            "vector_hnsw".to_string(),
+            1,
+            "vector".to_string(),
+            DistanceType::Cosine,
+            64,
+            8,
+        );
+        let batch = batch_rows(&schema, &[(3, at(0.2, 0.5)), (4, at(0.6, 1.0))]);
+        batch_store.append(batch.clone()).unwrap();
+        index_store
+            .insert_with_batch_position(&batch, 0, Some(0))
+            .unwrap();
+        let active = InMemoryMemTableRef {
+            batch_store,
+            index_store: Arc::new(index_store),
+            schema: schema.clone(),
+            generation: 2,
+        };
+
+        let search = |scale: f32, refine: Option<u32>, range: (Option<f32>, Option<f32>)| {
+            let snapshot = ShardSnapshot::new(shard_id)
+                .with_current_generation(2)
+                .with_sstable(1, "gen_1".to_string());
+            let collector =
+                LsmDataSourceCollector::without_base_table(base_uri.clone(), vec![snapshot])
+                    .with_in_memory_memtables(
+                        shard_id,
+                        InMemoryMemTables {
+                            active: active.clone(),
+                            frozen: vec![],
+                        },
+                    );
+            let planner = LsmVectorSearchPlanner::new(
+                collector,
+                vec!["id".to_string()],
+                schema.clone(),
+                "vector".to_string(),
+                DistanceType::Cosine,
+            )
+            .with_distance_scale(scale)
+            .with_refine_factor(refine)
+            .with_distance_range(range.0, range.1);
+            let query = query.clone();
+            async move {
+                let plan = planner
+                    .plan_search(&query, 4, 1, None, false, 1.0)
+                    .await
+                    .unwrap();
+                let batches: Vec<RecordBatch> = plan
+                    .execute(0, SessionContext::new().task_ctx())
+                    .unwrap()
+                    .try_collect()
+                    .await
+                    .unwrap();
+                collect_id_dist(&batches)
+            }
+        };
+        let assert_on_scale = |rows: &[(i32, f32)], scale: f32, what: &str| {
+            let ids: Vec<i32> = rows.iter().map(|(id, _)| *id).collect();
+            assert_eq!(ids, vec![1, 3, 2, 4], "{what}: wrong order, got {rows:?}");
+            for ((id, distance), (_, exact)) in rows.iter().zip(expected) {
+                assert!(
+                    (distance - exact * scale).abs() < 0.02 * scale,
+                    "{what}: id={id} reported {distance}, expected {}",
+                    exact * scale
+                );
+            }
+        };
+
+        assert_on_scale(&search(1.0, None, (None, None)).await, 1.0, "exact scale");
+        assert_on_scale(
+            &search(2.0, None, (None, None)).await,
+            2.0,
+            "SQ/PQ base scale",
+        );
+        // Refining reports exact distances whatever the scale.
+        assert_on_scale(&search(2.0, Some(1), (None, None)).await, 1.0, "refined");
+
+        // On the 2x scale, [0.3, 1.0) is cosine distance [0.15, 0.5): one row
+        // from each arm.
+        let ids: Vec<i32> = search(2.0, None, (Some(0.3), Some(1.0)))
+            .await
+            .into_iter()
+            .map(|(id, _)| id)
+            .collect();
+        assert_eq!(
+            ids,
+            vec![3, 2],
+            "distance_range must apply on the scale it is given in"
+        );
+    }
+
     #[tokio::test]
     async fn test_vector_search_sstable_superseded_by_newer_sstable() {
         // An older SSTable's stale row must be suppressed by a newer

From 9b3ce075ae264d9a4bc6cdeb4afa4b3ecb9a4cf9 Mon Sep 17 00:00:00 2001
From: Daniel Rammer <hamersaw@protonmail.com>
Date: Sat, 10 Oct 2026 21:27:42 -0500
Subject: [PATCH 2/3] style(java): apply spotless to LsmVectorSearchPlanner
 javadoc

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../java/org/lance/memwal/LsmVectorSearchPlanner.java  | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java b/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java
index 34e424c49ba..4da5652e89e 100644
--- a/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java
+++ b/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java
@@ -117,9 +117,9 @@ private native void nativeCreate(
    * @param nprobes number of IVF partitions to probe
    * @param columns columns to project; pass {@code null} to return all columns plus {@code
    *     _distance}
-   * @param refineBaseTable when true, the base-table and SSTable arms re-rank candidates with
-   *     exact distances (refine factor 1), so every arm reports exact distances. Auto-enabled
-   *     whenever stale filtering is on (see {@code overfetchFactor}).
+   * @param refineBaseTable when true, the base-table and SSTable arms re-rank candidates with exact
+   *     distances (refine factor 1), so every arm reports exact distances. Auto-enabled whenever
+   *     stale filtering is on (see {@code overfetchFactor}).
    * @param overfetchFactor over-fetch multiple for sources with rows superseded by newer
    *     generations. Must be at least {@code 1.0}:
    *     <ul>
@@ -177,8 +177,8 @@ public ExecutionPlan planSearch(
    * @param nprobes number of IVF partitions to probe
    * @param columns columns to project; pass {@code null} to return all columns plus {@code
    *     _distance}
-   * @param refineBaseTable when true, the base-table and SSTable arms re-rank candidates with
-   *     exact distances.
+   * @param refineBaseTable when true, the base-table and SSTable arms re-rank candidates with exact
+   *     distances.
    * @return an executable plan
    */
   public ExecutionPlan planSearch(

From 9ac82a0a9d4a43b43af38527f6a3ed0a1dc07402 Mon Sep 17 00:00:00 2001
From: Daniel Rammer <hamersaw@protonmail.com>
Date: Sun, 11 Oct 2026 04:02:15 -0500
Subject: [PATCH 3/3] docs(mem_wal): trim distance-scale comments and fix
 refine auto-enable docs

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../lance/memwal/LsmVectorSearchPlanner.java  |  3 +-
 python/python/lance/mem_wal.py                |  5 +-
 rust/lance/src/dataset/mem_wal/index/hnsw.rs  |  7 ++-
 .../src/dataset/mem_wal/memtable/flush.rs     |  5 +-
 .../src/dataset/mem_wal/scanner/builder.rs    |  8 ++--
 .../src/dataset/mem_wal/scanner/projection.rs |  4 +-
 .../dataset/mem_wal/scanner/vector_search.rs  | 46 ++++++++-----------
 7 files changed, 31 insertions(+), 47 deletions(-)

diff --git a/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java b/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java
index 4da5652e89e..571bdedc5a5 100644
--- a/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java
+++ b/java/src/main/java/org/lance/memwal/LsmVectorSearchPlanner.java
@@ -118,8 +118,7 @@ private native void nativeCreate(
    * @param columns columns to project; pass {@code null} to return all columns plus {@code
    *     _distance}
    * @param refineBaseTable when true, the base-table and SSTable arms re-rank candidates with exact
-   *     distances (refine factor 1), so every arm reports exact distances. Auto-enabled whenever
-   *     stale filtering is on (see {@code overfetchFactor}).
+   *     distances. Auto-enabled when a newer generation blocks the base table.
    * @param overfetchFactor over-fetch multiple for sources with rows superseded by newer
    *     generations. Must be at least {@code 1.0}:
    *     <ul>
diff --git a/python/python/lance/mem_wal.py b/python/python/lance/mem_wal.py
index 04056199798..f0c735eb2e4 100644
--- a/python/python/lance/mem_wal.py
+++ b/python/python/lance/mem_wal.py
@@ -583,9 +583,8 @@ def plan_search(
             omitted.
         refine_base_table : bool, optional
             When true, the base-table and SSTable arms re-rank their candidates
-            with exact distances, so every arm reports exact distances.
-            Memtable arms are already exact. Auto-enabled whenever stale
-            filtering is on (see ``overfetch_factor``).
+            with exact distances. Auto-enabled when a newer generation blocks
+            the base table.
         overfetch_factor : float, optional
             Over-fetch multiple for sources with rows superseded by newer
             generations. Must be at least ``1.0``. At ``1.0`` each affected
diff --git a/rust/lance/src/dataset/mem_wal/index/hnsw.rs b/rust/lance/src/dataset/mem_wal/index/hnsw.rs
index 51fb7d45a1d..b98f65ca1eb 100644
--- a/rust/lance/src/dataset/mem_wal/index/hnsw.rs
+++ b/rust/lance/src/dataset/mem_wal/index/hnsw.rs
@@ -569,10 +569,9 @@ async fn write_sq_storage(
     };
     let row_ids = column(ROW_ID)?;
     let vectors = column(FLAT_COLUMN)?;
-    // A cosine index normalizes the query and ranks by L2, so it must store
-    // normalized vectors, as a regular cosine build does. The graph is
-    // unaffected: cosine over raw vectors orders neighbors the same as L2 over
-    // normalized ones.
+    // Cosine search ranks L2 against a normalized query, so stored vectors must
+    // be normalized. The graph is unaffected: raw cosine and normalized L2 order
+    // neighbors identically.
     let vectors: Arc<dyn Array> = if distance_type == DistanceType::Cosine {
         Arc::new(lance_linalg::kernels::normalize_fsl(
             vectors.as_fixed_size_list(),
diff --git a/rust/lance/src/dataset/mem_wal/memtable/flush.rs b/rust/lance/src/dataset/mem_wal/memtable/flush.rs
index ae47cbd7f32..bbc1e7cf3d0 100644
--- a/rust/lance/src/dataset/mem_wal/memtable/flush.rs
+++ b/rust/lance/src/dataset/mem_wal/memtable/flush.rs
@@ -1878,9 +1878,8 @@ mod tests {
         );
     }
 
-    /// A flushed cosine index must hold normalized vectors: its search
-    /// normalizes the query and ranks by L2, which only tracks cosine when the
-    /// stored side is normalized too.
+    /// A flushed cosine index ranks L2 against a normalized query, so it must
+    /// store normalized vectors.
     #[tokio::test]
     async fn test_flusher_cosine_index_ranks_unnormalized_vectors() {
         use super::super::super::index::IndexStore;
diff --git a/rust/lance/src/dataset/mem_wal/scanner/builder.rs b/rust/lance/src/dataset/mem_wal/scanner/builder.rs
index 261142e2c8a..ca4728587c1 100644
--- a/rust/lance/src/dataset/mem_wal/scanner/builder.rs
+++ b/rust/lance/src/dataset/mem_wal/scanner/builder.rs
@@ -48,8 +48,7 @@ struct LsmVectorQuery {
     k: usize,
     /// IVF partition probe bounds for indexed arms.
     probe_bounds: ProbeBounds,
-    /// Refine factor for the approximate-index arms; `None` leaves them on the
-    /// index's native distance scale.
+    /// Refine factor for the base and SSTable arms; `None` leaves them unrefined.
     refine_factor: Option<u32>,
     /// Distance metric; `None` defaults to L2 (matching the unindexed memtable arm).
     metric_type: Option<DistanceType>,
@@ -528,9 +527,8 @@ impl LsmScanner {
         self
     }
 
-    /// Re-rank the approximate-index arms' candidates by exact distance,
-    /// searching `refine_factor` times as many. `0` clears it. No-op unless
-    /// `nearest` was set.
+    /// Re-rank base and SSTable candidates by exact distance over `refine_factor`
+    /// times as many. `0` clears it; no-op unless `nearest` was set.
     pub fn refine(mut self, refine_factor: u32) -> Self {
         if let Some(q) = self.nearest.as_mut() {
             q.refine_factor = (refine_factor > 0).then_some(refine_factor);
diff --git a/rust/lance/src/dataset/mem_wal/scanner/projection.rs b/rust/lance/src/dataset/mem_wal/scanner/projection.rs
index ddc25c7210a..ca2bf7bc1aa 100644
--- a/rust/lance/src/dataset/mem_wal/scanner/projection.rs
+++ b/rust/lance/src/dataset/mem_wal/scanner/projection.rs
@@ -218,8 +218,8 @@ pub fn canonical_output_schema(
     Ok(Arc::new(Schema::new(fields)))
 }
 
-/// Wrap `plan` so `_distance` is multiplied by `factor`; all other columns are
-/// forwarded unchanged. A no-op when `factor` is `1.0`.
+/// Multiply `plan`'s `_distance` by `factor`, forwarding other columns; no-op
+/// at `1.0`.
 pub fn scale_distance(plan: Arc<dyn ExecutionPlan>, factor: f32) -> Result<Arc<dyn ExecutionPlan>> {
     if factor == 1.0 {
         return Ok(plan);
diff --git a/rust/lance/src/dataset/mem_wal/scanner/vector_search.rs b/rust/lance/src/dataset/mem_wal/scanner/vector_search.rs
index 20f737eb30d..8b22118c33c 100644
--- a/rust/lance/src/dataset/mem_wal/scanner/vector_search.rs
+++ b/rust/lance/src/dataset/mem_wal/scanner/vector_search.rs
@@ -132,17 +132,14 @@ pub struct LsmVectorSearchPlanner {
     /// generation's `IVF_HNSW_SQ` — and is ignored by an arm whose index is not
     /// a graph. `None` leaves each arm at its own default.
     ef: Option<usize>,
-    /// Refine factor for every approximate-index arm; see
-    /// [`Self::with_refine_factor`].
+    /// See [`Self::with_refine_factor`].
     refine_factor: Option<u32>,
-    /// Multiple of the exact distance every unrefined arm reports; see
-    /// [`Self::with_distance_scale`].
+    /// See [`Self::with_distance_scale`].
     distance_scale: f32,
 }
 
-/// Multiple of the exact distance a flushed generation's index reports. It is
-/// written as `IVF_HNSW_SQ`, which ranks cosine as L2 over normalized vectors,
-/// `2 * (1 - cos)`; the other metrics it ranks on their own scale.
+/// Multiple of the exact distance a flushed generation's `IVF_HNSW_SQ` index
+/// reports: 2 for cosine (L2 over normalized vectors), otherwise 1.
 fn sstable_distance_scale(distance_type: lance_linalg::distance::DistanceType) -> f32 {
     match distance_type {
         lance_linalg::distance::DistanceType::Cosine => 2.0,
@@ -214,27 +211,23 @@ impl LsmVectorSearchPlanner {
         self
     }
 
-    /// Set the refine factor. `Some(rf)` refines the base and SSTable arms by
-    /// `rf`, so every arm reports exact distances. `None` leaves them
-    /// unrefined; see [`Self::with_distance_scale`].
+    /// Refine the base and SSTable arms by `rf` so every arm reports exact
+    /// distances; `None` leaves them unrefined.
     pub fn with_refine_factor(mut self, refine_factor: Option<u32>) -> Self {
         self.refine_factor = refine_factor;
         self
     }
 
-    /// Report unrefined distances as `scale` times the exact distance, which
-    /// lines this plan up with a base search the caller merges it with: Lance
-    /// ranks cosine on SQ and PQ indexes as L2 over normalized vectors, twice
-    /// `1 - cos`. The memtable arm scales its exact distances, and the SSTable
-    /// arms convert from their own index's multiple. The distance range is
-    /// given on the same scale. Ignored when refining. Defaults to `1.0`.
+    /// Report unrefined distances, and take the distance range, as `scale`
+    /// times the exact distance. Cosine SQ/PQ indexes report `2 * (1 - cos)`,
+    /// so `2.0` matches such a base. Ignored when refining; defaults to `1.0`.
     pub fn with_distance_scale(mut self, scale: f32) -> Self {
         self.distance_scale = scale;
         self
     }
 
-    /// The factor that brings an arm reporting `native` times the exact
-    /// distance to the plan's scale.
+    /// Factor from an arm's `native` multiple of the exact distance to the
+    /// plan's scale; `1.0` when refining.
     fn arm_scale(&self, refine_factor: Option<u32>, native: f32) -> f32 {
         match refine_factor {
             Some(_) => 1.0,
@@ -297,10 +290,9 @@ impl LsmVectorSearchPlanner {
     /// * `k` - Number of nearest neighbors to return
     /// * `nprobes` - Exact number of IVF partitions to search (for IVF-based indexes)
     /// * `projection` - Columns to include in output (None = all columns)
-    /// * `refine_base_table` - When true and no refine factor is set, every
-    ///   approximate-index arm re-ranks its candidates with exact distances
-    ///   (refine factor 1), so the base table's are exact too. Auto-enabled
-    ///   whenever a newer generation blocks the base table (see below).
+    /// * `refine_base_table` - When true and no refine factor is set, refine
+    ///   the base and SSTable arms by 1. Auto-enabled when a newer generation
+    ///   blocks the base table.
     /// * `overfetch_factor` - Controls how aggressively sources over-fetch to
     ///   backfill the rows dropped by stale-row filtering. Values below `1.0`
     ///   are rejected; stale filtering is always enabled.
@@ -399,9 +391,8 @@ impl LsmVectorSearchPlanner {
             true, // include _distance — KNN always produces it
         )?;
 
-        // The merge ranks every arm by `_distance`, so all of them must report
-        // on one scale: exact when refining, `distance_scale` when not. A
-        // blocked base over-fetches its approximate candidates, so it refines.
+        // The merge needs every arm on one scale. A blocked base over-fetches
+        // approximate candidates, so it refines.
         // `block_lists` is non-empty exactly when a newer generation exists.
         let has_base = sources
             .iter()
@@ -3446,9 +3437,8 @@ mod tests {
         );
     }
 
-    /// The SSTable arm's `IVF_HNSW_SQ` index ranks cosine as `2 * (1 - cos)` and
-    /// the memtable arm ranks it exactly, so the plan must put both on the
-    /// requested scale or the merge interleaves them wrongly.
+    /// SSTable `IVF_HNSW_SQ` reports cosine as `2 * (1 - cos)` and the memtable
+    /// reports it exactly; both must land on the requested scale.
     #[tokio::test]
     async fn test_vector_search_cosine_arms_share_one_scale() {
         use crate::dataset::mem_wal::scanner::collector::{InMemoryMemTableRef, InMemoryMemTables};