[ENH] Limit maximum number of collection for single GC

rust/garbage_collector/src/garbage_collector_component.rs

@@ -196,7 +196,10 @@ impl Handler<GarbageCollectMessage> for GarbageCollector {
         tracing::info!("Getting collections to gc");
         let collections_to_gc = self
             .sysdb_client
-            .get_collections_to_gc()
+            .get_collections_to_gc(
+                Some(self.relative_cutoff_time.as_secs()),
+                Some(self.max_collections_to_gc.into()),
+            )
             .await
             .expect("Failed to get collections to gc");
         tracing::info!("Got {} collections to gc", collections_to_gc.len());

rust/garbage_collector/src/garbage_collector_orchestrator.rs

@@ -850,7 +850,7 @@ mod tests {
             .unwrap();
 
         // Get collection info for GC from sysdb
-        let collections_to_gc = sysdb.get_collections_to_gc().await.unwrap();
+        let collections_to_gc = sysdb.get_collections_to_gc(None, None).await.unwrap();
         let collection_info = collections_to_gc
             .iter()
             .find(|c| c.id == collection_id)
@@ -994,7 +994,7 @@ mod tests {
             .unwrap();
 
         // Get collection info for GC from sysdb
-        let collections_to_gc = sysdb.get_collections_to_gc().await.unwrap();
+        let collections_to_gc = sysdb.get_collections_to_gc(None, None).await.unwrap();
         let collection_info = collections_to_gc
             .iter()
             .find(|c| c.id == collection_id)
@@ -1152,7 +1152,7 @@ mod tests {
             .unwrap();
 
         // Get collection info for GC from sysdb
-        let collections_to_gc = sysdb.get_collections_to_gc().await.unwrap();
+        let collections_to_gc = sysdb.get_collections_to_gc(None, None).await.unwrap();
         let collection_info = collections_to_gc
             .iter()
             .find(|c| c.id == collection_id)

rust/sysdb/src/sysdb.rs

@@ -317,10 +317,12 @@ impl SysDb {
 
     pub async fn get_collections_to_gc(
         &mut self,
+        cutoff_time_secs: Option<u64>,
+        limit: Option<u64>,
     ) -> Result<Vec<CollectionToGcInfo>, GetCollectionsToGcError> {
         match self {
-            SysDb::Grpc(grpc) => grpc.get_collections_to_gc().await,
-            SysDb::Sqlite(_) => todo!(),
+            SysDb::Grpc(grpc) => grpc.get_collections_to_gc(cutoff_time_secs, limit).await,
+            SysDb::Sqlite(_) => unimplemented!("Garbage collection does not work for local chroma"),
             SysDb::Test(_) => todo!(),
         }
     }
@@ -892,12 +894,14 @@ impl GrpcSysDb {
 
     pub async fn get_collections_to_gc(
         &mut self,
+        cutoff_time_secs: Option<u64>,
+        limit: Option<u64>,
     ) -> Result<Vec<CollectionToGcInfo>, GetCollectionsToGcError> {
         let res = self
             .client
             .list_collections_to_gc(chroma_proto::ListCollectionsToGcRequest {
-                cutoff_time_secs: None,
-                limit: None,
+                cutoff_time_secs,
+                limit,
             })
             .await;
 

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

@@ -268,7 +268,7 @@ impl CompactOrchestrator {
     ) {
         self.state = ExecutionState::MaterializeApplyCommitFlush;
 
-        // NOTE: We allow writers no be uninitialized for the case when the materialized logs are empty
+        // NOTE: We allow writers to be uninitialized for the case when the materialized logs are empty
         let record_reader = self
             .get_segment_writers()
             .ok()