support more than one vertex per cell

Author: shisoft

src/apps/hnsw/coordinator/mod.rs

@@ -24,8 +24,7 @@ use crate::{
 use super::{
     measurements::MetricEncoding,
     partition::{
-        index::HNSWIndex, service::service::AsyncServiceClient as PartitionSvrClient,
-        types::OrderedFloat,
+        index::HNSWIndex, schema::{CELL_FIELD_ID, HNSW_VERTEX_SCHEMA_ID}, service::service::AsyncServiceClient as PartitionSvrClient, types::OrderedFloat
     },
 };
 use bifrost::service_with_id;
@@ -39,7 +38,7 @@ service! {
     rpc delete_index(schema: u32, field_id: u64) -> Result<(), String>;
 
     rpc new_cell(schema: u32, field_id: u64, cell_id: Id, ef_construction: u64, metric: MetricEncoding) -> Result<(), String>;
-    rpc del_cell(schema: u32, field_id: u64, cell_id: Id) -> Result<(), String>;
+    rpc del_cell(cell_id: Id) -> Result<(), String>;
     rpc update_cell(schema: u32, field_id: u64, cell_id: Id, ef_construction: u64, metric: MetricEncoding) -> Result<(), String>;
 
     rpc query_top(schema: u32, field_id: u64, query: OwnedPrimArray, ef: u64, max_iter: u8, metric: MetricEncoding) -> Result<Option<(Id, Distance)>, String>;
@@ -245,27 +244,32 @@ impl Service for HNSWIndexService {
 
     fn del_cell<'a>(
         &'a self,
-        schema: u32,
-        field_id: u64,
         cell_id: Id,
     ) -> BoxFuture<'a, Result<(), String>> {
         async move {
+            // Use the hash index to get the vertex id by the cell id
             let cell_id_value = OwnedValue::Id(cell_id);
-            let index_id = get_hash_id_from_value(schema, field_id, &cell_id_value);
+            let index_id = get_hash_id_from_value(HNSW_VERTEX_SCHEMA_ID, CELL_FIELD_ID, &cell_id_value);
             let index_vertex = self
                 .indexer
-                .hashed_query(index_id, field_id, &cell_id_value)
+                .hashed_query(index_id, CELL_FIELD_ID, &cell_id_value)
                 .await
                 .map_err(|e| format!("Failed to get index client: {:?}", e))?
                 .map_err(|e| format!("Failed to get index client: {:?}", e))?;
             self.engine.graph_transaction(DefaultPartitioner, |txn| {
+                let index_vertex = index_vertex.clone();
                 async move {
-                    txn.remove_vertex(index_vertex).await
+                    for vid in &index_vertex {
+                        if let Err(e) = txn.remove_vertex(vid).await? {
+                            return Ok(Err(e));
+                        }
+                    }
+                    Ok(Ok(()))
                 }
             })
             .await
-            .map_err(|e| format!("Failed to delete cell: {:?}", e))?
-            .map_err(|e| format!("Failed to delete cell: {:?}", e))?;
+            .map_err(|e| format!("Failed to delete vertex: {:?}", e))?
+            .map_err(|e| format!("Failed to delete vertex: {:?}", e))?;
             Ok(())
         }
         .boxed()
@@ -280,7 +284,7 @@ impl Service for HNSWIndexService {
         metric: MetricEncoding,
     ) -> BoxFuture<'a, Result<(), String>> {
         async move {
-            let _ = self.del_cell(schema, field_id, cell_id).await;
+            let _ = self.del_cell(cell_id).await;
             self.new_cell(schema, field_id, cell_id, ef_construction, metric).await
         }
         .boxed()