[ENH] Delete version files in GC. (#4068)

## Description of changes

*Summarize the changes made by this PR.*
 - Improvements & Bug fixes
- Delete version files as part of GC. SysDb creates new version file at
every compaction since version files are immutable. Clean up these extra
version files.
 - New functionality
   - ...

## Test plan
*How are these changes tested?*

- [ ] Tests pass locally with `pytest` for python, `yarn test` for js,
`cargo test` for rust

## Documentation Changes
*Are all docstrings for user-facing APIs updated if required? Do we need
to make documentation changes in the [docs
repository](https://github.com/chroma-core/docs)?*

Author: rohitcpbot

rust/garbage_collector/src/garbage_collector_orchestrator.rs

@@ -471,7 +471,9 @@ impl Handler<TaskResult<DeleteUnusedFilesOutput, DeleteUnusedFilesError>>
             .expect("Epoch ID should be set");
 
         let delete_versions_task = wrap(
-            Box::new(DeleteVersionsAtSysDbOperator {}),
+            Box::new(DeleteVersionsAtSysDbOperator {
+                storage: self.storage.clone(),
+            }),
             DeleteVersionsAtSysDbInput {
                 version_file,
                 epoch_id,

rust/garbage_collector/src/operators/delete_versions_at_sysdb.rs

@@ -1,13 +1,24 @@
 use async_trait::async_trait;
 use chroma_error::{ChromaError, ErrorCodes};
+use chroma_storage::Storage;
 use chroma_sysdb::SysDb;
 use chroma_system::{Operator, OperatorType};
 use chroma_types::chroma_proto::{CollectionVersionFile, VersionListForCollection};
+use futures::stream::StreamExt;
 use std::collections::HashSet;
 use thiserror::Error;
 
-#[derive(Clone, Debug)]
-pub struct DeleteVersionsAtSysDbOperator {}
+#[derive(Clone)]
+pub struct DeleteVersionsAtSysDbOperator {
+    pub storage: Storage,
+}
+
+impl std::fmt::Debug for DeleteVersionsAtSysDbOperator {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DeleteVersionsAtSysDbOperator")
+            .finish_non_exhaustive()
+    }
+}
 
 #[derive(Debug)]
 pub struct DeleteVersionsAtSysDbInput {
@@ -29,6 +40,8 @@ pub struct DeleteVersionsAtSysDbOutput {
 pub enum DeleteVersionsAtSysDbError {
     #[error("Error deleting versions in sysdb: {0}")]
     SysDBError(String),
+    #[error("Error deleting version file {path}: {message}")]
+    DeleteFileError { path: String, message: String },
 }
 
 impl ChromaError for DeleteVersionsAtSysDbError {
@@ -37,6 +50,72 @@ impl ChromaError for DeleteVersionsAtSysDbError {
     }
 }
 
+impl DeleteVersionsAtSysDbOperator {
+    async fn delete_version_files(
+        &self,
+        version_file: &CollectionVersionFile,
+        versions_to_delete: &[i64],
+    ) {
+        // Handle case where version_history is None
+        let version_history = match &version_file.version_history {
+            Some(history) => history,
+            None => return, // Nothing to delete if there's no version history
+        };
+
+        let version_files_to_delete: Vec<String> = version_history
+            .versions
+            .iter()
+            .filter(|v| versions_to_delete.contains(&v.version))
+            .filter_map(|v| {
+                if !v.version_file_name.is_empty() {
+                    Some(v.version_file_name.clone())
+                } else {
+                    None
+                }
+            })
+            .collect();
+
+        if version_files_to_delete.is_empty() {
+            return;
+        }
+
+        tracing::info!(
+            files = ?version_files_to_delete,
+            "Deleting version files"
+        );
+
+        let mut futures = Vec::new();
+        for file_path in &version_files_to_delete {
+            let storage = self.storage.clone();
+            let path = file_path.clone();
+            futures.push(async move {
+                storage
+                    .delete(&path)
+                    .await
+                    .map_err(|e| (path, e.to_string()))
+            });
+        }
+
+        let num_futures = futures.len();
+        let results = futures::stream::iter(futures)
+            .buffer_unordered(num_futures)
+            .collect::<Vec<_>>()
+            .await;
+
+        // Process any errors that occurred during file deletion
+        for result in results {
+            if let Err((path, error)) = result {
+                tracing::warn!(
+                    error = %error,
+                    path = %path,
+                    "Failed to delete version file {}, continuing since it could have been deleted already",
+                    path
+                );
+            }
+        }
+    }
+}
+
 #[async_trait]
 impl Operator<DeleteVersionsAtSysDbInput, DeleteVersionsAtSysDbOutput>
     for DeleteVersionsAtSysDbOperator
@@ -69,6 +148,10 @@ impl Operator<DeleteVersionsAtSysDbInput, DeleteVersionsAtSysDbOutput>
         let mut sysdb = input.sysdb_client.clone();
 
         if !input.versions_to_delete.versions.is_empty() {
+            // First, delete the version files from the storage.
+            self.delete_version_files(&input.version_file, &input.versions_to_delete.versions)
+                .await;
+
             tracing::info!(
                 versions = ?input.versions_to_delete.versions,
                 "Deleting versions from SysDB"
@@ -110,12 +193,23 @@ impl Operator<DeleteVersionsAtSysDbInput, DeleteVersionsAtSysDbOutput>
 #[cfg(test)]
 mod tests {
     use super::*;
+    use chroma_storage::local::LocalStorage;
     use chroma_sysdb::TestSysDb;
+    use chroma_types::chroma_proto;
+    use tempfile::TempDir;
 
     #[tokio::test]
     async fn test_delete_versions_success() {
+        let tmp_dir = TempDir::new().unwrap();
+        let storage = Storage::Local(LocalStorage::new(tmp_dir.path().to_str().unwrap()));
         let sysdb = SysDb::Test(TestSysDb::new());
-        let version_file = CollectionVersionFile::default();
+
+        // Create a version file with actual version history
+        let version_file = CollectionVersionFile {
+            version_history: Some(chroma_proto::CollectionVersionHistory { versions: vec![] }),
+            ..Default::default()
+        };
+
         let versions_to_delete = VersionListForCollection {
             collection_id: "test_collection".to_string(),
             database_id: "default".to_string(),
@@ -131,7 +225,9 @@ mod tests {
             unused_s3_files: HashSet::new(),
         };
 
-        let operator = DeleteVersionsAtSysDbOperator {};
+        let operator = DeleteVersionsAtSysDbOperator {
+            storage: storage.clone(),
+        };
         let result = operator.run(&input).await;
 
         assert!(result.is_ok());
@@ -142,6 +238,8 @@ mod tests {
 
     #[tokio::test]
     async fn test_delete_versions_empty_list() {
+        let tmp_dir = TempDir::new().unwrap();
+        let storage = Storage::Local(LocalStorage::new(tmp_dir.path().to_str().unwrap()));
         let sysdb = SysDb::Test(TestSysDb::new());
         let version_file = CollectionVersionFile::default();
         let versions_to_delete = VersionListForCollection {
@@ -159,12 +257,163 @@ mod tests {
             unused_s3_files: HashSet::new(),
         };
 
-        let operator = DeleteVersionsAtSysDbOperator {};
+        let operator = DeleteVersionsAtSysDbOperator {
+            storage: storage.clone(),
+        };
         let result = operator.run(&input).await;
 
         assert!(result.is_ok());
         let output = result.unwrap();
         assert_eq!(output.version_file, version_file);
         assert_eq!(output.versions_to_delete, versions_to_delete);
     }
+
+    #[tokio::test]
+    async fn test_delete_version_files() {
+        let tmp_dir = TempDir::new().unwrap();
+        let storage = Storage::Local(LocalStorage::new(tmp_dir.path().to_str().unwrap()));
+
+        // Create test files in the temporary directory
+        let test_files = vec!["version_1", "version_2"];
+        for file in &test_files {
+            std::fs::write(tmp_dir.path().join(file), "test content").unwrap();
+        }
+
+        // Create version file with history
+        let version_file = CollectionVersionFile {
+            version_history: Some(chroma_proto::CollectionVersionHistory {
+                versions: vec![
+                    chroma_proto::CollectionVersionInfo {
+                        version: 1,
+                        version_file_name: "version_1".to_string(),
+                        ..Default::default()
+                    },
+                    chroma_proto::CollectionVersionInfo {
+                        version: 2,
+                        version_file_name: "version_2".to_string(),
+                        ..Default::default()
+                    },
+                    chroma_proto::CollectionVersionInfo {
+                        version: 3,
+                        version_file_name: "".to_string(), // Empty file name to test filtering
+                        ..Default::default()
+                    },
+                ],
+            }),
+            ..Default::default()
+        };
+
+        let operator = DeleteVersionsAtSysDbOperator {
+            storage: storage.clone(),
+        };
+
+        // Test deleting specific versions
+        operator
+            .delete_version_files(&version_file, &[1, 2, 3])
+            .await;
+
+        // Verify files were deleted
+        for file in &test_files {
+            assert!(
+                !tmp_dir.path().join(file).exists(),
+                "File {} should be deleted",
+                file
+            );
+        }
+
+        // Test with non-existent files (should not panic)
+        operator
+            .delete_version_files(&version_file, &[1, 2, 3])
+            .await;
+    }
+
+    #[tokio::test]
+    async fn test_delete_version_files_no_history() {
+        let tmp_dir = TempDir::new().unwrap();
+        let storage = Storage::Local(LocalStorage::new(tmp_dir.path().to_str().unwrap()));
+
+        // Create version file without history
+        let version_file = CollectionVersionFile {
+            version_history: None,
+            ..Default::default()
+        };
+
+        let operator = DeleteVersionsAtSysDbOperator {
+            storage: storage.clone(),
+        };
+
+        // Should return early without error
+        operator
+            .delete_version_files(&version_file, &[1, 2, 3])
+            .await;
+    }
+
+    #[tokio::test]
+    async fn test_operator_deletes_version_files() {
+        let tmp_dir = TempDir::new().unwrap();
+        let storage = Storage::Local(LocalStorage::new(tmp_dir.path().to_str().unwrap()));
+        let sysdb = SysDb::Test(TestSysDb::new());
+
+        // Create test files in the temporary directory
+        let test_files = vec!["version_1", "version_2"];
+        for file in &test_files {
+            std::fs::write(tmp_dir.path().join(file), "test content").unwrap();
+        }
+
+        // Create version file with history
+        let version_file = CollectionVersionFile {
+            version_history: Some(chroma_proto::CollectionVersionHistory {
+                versions: vec![
+                    chroma_proto::CollectionVersionInfo {
+                        version: 1,
+                        version_file_name: "version_1".to_string(),
+                        ..Default::default()
+                    },
+                    chroma_proto::CollectionVersionInfo {
+                        version: 2,
+                        version_file_name: "version_2".to_string(),
+                        ..Default::default()
+                    },
+                ],
+            }),
+            ..Default::default()
+        };
+
+        let versions_to_delete = VersionListForCollection {
+            collection_id: "test_collection".to_string(),
+            database_id: "default".to_string(),
+            tenant_id: "default".to_string(),
+            versions: vec![1, 2],
+        };
+
+        let input = DeleteVersionsAtSysDbInput {
+            version_file: version_file.clone(),
+            versions_to_delete: versions_to_delete.clone(),
+            sysdb_client: sysdb,
+            epoch_id: 123,
+            unused_s3_files: HashSet::new(),
+        };
+
+        let operator = DeleteVersionsAtSysDbOperator {
+            storage: storage.clone(),
+        };
+
+        // Run the operator
+        let result = operator.run(&input).await;
+        assert!(result.is_ok());
+
+        // Verify files were deleted
+        for file in &test_files {
+            assert!(
+                !tmp_dir.path().join(file).exists(),
+                "File {} should have been deleted by the operator",
+                file
+            );
+        }
+
+        // Verify the output matches our expectations
+        let output = result.unwrap();
+        assert_eq!(output.version_file, version_file);
+        assert_eq!(output.versions_to_delete, versions_to_delete);
+    }
 }