properly initialize vector indexer

Author: shisoft

src/apps/hnsw/coordinator/mod.rs

@@ -7,7 +7,6 @@ use ahash::HashSet;
 use bifrost::{
     conshash::ConsistentHashing,
     dispatch_rpc_service_functions,
-
     raft::client::RaftClient,
     rpc::cluster::{
         all_server_ids, broadcast_to_members, broadcast_with_server_ids, client_by_server_id,

src/apps/hnsw/mod.rs

@@ -8,26 +8,16 @@ use futures::{
     TryFutureExt,
 };
 use neb::{
-    index::{
-        builder::IndexError,
-        vector::{set_vector_index_core, VectorIndexerCore},
-    },
-
+    index::{builder::IndexError, vector::VectorIndexerCore},
     server::NebServer,
 };
 
-
-use crate::{
-    apps::hnsw::coordinator::AsyncServiceClient,
-    server::MorpheusServer,
-};
-use neb::index::vector::MetricEncoding;
-use dovahkiin::types::OwnedValue;
+use crate::{apps::hnsw::coordinator::AsyncServiceClient, server::MorpheusServer};
 use dovahkiin::types::Map;
+use dovahkiin::types::OwnedValue;
+use neb::index::vector::MetricEncoding;
 use neb::ram::types::OwnedPrimArray;
 
-
-
 pub mod coordinator;
 pub mod measurements;
 pub mod partition;
@@ -87,8 +77,8 @@ impl VectorIndexerCore for VectorIndexer {
         self.coordinator
             .del_cell(*cell_id)
             .map_err(IndexError::RPCError)
-            .and_then(|r| async move { 
-                //r.map_err(IndexError::Other) 
+            .and_then(|r| async move {
+                //r.map_err(IndexError::Other)
                 Ok(()) // Do not output index deletion errors
             })
             .boxed()
@@ -128,13 +118,19 @@ impl VectorIndexer {
         }
     }
 
-    pub fn set_core(self) {
-        set_vector_index_core(self);
+    pub fn set_core(self, neb_server: &NebServer) {
+        neb_server
+            .indexer
+            .as_ref()
+            .unwrap()
+            .clients
+            .vector_client
+            .set_vector_index_core(self);
     }
 
     pub async fn new_and_set_core(morph: &MorpheusServer) {
         let indexer = Self::new(morph).await;
-        indexer.set_core();
+        indexer.set_core(&morph.neb_server);
     }
 
     pub fn set_ef_construction(&mut self, ef_construction: u64) {
@@ -147,8 +143,18 @@ unsafe impl Sync for VectorIndexer {}
 
 #[cfg(test)]
 mod e2e_tests {
-    use dovahkiin::{data_map, types::{OwnedPrimArray, Type}};
-    use neb::{index::vector::{self, MetricEncoding}, ram::{cell::OwnedCell, schema::{Field, IndexType, Schema}, types::RandValue}};
+    use dovahkiin::{
+        data_map,
+        types::{OwnedPrimArray, Type},
+    };
+    use neb::{
+        index::vector::{self, MetricEncoding},
+        ram::{
+            cell::OwnedCell,
+            schema::{Field, IndexType, Schema},
+            types::RandValue,
+        },
+    };
 
     use crate::{apps::hnsw::coordinator::Service, tests::start_server};
 
@@ -159,46 +165,66 @@ mod e2e_tests {
     const CELL_SCHEMA_ID: u32 = 123;
 
     #[tokio::test]
-    async fn test_e2e() {
-        let server = start_server(5010, "vector_indexer_e2e_test").await.unwrap();
+    async fn single_cell() {
+        let server = start_server(5910, "vector_indexer_e2e_test").await.unwrap();
         let _partition = server.init_hnsw_index_partition_service().await.unwrap();
         let coordinator = server.init_hnsw_index_service().await.unwrap();
         let indexer = VectorIndexer::new(&server).await;
-        indexer.set_core();
-        assert!(vector::is_vector_index_core_set());
+        indexer.set_core(&server.neb_server);
+        assert!(server
+            .neb_server
+            .indexer
+            .as_ref()
+            .unwrap()
+            .clients
+            .vector_client
+            .is_vector_index_core_set());
 
         let neb_client = server.neb_client.clone();
         let cell_schema = Schema::new_with_id(
-            CELL_SCHEMA_ID, "test_schema", None, 
-            Field::new_schema(
-                vec![
-                    Field::new_indexed_array(
-                        VECTOR, 
-                        Type::F32,
-                        vec![IndexType::Vector(MetricEncoding::L2)]
-                    )
-                ]
-            ), 
-            false, false
+            CELL_SCHEMA_ID,
+            "test_schema",
+            None,
+            Field::new_schema(vec![Field::new_indexed_array(
+                VECTOR,
+                Type::F32,
+                vec![IndexType::Vector(MetricEncoding::L2)],
+            )]),
+            false,
+            false,
         );
-        neb_client.new_schema_with_id(cell_schema).await.unwrap().unwrap();
+        neb_client
+            .new_schema_with_id(cell_schema)
+            .await
+            .unwrap()
+            .unwrap();
         let cell_id = Id::rand();
         let vector1 = vec![1.0, 2.0, 3.0];
         let map = data_map!(VECTOR: OwnedValue::PrimArray(OwnedPrimArray::F32(vector1.clone())));
         let cell1 = OwnedCell::new_with_id(CELL_SCHEMA_ID, &cell_id, OwnedValue::Map(map));
-        assert!(vector::is_vector_index_core_set());
+        assert!(server
+            .neb_server
+            .indexer
+            .as_ref()
+            .unwrap()
+            .clients
+            .vector_client
+            .is_vector_index_core_set());
         neb_client.write_cell(cell1).await.unwrap().unwrap();
-        let query = coordinator.query_top_k(
-            CELL_SCHEMA_ID, 
-            VECTOR_FIELD_ID, 
-            OwnedPrimArray::F32(vector1), 
-            10, 
-            32,
-            10, 
-            MetricEncoding::L2
-        ).await.unwrap();
+        let query = coordinator
+            .query_top_k(
+                CELL_SCHEMA_ID,
+                VECTOR_FIELD_ID,
+                OwnedPrimArray::F32(vector1),
+                10,
+                32,
+                10,
+                MetricEncoding::L2,
+            )
+            .await
+            .unwrap();
         assert_eq!(query.len(), 1);
         assert_eq!(query[0].0, cell_id);
         assert_eq!(query[0].1, 0.0);
     }
-}
+}

src/apps/hnsw/partition/search.rs

@@ -8,25 +8,21 @@ use std::{
 
 use ahash::{AHashSet, HashMap};
 use bifrost::{conshash::ConsistentHashing, raft::client::RaftClient};
-use dovahkiin::types::{Map as DovMap};
+use dovahkiin::types::Map as DovMap;
 use dovahkiin::{
     data_map,
     types::{Id, OwnedPrimArray, SharedValue},
 };
 use lightning::map::Map;
 use lightning::map::PtrHashMap;
-use log4rs::append;
 use neb::{
     index::{ranged::trees::Ordering, IndexerClients},
     query::data_client::IndexedDataClient,
     ram::{
-        cell::{Cell, OwnedCell},
         chunk::Chunks,
         types::RandId,
     },
 };
-use parking_lot::RwLock;
-use rand::{seq::SliceRandom, Rng};
 
 use crate::{
     apps::hnsw::measurements::{vector_distance, HnswMetric, Metric, MetricEncoding},