[ENH]: Make all functions incremental

Author: tanujnay112

rust/segment/src/types.rs

@@ -125,7 +125,7 @@ impl ChromaError for LogMaterializerError {
 #[derive(Debug)]
 struct MaterializedLogRecord {
     // False if the record exists only in the log, otherwise true.
-    offset_id_exists_in_segment: bool,
+    pub offset_id_exists_in_segment: bool,
     // If present in the record segment then it is the offset id
     // in the record segment at which the record was found.
     // If not present in the segment then it is the offset id
@@ -504,26 +504,30 @@ impl MaterializeLogsResult {
     /// # Note
     /// This is primarily intended for testing and should not be used in production code.
     /// Use the `materialize_logs` function instead for proper log materialization.
-    pub fn from_logs_for_test(logs: Chunk<LogRecord>) -> Result<Self, LogMaterializerError> {
-        let mut materialized = Vec::new();
-        for (index, (log_record, _)) in logs.iter().enumerate() {
-            let offset_id = (index + 1) as u32;
-            let mut mat_record =
-                MaterializedLogRecord::from_log_record(offset_id, index, log_record)?;
-
-            // Override the operation for delete records
-            if log_record.record.operation == Operation::Delete {
-                mat_record.final_operation = MaterializedLogOperation::DeleteExisting;
-                mat_record.final_document_at_log_index = None;
-                mat_record.final_embedding_at_log_index = None;
-            }
-
-            materialized.push(mat_record);
-        }
-        Ok(Self {
-            logs,
-            materialized: Chunk::new(materialized.into()),
-        })
+    pub async fn from_logs_for_test(
+        logs: Chunk<LogRecord>,
+        record_segment_reader: &Option<RecordSegmentReader<'_>>,
+    ) -> Result<Self, LogMaterializerError> {
+        // let mut materialized = Vec::new();
+        // for (index, (log_record, _)) in logs.iter().enumerate() {
+        //     let offset_id = (index + 1) as u32;
+        //     let mut mat_record =
+        //         MaterializedLogRecord::from_log_record(offset_id, index, log_record)?;
+
+        //     // Override the operation for delete records
+        //     if log_record.record.operation == Operation::Delete {
+        //         mat_record.final_operation = MaterializedLogOperation::DeleteExisting;
+        //         mat_record.final_document_at_log_index = None;
+        //         mat_record.final_embedding_at_log_index = None;
+        //     }
+
+        //     materialized.push(mat_record);
+        // }
+        // Ok(Self {
+        //     logs,
+        //     materialized: Chunk::new(materialized.into()),
+        // })
+        materialize_logs(record_segment_reader, logs, None).await
     }
 }
 

rust/types/src/operation.rs

@@ -32,6 +32,18 @@ pub enum MaterializedLogOperation {
     DeleteExisting,
 }
 
+impl std::fmt::Display for MaterializedLogOperation {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            MaterializedLogOperation::Initial => write!(f, "Initial"),
+            MaterializedLogOperation::AddNew => write!(f, "AddNew"),
+            MaterializedLogOperation::OverwriteExisting => write!(f, "OverwriteExisting"),
+            MaterializedLogOperation::UpdateExisting => write!(f, "UpdateExisting"),
+            MaterializedLogOperation::DeleteExisting => write!(f, "DeleteExisting"),
+        }
+    }
+}
+
 #[derive(Error, Debug)]
 pub enum OperationConversionError {
     #[error("Invalid operation, valid operations are: Add, Upsert, Update, Delete")]