[PERF] Make pull logs an I/O operator so it runs on main runtime (#2641)

## Description of changes

*Summarize the changes made by this PR.*
 - Improvements & Bug fixes
- This moves pull logs to use the IO operator functionality introduced
in #2603
	 - Renamed IoOperator to IO since its a bit redundant. 
 - New functionality
	 - None

## Test plan
*How are these changes tested?*
Existing tests cover this change
- [x] Tests pass locally with `pytest` for python, `yarn test` for js,
`cargo test` for rust

## Documentation Changes
None

rust/worker/src/execution/dispatcher.rs

@@ -95,7 +95,7 @@ impl Dispatcher {
     /// - task: The task to enqueue
     async fn enqueue_task(&mut self, task: TaskMessage) {
         match task.get_type() {
-            OperatorType::IoOperator => {
+            OperatorType::IO => {
                 tokio::spawn(async move {
                     task.run().await;
                 });
@@ -292,7 +292,7 @@ mod tests {
         }
 
         fn get_type(&self) -> OperatorType {
-            OperatorType::IoOperator
+            OperatorType::IO
         }
     }
 

rust/worker/src/execution/operator.rs

@@ -7,7 +7,7 @@ use thiserror::Error;
 use uuid::Uuid;
 
 pub(crate) enum OperatorType {
-    IoOperator,
+    IO,
     Other,
 }
 

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

@@ -1,9 +1,7 @@
-use crate::execution::operator::Operator;
-use crate::log::log::Log;
-use crate::log::log::PullLogsError;
+use crate::execution::operator::{Operator, OperatorType};
+use crate::log::log::{Log, PullLogsError};
 use async_trait::async_trait;
-use chroma_types::Chunk;
-use chroma_types::LogRecord;
+use chroma_types::{Chunk, LogRecord};
 use uuid::Uuid;
 
 /// The pull logs operator is responsible for reading logs from the log service.
@@ -92,6 +90,10 @@ impl Operator<PullLogsInput, PullLogsOutput> for PullLogsOperator {
         "PullLogsOperator"
     }
 
+    fn get_type(&self) -> OperatorType {
+        OperatorType::IO
+    }
+
     async fn run(&self, input: &PullLogsInput) -> Result<PullLogsOutput, PullLogsError> {
         // We expect the log to be cheaply cloneable, we need to clone it since we need
         // a mutable reference to it. Not necessarily the best, but it works for our needs.

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

@@ -110,6 +110,6 @@ impl Operator<RecordSegmentPrefetchIoInput, RecordSegmentPrefetchIoOutput>
     }
 
     fn get_type(&self) -> OperatorType {
-        OperatorType::IoOperator
+        OperatorType::IO
     }
 }