fix: read_all_batches schema discovery and test store root

- Auto-detect parquet schema from first file for proper column projection
  when using LakeSoulReader in read_all_batches
- Use LocalFileSystem::new() (root at /) in tests so absolute index_prefix
  paths work correctly (matching Python behavior)
- Mark test_read_parquet_schema and test_build_and_list_files as #[ignore]
  (require prepare_data.py)

Signed-off-by: chenxu <chenxu@dmetasoul.com>

Author: dmetasoul01

python/uv.lock

@@ -195,13 +195,35 @@ impl VectorShardIndexBuilder {
 
         let prefix = self.table_prefix();
 
+        // Read the schema from the first parquet file so the reader knows
+        // which columns to project.
+        let schema = self
+            .file_paths
+            .first()
+            .and_then(|p| {
+                let path = p
+                    .trim_start_matches("file://")
+                    .trim_start_matches("s3://")
+                    .trim_start_matches("s3a://");
+                std::fs::File::open(path).ok()
+            })
+            .and_then(|f| {
+                parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(f)
+                    .ok()
+            })
+            .map(|b| b.schema().clone());
+
         let mut config_builder = LakeSoulIOConfigBuilder::new()
             .with_files(self.file_paths.clone())
             .with_prefix(prefix)
             .with_primary_keys(vec![pk_col.clone()])
             .with_batch_size(8192)
             .with_thread_num(1);
 
+        if let Some(s) = schema {
+            config_builder = config_builder.with_schema(s);
+        }
+
         // Apply object store options
         for (k, v) in &self.object_store_options {
             config_builder =

rust/lakesoul-io/src/vector/builder.rs

@@ -196,7 +196,6 @@ pub fn extract_vector_batch(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use arrow_array::types::Float32Type;
     use std::sync::Arc;
 
     fn make_fixed_size_list_batch(
@@ -205,7 +204,6 @@ mod tests {
         dim: usize,
     ) -> RecordBatch {
         let id_array = Arc::new(UInt64Array::from(ids)) as arrow_array::ArrayRef;
-        let n = vectors.len();
 
         // Build FixedSizeList: flatten all values
         let flat: Vec<f32> = vectors.iter().flatten().copied().collect();

rust/lakesoul-io/src/vector/reader.rs

@@ -28,9 +28,7 @@ const DIM: usize = 200;
 
 /// Helper: read .fvecs file into Vec<Vec<f32>>
 fn read_fvecs(path: &str, n: Option<usize>) -> Vec<Vec<f32>> {
-    use std::io::Seek;
     let mut f = std::fs::File::open(path).unwrap();
-    let f_len = f.metadata().unwrap().len() as usize;
     let mut read_one = || -> Option<Vec<f32>> {
         let mut dim_buf = [0u8; 4];
         f.read_exact(&mut dim_buf).ok()?;
@@ -97,7 +95,7 @@ fn compute_recall(predicted: &[u64], ground_truth: &[i32], k: usize) -> f64 {
 #[ignore = "requires prepare_data.py to be run first"]
 async fn test_glove_e2e_build_and_search() {
     let tmp = TempDir::new().unwrap();
-    let store = Arc::new(LocalFileSystem::new_with_prefix(tmp.path()).unwrap());
+    let store = Arc::new(LocalFileSystem::new());
     let index_prefix = "_vector_index/vec/-5/0/";
 
     let config = VectorIndexConfig {
@@ -156,8 +154,8 @@ async fn test_glove_e2e_build_and_search() {
 
 /// Quick test: verify LakeSoulReader can read our parquet schema
 #[tokio::test]
+#[ignore = "requires prepare_data.py to be run first"]
 async fn test_read_parquet_schema() {
-    use futures::StreamExt;
     use lakesoul_io::config::LakeSoulIOConfigBuilder;
     use lakesoul_io::reader::LakeSoulReader;
 
@@ -208,9 +206,10 @@ async fn test_read_parquet_schema() {
 
 /// Quick test: build and verify files on disk
 #[tokio::test]
+#[ignore = "requires prepare_data.py to be run first"]
 async fn test_build_and_list_files() {
     let tmp = TempDir::new().unwrap();
-    let store = Arc::new(LocalFileSystem::new_with_prefix(tmp.path()).unwrap());
+    let store = Arc::new(LocalFileSystem::new());
     let index_prefix = "_vector_index/vec/-5/0/";
 
     let config = VectorIndexConfig {
@@ -267,12 +266,10 @@ async fn test_reader_with_vector_search() {
         OPTION_KEY_VECTOR_SEARCH_TOP_K,
     };
     use lakesoul_io::reader::LakeSoulReader;
-    use std::io::Read;
 
     let tmp = TempDir::new().unwrap();
     let tmp_path = tmp.path().to_str().unwrap().to_string();
     let parquet_path = format!("{}/part-abc_0000.parquet", tmp_path);
-    let index_prefix = "_vector_index/vec/-5/0/";
     let pk_col = "id";
     let vec_col = "vec";
 
@@ -325,7 +322,7 @@ async fn test_reader_with_vector_search() {
     }
 
     // 2. Build vector index
-    let store = Arc::new(LocalFileSystem::new_with_prefix(tmp.path()).unwrap());
+    let store = Arc::new(LocalFileSystem::new());
     let config = VectorIndexConfig {
         column_name: vec_col.to_string(),
         dim,

rust/lakesoul-io/tests/vector_e2e_test.rs

@@ -47,3 +47,6 @@ tempfile = { workspace = true }
 
 [lints]
 workspace = true
+
+[features]
+huge_pages = []

rust/lakesoul-vector/Cargo.toml

@@ -128,7 +128,7 @@ impl IvfRabitqBuilder {
                     _ => crate::rabitq::manifest::load_manifest(mstore).await?,
                 };
             let mut clusters = Vec::with_capacity(cluster_map.len());
-            for (_cid, entry) in cluster_map.iter() {
+            for entry in cluster_map.values() {
                 // Read centroid from the base segment (version 0).
                 let base = entry.base_segment().ok_or_else(|| {
                     RabitqError::InvalidPersistence("cluster has no base segment")

rust/lakesoul-vector/src/rabitq/ivf/builder.rs

@@ -554,10 +554,7 @@ impl IvfRabitqIndex {
                     );
                 }
 
-                (
-                    indices.to_vec(),
-                    quantized_vectors,
-                )
+                (indices.to_vec(), quantized_vectors)
             })
             .collect();
         println!("Quantization complete");
@@ -1431,9 +1428,10 @@ impl IvfRabitqIndex {
 
                 // Apply filter if provided
                 if let Some(filter_bitmap) = filter
-                    && !filter_bitmap.contains(vector_id as u32) {
-                        continue;
-                    }
+                    && !filter_bitmap.contains(vector_id as u32)
+                {
+                    continue;
+                }
 
                 // Use pre-computed values (vectorized above)
                 let ip_x0_qr = ip_x0_qr_values[i];
@@ -1551,9 +1549,10 @@ impl IvfRabitqIndex {
             .zip(cluster.pending_vectors.iter())
         {
             if let Some(bitmap) = filter
-                && !bitmap.contains(vec_id as u32) {
-                    continue;
-                }
+                && !bitmap.contains(vec_id as u32)
+            {
+                continue;
+            }
 
             // Unpack binary code
             let binary_code = qvec.unpack_binary_code();
@@ -1606,7 +1605,7 @@ impl IvfRabitqIndex {
     }
 
     #[cfg(test)]
-    pub(crate) fn search_with_diagnostics(
+    pub(crate) fn _search_with_diagnostics(
         &self,
         query: &[f32],
         params: SearchParams,
@@ -1618,7 +1617,7 @@ impl IvfRabitqIndex {
     }
 
     #[cfg(test)]
-    pub(crate) fn search_naive(
+    pub(crate) fn _search_naive(
         &self,
         query: &[f32],
         params: SearchParams,
@@ -2712,7 +2711,7 @@ impl ClusterData {
     }
 
     /// Reconstruct all `QuantizedVector` values (batched + pending).
-    fn collect_quantized_vectors(&self, padded_dim: usize) -> Vec<QuantizedVector> {
+    fn _collect_quantized_vectors(&self, padded_dim: usize) -> Vec<QuantizedVector> {
         let ex_bits = self.ex_bits;
         let dim_bytes = padded_dim / 8;
         let mut result = Vec::with_capacity(self.total_vectors());
@@ -3023,7 +3022,7 @@ mod batch_search_tests {
         query_precomp.build_lut(padded_dim);
 
         // Debug LUT parameters
-        if let Some(lut) = query_precomp.lut {
+        if let Some(ref lut) = query_precomp.lut {
             println!("\n=== L2 Test LUT params ===");
             println!(
                 "delta={:.6}, sum_vl_lut={:.6}, k1x_sum_q={:.6}",

rust/lakesoul-vector/src/rabitq/ivf/mod.rs

@@ -524,7 +524,10 @@ fn assign_points_for_update(
             state.counts[cluster] += 1;
             let vector = &data_chunk[row * dim..(row + 1) * dim];
             let sum_offset = cluster * dim;
-            for (sum, &v) in state.sums[sum_offset..sum_offset + dim].iter_mut().zip(vector.iter()) {
+            for (sum, &v) in state.sums[sum_offset..sum_offset + dim]
+                .iter_mut()
+                .zip(vector.iter())
+            {
                 *sum += v;
             }
             insert_candidate(&mut chunk_candidates, (dist, row));
@@ -553,13 +556,13 @@ fn insert_candidate(candidates: &mut Vec<(f32, usize)>, candidate: (f32, usize))
         return;
     }
     if let Some((last_dist, _)) = candidates.last()
-        && candidate.0 > *last_dist {
-            candidates.pop();
-            candidates.push(candidate);
-            candidates.sort_unstable_by(|a, b| {
-                b.0.partial_cmp(&a.0).unwrap_or(Ordering::Equal)
-            });
-        }
+        && candidate.0 > *last_dist
+    {
+        candidates.pop();
+        candidates.push(candidate);
+        candidates
+            .sort_unstable_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(Ordering::Equal));
+    }
 }
 
 fn update_centroids(

rust/lakesoul-vector/src/rabitq/kmeans.rs

@@ -141,7 +141,7 @@ fn hup_ref(h: &mut Option<&mut Hasher>, data: &[u8]) {
 
 // ---- conversions ----
 
-fn u2u64(v: usize) -> Result<u64, RabitqError> {
+fn _u2u64(v: usize) -> Result<u64, RabitqError> {
     u64::try_from(v).map_err(|_| RabitqError::InvalidPersistence("usize exceeds u64"))
 }
 fn uf64(v: u64) -> Result<usize, RabitqError> {