[ENH] Add write segments operator

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

@@ -0,0 +1,22 @@
+use crate::execution::operator::Operator;
+use async_trait::async_trait;
+
+#[derive(Debug)]
+pub struct FlushS3Operator {}
+
+#[derive(Debug)]
+pub struct FlushS3Input {}
+
+#[derive(Debug)]
+pub struct FlushS3Output {}
+
+pub type WriteSegmentsResult = Result<FlushS3Output, ()>;
+
+#[async_trait]
+impl Operator<FlushS3Input, FlushS3Output> for FlushS3Operator {
+    type Error = ();
+
+    async fn run(&self, input: &FlushS3Input) -> WriteSegmentsResult {
+        Ok(FlushS3Output {})
+    }
+}

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

@@ -1,5 +1,7 @@
 pub(super) mod brute_force_knn;
+pub(super) mod flush_s3;
 pub(super) mod flush_sysdb;
 pub(super) mod normalize_vectors;
 pub(super) mod partition;
 pub(super) mod pull_log;
+pub(super) mod write_segments;

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

@@ -0,0 +1,22 @@
+use crate::execution::operator::Operator;
+use async_trait::async_trait;
+
+#[derive(Debug)]
+pub struct WriteSegmentsOperator {}
+
+#[derive(Debug)]
+pub struct WriteSegmentsInput {}
+
+#[derive(Debug)]
+pub struct WriteSegmentsOutput {}
+
+pub type WriteSegmentsResult = Result<WriteSegmentsOutput, ()>;
+
+#[async_trait]
+impl Operator<WriteSegmentsInput, WriteSegmentsOutput> for WriteSegmentsOperator {
+    type Error = ();
+
+    async fn run(&self, input: &WriteSegmentsInput) -> WriteSegmentsResult {
+        Ok(WriteSegmentsOutput {})
+    }
+}

rust/worker/src/execution/orchestration/compact.rs

@@ -46,6 +46,7 @@ enum ExecutionState {
     Partition,
     Write,
     Flush,
+    Register,
     Finished,
 }
 
@@ -62,6 +63,8 @@ pub struct CompactOrchestrator {
     sysdb: Box<dyn SysDb>,
     // Dispatcher
     dispatcher: Box<dyn Receiver<TaskMessage>>,
+    // number of write segments tasks
+    num_write_tasks: i32,
     // Result Channel
     result_channel:
         Option<tokio::sync::oneshot::Sender<Result<CompactionResponse, Box<dyn ChromaError>>>>,
@@ -96,6 +99,7 @@ impl CompactOrchestrator {
             log,
             sysdb,
             dispatcher,
+            num_write_tasks: 0,
             result_channel,
         }
     }
@@ -145,18 +149,24 @@ impl CompactOrchestrator {
     async fn write(&mut self, records: Vec<DataChunk>) {
         self.state = ExecutionState::Write;
 
+        self.num_write_tasks = records.len() as i32;
         for record in records {
             // TODO: implement write
         }
     }
 
+    async fn flush_s3(&mut self, self_address: Box<dyn Receiver<WriteSegmentsResult>>) {
+        self.state = ExecutionState::Flush;
+        // TODO: implement flush to s3
+    }
+
     async fn flush_sysdb(
         &mut self,
         log_position: i64,
         segment_flush_info: Vec<SegmentFlushInfo>,
         self_address: Box<dyn Receiver<FlushSysDbResult>>,
     ) {
-        self.state = ExecutionState::Flush;
+        self.state = ExecutionState::Register;
         let operator = FlushSysDbOperator::new();
         let input = FlushSysDbInput::new(
             self.compaction_job.tenant_id.clone(),
@@ -265,3 +275,25 @@ impl Handler<PartitionResult> for CompactOrchestrator {
         let _ = result_channel.send(Ok(response));
     }
 }
+
+#[async_trait]
+impl Handler<WriteSegmentsResult> for CompactOrchestrator {
+    async fn handle(
+        &mut self,
+        message: WriteSegmentsResult,
+        _ctx: &crate::system::ComponentContext<CompactOrchestrator>,
+    ) {
+        match message {
+            Ok(result) => {
+                // Log an error
+                self.num_write_tasks -= 1;
+            }
+            Err(e) => {
+                // Log an error
+            }
+        }
+        if self.num_write_tasks == 0 {
+            self.flush_s3(_ctx.sender.as_receiver()).await;
+        }
+    }
+}