[ENH] Eventual consistency in query nodes

idl/chromadb/proto/query_executor.proto

@@ -105,7 +105,7 @@ message RankExpr {
         optional float default = 4;
         bool return_rank = 5;
     }
-    
+
     message RankPair {
         RankExpr left = 1;
         RankExpr right = 2;
@@ -170,9 +170,22 @@ message SearchPayload {
     SelectOperator select = 4;
 }
 
+// ReadLevel specifies which data sources to read from during queries.
+// This affects consistency vs performance tradeoffs.
+enum ReadLevel {
+    // Read from both the index and the write-ahead log (default).
+    // Provides full consistency with all committed writes visible.
+    INDEX_AND_WAL = 0;
+    // Read only from the index, skipping the write-ahead log.
+    // Provides eventual consistency - recent uncommitted writes may not be visible.
+    INDEX_ONLY = 1;
+}
+
 message SearchPlan {
     ScanOperator scan = 1;
     repeated SearchPayload payloads = 2;
+    // Specifies the read level for this query
+    ReadLevel read_level = 3;
 }
 
 message SearchRecord {
@@ -198,4 +211,3 @@ service QueryExecutor {
     rpc KNN(KNNPlan) returns (KNNBatchResult) {}
     rpc Search(SearchPlan) returns (SearchResult) {}
 }
-

rust/frontend/src/impls/service_based_frontend.rs

@@ -20,7 +20,7 @@ use chroma_sysdb::{GetCollectionsOptions, SysDb};
 use chroma_system::System;
 use chroma_types::{
     operator::{Filter, KnnBatch, KnnProjection, Limit, Projection, Scan},
-    plan::{Count, Get, Knn, Search},
+    plan::{Count, Get, Knn, ReadLevel, Search},
     AddCollectionRecordsError, AddCollectionRecordsRequest, AddCollectionRecordsResponse,
     AttachFunctionRequest, AttachFunctionResponse, Cmek, Collection, CollectionUuid,
     CountCollectionsError, CountCollectionsRequest, CountCollectionsResponse, CountRequest,
@@ -1834,6 +1834,7 @@ impl ServiceBasedFrontend {
                 collection_and_segments,
             },
             payloads: request.searches,
+            read_level: ReadLevel::IndexAndWal,
         };
 
         // Execute the single search plan using the executor

rust/types/src/execution/plan.rs

@@ -543,16 +543,47 @@ impl TryFrom<SearchPayload> for chroma_proto::SearchPayload {
     }
 }
 
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub enum ReadLevel {
+    /// Read from both the index and the write-ahead log (default).
+    /// Provides full consistency with all committed writes visible.
+    #[default]
+    IndexAndWal,
+    /// Read only from the index, skipping the write-ahead log.
+    /// Provides eventual consistency - recent uncommitted writes may not be visible.
+    IndexOnly,
+}
+
+impl From<chroma_proto::ReadLevel> for ReadLevel {
+    fn from(value: chroma_proto::ReadLevel) -> Self {
+        match value {
+            chroma_proto::ReadLevel::IndexAndWal => ReadLevel::IndexAndWal,
+            chroma_proto::ReadLevel::IndexOnly => ReadLevel::IndexOnly,
+        }
+    }
+}
+
+impl From<ReadLevel> for chroma_proto::ReadLevel {
+    fn from(value: ReadLevel) -> Self {
+        match value {
+            ReadLevel::IndexAndWal => chroma_proto::ReadLevel::IndexAndWal,
+            ReadLevel::IndexOnly => chroma_proto::ReadLevel::IndexOnly,
+        }
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct Search {
     pub scan: Scan,
     pub payloads: Vec<SearchPayload>,
+    pub read_level: ReadLevel,
 }
 
 impl TryFrom<chroma_proto::SearchPlan> for Search {
     type Error = QueryConversionError;
 
     fn try_from(value: chroma_proto::SearchPlan) -> Result<Self, Self::Error> {
+        let read_level = value.read_level().into();
         Ok(Self {
             scan: value
                 .scan
@@ -563,6 +594,7 @@ impl TryFrom<chroma_proto::SearchPlan> for Search {
                 .into_iter()
                 .map(TryInto::try_into)
                 .collect::<Result<Vec<_>, _>>()?,
+            read_level,
         })
     }
 }
@@ -578,6 +610,7 @@ impl TryFrom<Search> for chroma_proto::SearchPlan {
                 .into_iter()
                 .map(TryInto::try_into)
                 .collect::<Result<Vec<_>, _>>()?,
+            read_level: chroma_proto::ReadLevel::from(value.read_level).into(),
         })
     }
 }

rust/worker/benches/query.rs

@@ -8,7 +8,7 @@ use chroma_benchmark::{
 use chroma_config::{registry::Registry, Configurable};
 use chroma_segment::test::TestDistributedSegment;
 use chroma_system::{ComponentHandle, Dispatcher, Orchestrator, System};
-use chroma_types::operator::Knn;
+use chroma_types::{operator::Knn, plan::ReadLevel};
 use criterion::{criterion_group, criterion_main, Criterion};
 use futures::{stream, StreamExt, TryStreamExt};
 use load::{
@@ -39,6 +39,7 @@ fn trivial_knn_filter(
         test_segments.into(),
         empty_fetch_log(collection_uuid),
         trivial_filter(),
+        ReadLevel::IndexAndWal,
     )
 }
 
@@ -57,6 +58,7 @@ fn always_true_knn_filter(
         test_segments.into(),
         empty_fetch_log(collection_uuid),
         always_true_filter_for_modulo_metadata(),
+        ReadLevel::IndexAndWal,
     )
 }
 
@@ -75,6 +77,7 @@ fn always_false_knn_filter(
         test_segments.into(),
         empty_fetch_log(collection_uuid),
         always_false_filter_for_modulo_metadata(),
+        ReadLevel::IndexAndWal,
     )
 }
 

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

@@ -12,8 +12,8 @@ use chroma_system::{
     OrchestratorContext, PanicError, TaskError, TaskMessage, TaskResult,
 };
 use chroma_types::{
-    operator::Filter, CollectionAndSegments, HnswParametersFromSegmentError, SchemaError,
-    SegmentType,
+    operator::Filter, plan::ReadLevel, CollectionAndSegments, HnswParametersFromSegmentError,
+    SchemaError, SegmentType,
 };
 use opentelemetry::trace::TraceContextExt;
 use thiserror::Error;
@@ -178,6 +178,9 @@ pub struct KnnFilterOrchestrator {
     // Fetched logs
     fetched_logs: Option<FetchLogOutput>,
 
+    // Read level for consistency vs performance tradeoff
+    read_level: ReadLevel,
+
     // Pipelined operators
     filter: Filter,
 
@@ -186,6 +189,7 @@ pub struct KnnFilterOrchestrator {
 }
 
 impl KnnFilterOrchestrator {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         blockfile_provider: BlockfileProvider,
         dispatcher: ComponentHandle<Dispatcher>,
@@ -194,6 +198,7 @@ impl KnnFilterOrchestrator {
         collection_and_segments: CollectionAndSegments,
         fetch_log: FetchLogOperator,
         filter: Filter,
+        read_level: ReadLevel,
     ) -> Self {
         let context = OrchestratorContext::new(dispatcher);
         Self {
@@ -204,10 +209,29 @@ impl KnnFilterOrchestrator {
             collection_and_segments,
             fetch_log,
             fetched_logs: None,
+            read_level,
             filter,
             result_channel: None,
         }
     }
+
+    fn create_filter_task(
+        &self,
+        logs: FetchLogOutput,
+        ctx: &ComponentContext<Self>,
+    ) -> TaskMessage {
+        wrap(
+            Box::new(self.filter.clone()),
+            FilterInput {
+                logs,
+                blockfile_provider: self.blockfile_provider.clone(),
+                metadata_segment: self.collection_and_segments.metadata_segment.clone(),
+                record_segment: self.collection_and_segments.record_segment.clone(),
+            },
+            ctx.receiver(),
+            self.context.task_cancellation_token.clone(),
+        )
+    }
 }
 
 #[async_trait]
@@ -272,14 +296,28 @@ impl Orchestrator for KnnFilterOrchestrator {
         Span::current().add_link(prefetch_span.context().span().span_context().clone());
         tasks.push((prefetch_metadata_task, Some(prefetch_span)));
 
-        // Fetch log task.
-        let fetch_log_task = wrap(
-            Box::new(self.fetch_log.clone()),
-            (),
-            ctx.receiver(),
-            self.context.task_cancellation_token.clone(),
-        );
-        tasks.push((fetch_log_task, Some(Span::current())));
+        match self.read_level {
+            ReadLevel::IndexOnly => {
+                // For IndexOnly queries, skip log fetching and use empty logs
+                tracing::info!("Skipping log fetch for IndexOnly read level");
+                let empty_logs = FetchLogOutput::new(Vec::new().into());
+                self.fetched_logs = Some(empty_logs.clone());
+
+                // Immediately schedule the filter task with empty logs
+                let filter_task = self.create_filter_task(empty_logs, ctx);
+                tasks.push((filter_task, Some(Span::current())));
+            }
+            ReadLevel::IndexAndWal => {
+                // Fetch log task for full consistency.
+                let fetch_log_task = wrap(
+                    Box::new(self.fetch_log.clone()),
+                    (),
+                    ctx.receiver(),
+                    self.context.task_cancellation_token.clone(),
+                );
+                tasks.push((fetch_log_task, Some(Span::current())));
+            }
+        }
 
         tasks
     }
@@ -326,17 +364,7 @@ impl Handler<TaskResult<FetchLogOutput, FetchLogError>> for KnnFilterOrchestrato
 
         self.fetched_logs = Some(output.clone());
 
-        let task = wrap(
-            Box::new(self.filter.clone()),
-            FilterInput {
-                logs: output,
-                blockfile_provider: self.blockfile_provider.clone(),
-                metadata_segment: self.collection_and_segments.metadata_segment.clone(),
-                record_segment: self.collection_and_segments.record_segment.clone(),
-            },
-            ctx.receiver(),
-            self.context.task_cancellation_token.clone(),
-        );
+        let task = self.create_filter_task(output, ctx);
         self.send(task, ctx, Some(Span::current())).await;
     }
 }

rust/worker/src/server.rs

@@ -17,7 +17,7 @@ use chroma_types::{
         query_executor_server::{QueryExecutor, QueryExecutorServer},
     },
     operator::{GetResult, Knn, KnnBatch, KnnBatchResult, KnnProjection, QueryVector, Scan},
-    plan::SearchPayload,
+    plan::{ReadLevel, SearchPayload},
     CollectionAndSegments, SegmentType,
 };
 use futures::{stream, StreamExt, TryStreamExt};
@@ -342,6 +342,7 @@ impl WorkerServer {
             collection_and_segments.clone(),
             fetch_log,
             filter.try_into()?,
+            ReadLevel::IndexAndWal, // Full consistency for KNN queries
         );
 
         let matching_records = match knn_filter_orchestrator.run(system.clone()).await {
@@ -478,6 +479,7 @@ impl WorkerServer {
         &self,
         scan: chroma_proto::ScanOperator,
         payload: chroma_proto::SearchPayload,
+        read_level: ReadLevel,
     ) -> Result<RankOrchestratorOutput, Status> {
         let collection_and_segments = Scan::try_from(scan)?.collection_and_segments;
         let search_payload = SearchPayload::try_from(payload)?;
@@ -497,6 +499,7 @@ impl WorkerServer {
             collection_and_segments.clone(),
             fetch_log,
             search_payload.filter.clone(),
+            read_level, // Use the specified read level
         );
 
         let knn_filter_output = match knn_filter_orchestrator.run(self.system.clone()).await {
@@ -618,14 +621,15 @@ impl WorkerServer {
         search: Request<chroma_proto::SearchPlan>,
     ) -> Result<Response<chroma_proto::SearchResult>, Status> {
         let search_plan = search.into_inner();
+        let read_level: ReadLevel = search_plan.read_level().into();
         let scan = search_plan
             .scan
             .ok_or(Status::invalid_argument("Invalid Scan Operator"))?;
 
         let futures = search_plan
             .payloads
             .into_iter()
-            .map(|payload| self.orchestrate_search(scan.clone(), payload));
+            .map(|payload| self.orchestrate_search(scan.clone(), payload, read_level));
 
         let orchestrator_results = stream::iter(futures)
             .buffered(32) // Process up to 32 payloads concurrently