Get rid of Initial operation state

rust/types/src/operation.rs

@@ -13,9 +13,6 @@ pub enum Operation {
 
 #[derive(Clone, Debug, PartialEq)]
 pub enum MaterializedLogOperation {
-    // Set when the record is initially read from the segment
-    // before it is processed based on state of the log.
-    Initial,
     // Set for records that don't exist in the segment and
     // have been encountered for the first time in the log.
     AddNew,

rust/worker/src/execution/operators/filter.rs

@@ -113,10 +113,7 @@ impl<'me> MetadataLogReader<'me> {
         let mut updated_offset_ids = RoaringBitmap::new();
         let mut user_id_to_offset_id = HashMap::new();
         for (log, _) in logs.iter() {
-            if !matches!(
-                log.final_operation,
-                MaterializedLogOperation::Initial | MaterializedLogOperation::AddNew
-            ) {
+            if !matches!(log.final_operation, MaterializedLogOperation::AddNew) {
                 updated_offset_ids.insert(log.offset_id);
             }
             if !matches!(

rust/worker/src/segment/distributed_hnsw_segment.rs

@@ -284,9 +284,6 @@ impl<'a> SegmentWriter<'a> for DistributedHNSWSegmentWriter {
                         }
                     }
                 }
-                MaterializedLogOperation::Initial => panic!(
-                    "Invariant violation. Mat records should not contain logs in initial state"
-                ),
             }
         }
         Ok(())

rust/worker/src/segment/metadata_segment.rs

@@ -685,7 +685,6 @@ impl<'log_records> SegmentWriter<'log_records> for MetadataSegmentWriter<'_> {
                         }
 
                 },
-                MaterializedLogOperation::Initial => panic!("Not expected mat records in the initial state")
             }
         }
         tracing::info!("Applied {} records to metadata segment", count,);

rust/worker/src/segment/record_segment.rs

@@ -370,7 +370,11 @@ impl<'a> SegmentWriter<'a> for RecordSegmentWriter {
                         .id_to_user_id
                         .as_ref()
                         .unwrap()
-                        .set::<u32, String>("", log_record.offset_id, log_record.user_id.unwrap().to_string())
+                        .set::<u32, String>(
+                            "",
+                            log_record.offset_id,
+                            log_record.user_id.unwrap().to_string(),
+                        )
                         .await
                     {
                         Ok(()) => (),
@@ -395,7 +399,8 @@ impl<'a> SegmentWriter<'a> for RecordSegmentWriter {
                     // Set max offset id.
                     max_new_offset_id = max_new_offset_id.max(log_record.offset_id);
                 }
-                MaterializedLogOperation::UpdateExisting | MaterializedLogOperation::OverwriteExisting => {
+                MaterializedLogOperation::UpdateExisting
+                | MaterializedLogOperation::OverwriteExisting => {
                     // Offset id and user id do not need to change. Only data
                     // needs to change. Blockfile does not have Read then write
                     // semantics so we'll delete and insert.
@@ -470,7 +475,6 @@ impl<'a> SegmentWriter<'a> for RecordSegmentWriter {
                         }
                     }
                 }
-                MaterializedLogOperation::Initial => panic!("Invariant violation. Materialized logs should not have any logs in the initial state")
             }
         }
         self.max_new_offset_id

rust/worker/src/server.rs

@@ -605,7 +605,9 @@ mod tests {
         assert_eq!(response.unwrap_err().code(), tonic::Code::InvalidArgument);
     }
 
-    fn gen_knn_request(mut scan_operator: Option<chroma_proto::ScanOperator>) -> chroma_proto::KnnPlan {
+    fn gen_knn_request(
+        mut scan_operator: Option<chroma_proto::ScanOperator>,
+    ) -> chroma_proto::KnnPlan {
         if scan_operator.is_none() {
             scan_operator = Some(scan());
         }