fix all obilivous errors

Author: shisoft

src/apps/hnsw/partition/search.rs

@@ -222,6 +222,18 @@ impl HnswOnlinePartition {
 
         match cell.data[0usize] {
             SharedValue::Id(id) => Some(*id),
+            SharedValue::Null | SharedValue::NA => {
+                append_job_log(
+                    logger,
+                    job_id,
+                    JobLogLevel::Error,
+                    format!(
+                        "Cannot find vector cell in vertex id {:?} with body {:?}",
+                        vertex_id, cell.data
+                    ),
+                );
+                None
+            }
             _ => {
                 append_job_log(
                     logger,
@@ -315,15 +327,19 @@ impl HnswOnlinePartition {
         let selected_vertex = frontier
             .iter()
             .filter_map(|vertex_id| {
-                self.get_vertex_distance(
+                let (_cell_id, distance) = self.get_vertex_distance(
                     vertex_id, query, field_id, metric, metadata, chunks, logger, job_id,
-                )
+                )?;
+                Some((*vertex_id, distance))
             })
-            .filter(|(id, _)| !metadata.visited.contains(id))
+            .filter(|(vid, _)| !metadata.visited.contains(vid))
             // Greedy select the closest vertex
             .min_by(|(_, d1), (_, d2)| d1.partial_cmp(d2).unwrap());
         let selected_vertex = match selected_vertex {
-            Some(vertex) => vertex,
+            Some(vertex) => {
+                info!("Selected vertex at level {} is {:?}", metadata.level, vertex);
+                vertex
+            },
             None => {
                 append_job_log(
                     logger,
@@ -800,12 +816,22 @@ impl HnswOnlinePartition {
             let level_entry = metadata.level_entries.get(&level);
             if let Some(entry_point) = level_entry {
                 entry_points.push(*entry_point);
+                info!("Level {} entry point is {:?}", level, entry_point);
             }
             if level == 0 || level_entry.is_none() {
                 // For the lowest level, we need to use the search history
                 entry_points.extend(metadata.history.iter().map(|(id, _)| *id).take(1));
+                info!(
+                    "Entry point at level {} is {:?}, history is {:?}",
+                    level,
+                    entry_points,
+                    metadata
+                        .history
+                        .iter()
+                        .map(|(id, _)| *id)
+                        .collect::<Vec<_>>()
+                );
             }
-
             if entry_points.is_empty() {
                 // No entry points, we can't connect anything
                 append_job_log(

src/apps/hnsw/partition/tests.rs

@@ -7,6 +7,7 @@ use dovahkiin::data_map;
 use dovahkiin::types::{Id, OwnedMap, OwnedPrimArray, OwnedValue, Type};
 use lightning::map::{Map, PtrHashMap};
 use neb::client::AsyncClient;
+use neb::exec::query::env;
 use neb::ram::cell::{self, OwnedCell};
 use neb::ram::chunk::Chunks;
 use neb::ram::schema::{Field, Schema as NebSchema};
@@ -74,6 +75,7 @@ pub struct TestEnvironment {
 impl TestEnvironment {
     /// Create a new test environment with test vectors
     pub fn new(server_port: u32, partition_id: u64, field_id: u64, schema_id: u32) -> Self {
+        env_logger::init();
         Self {
             job_id: None,
             server_port,
@@ -97,7 +99,7 @@ impl TestEnvironment {
             .enumerate()
             .map(|(i, vector)| {
                 let cell_higher = self.partition_id; // Most share the same partition ID
-                let cell_lower = 2000 + i as u64; // Unique cell IDs
+                let cell_lower = 1000 + i as u64; // Unique cell IDs
                 TestVector::new(cell_higher, cell_lower, OwnedPrimArray::F32(vector))
             })
             .collect();
@@ -426,6 +428,24 @@ async fn index_one_cell() {
     env.create_index().await.unwrap();
     env.index_cells().await.unwrap();
 }
+
+#[tokio::test]
+async fn index_two_cells() {
+    let env = TestEnvironment::new(5001, 1, VECTOR_FIELD_ID, CELL_SCHEMA_ID)
+        .with_test_vectors(vec![vec![1.0, 2.0, 3.0], vec![4.0, 5.0, 6.0]])
+        .with_job_logger()
+        .with_job_id(JobId::new(1, 1))
+        .initialize_server()
+        .await
+        .initialize_schemas()
+        .await
+        .unwrap()
+        .initialize_cells()
+        .await
+        .unwrap();
+    env.create_index().await.unwrap();
+    env.index_cells().await.unwrap();
+}
 // Example of how to set up a functional test (commented out since it needs user implementation)
 /*
 #[async_std::test]

src/graph/id_list.rs

@@ -198,31 +198,40 @@ impl<'a> IdList<'a> {
         if let Some(id) = self.cached_type_list_id {
             return Ok(Ok(id));
         }
-
+        let field_id = self.field_id;
         // Read the type list ID from container
         let fields = match self
             .txn
-            .read_selected(self.container_id, vec![self.field_id])
+            .read_selected(self.container_id, vec![field_id])
             .await?
         {
             Some(f) => f,
             None => return Ok(Err(IdListError::ContainerCellNotFound)),
         };
-
+        let mut create_new_type_list = async || {
+            let result = self.create_new_type_list().await?;
+            if let Ok(new_id) = result {
+                self.cached_type_list_id = Some(new_id);
+            }
+            Ok(result)
+        };
         match &fields[0usize] {
             OwnedValue::Id(id) => {
                 if id.is_unit_id() && ensure_container {
-                    let result = self.create_new_type_list().await?;
-                    if let Ok(new_id) = result {
-                        self.cached_type_list_id = Some(new_id);
-                    }
-                    Ok(result)
+                    create_new_type_list().await
                 } else {
                     self.cached_type_list_id = Some(*id);
                     Ok(Ok(*id))
                 }
             }
-            _ => Ok(Err(IdListError::FormatError)),
+            OwnedValue::Null | OwnedValue::NA => {
+                trace!("Creating new type list for {}", field_id);
+                create_new_type_list().await
+            }
+            _ => {
+                error!("Failed to get type list id, got {:?}", fields);
+                Ok(Err(IdListError::FormatError))
+            }
         }
     }
 
@@ -253,9 +262,12 @@ impl<'a> IdList<'a> {
         ensure_container: bool,
     ) -> Result<Result<Id, IdListError>, TxnError> {
         // Read the type list cell
-        let mut type_list_cell = match self.txn.read(type_list_id).await? {
+        let type_list_cell = match self.txn.read(type_list_id).await? {
             Some(cell) => cell,
-            None => return Ok(Err(IdListError::FormatError)),
+            None => {
+                error!("Failed to get type list cell, id {:?}", type_list_id);
+                return Ok(Err(IdListError::FormatError));
+            }
         };
 
         // Find existing list for this schema
@@ -264,6 +276,10 @@ impl<'a> IdList<'a> {
                 return Ok(Ok(list_id));
             }
         } else {
+            error!(
+                "Failed to get type list, index {:?} got {:?}",
+                *ID_TYPES_MAP_ID, type_list_cell
+            );
             return Ok(Err(IdListError::FormatError));
         }
 
@@ -312,6 +328,10 @@ impl<'a> IdList<'a> {
             self.txn.update(type_list_cell).await?;
             Ok(Ok(list_id))
         } else {
+            error!(
+                "Failed to get type list, index {:?} got {:?}",
+                *ID_TYPES_MAP_ID, type_list_cell
+            );
             Ok(Err(IdListError::FormatError))
         }
     }
@@ -490,9 +510,11 @@ impl<'a> IdList<'a> {
                             };
                             array.remove(index);
                         } else {
+                            error!("Failed to get list, index {:?} got {:?}", *LIST_KEY_ID, map);
                             return Ok(Err(IdListError::FormatError));
                         }
                     } else {
+                        error!("Failed to get list, {:?}", seg.data);
                         return Ok(Err(IdListError::FormatError));
                     }
                     self.txn.update(seg).await.unwrap();

src/graph/vertex/mod.rs

@@ -60,7 +60,10 @@ where
                     let (type_list_id, schemas_ids) =
                         match IdList::cell_types(&txn, id, field_id).await? {
                             Some(t) => t,
-                            None => return Ok(Err(RemoveError::FormatError)),
+                            None => {
+                                error!("Failed to get type list, {:?}", id);
+                                return Ok(Err(RemoveError::FormatError));
+                            }
                         };
                     for schema_id in schemas_ids {
                         let mut id_list =

src/job/logger.rs

@@ -44,7 +44,11 @@ pub fn append_job_log(
             });
         }
         None => {
-            error!("Failed to get job logger by job id: {:?}", job_id);
+            error!(
+                "Failed to get job logger by job id: {:?}, inserting new job logger",
+                job_id
+            );
+            job_logger.try_insert(job_id, Arc::new(Mutex::new(Vec::new())));
         }
     }
 }