[ENH]: allow deleting v0 from version file & correctly propagate errors on DeleteCollectionVersion

go/pkg/sysdb/coordinator/table_catalog.go

@@ -1138,6 +1138,7 @@ func (tc *Catalog) createSegmentImpl(txCtx context.Context, createSegment *model
 		Type:         createSegment.Type,
 		Scope:        createSegment.Scope,
 		Ts:           ts,
+		FilePaths:    createSegment.FilePaths,
 	}
 	err := tc.metaDomain.SegmentDb(txCtx).Insert(dbSegment)
 	if err != nil {
@@ -2000,11 +2001,11 @@ func (tc *Catalog) getNumberOfActiveVersions(versionFilePb *coordinatorpb.Collec
 }
 
 func (tc *Catalog) getOldestVersionTs(versionFilePb *coordinatorpb.CollectionVersionFile) time.Time {
-	if versionFilePb.GetVersionHistory() == nil || len(versionFilePb.GetVersionHistory().Versions) <= 1 {
+	if versionFilePb.GetVersionHistory() == nil || len(versionFilePb.GetVersionHistory().Versions) == 0 {
 		// Returning a zero timestamp that represents an unset value.
 		return time.Time{}
 	}
-	oldestVersionTs := versionFilePb.GetVersionHistory().Versions[1].CreatedAtSecs
+	oldestVersionTs := versionFilePb.GetVersionHistory().Versions[0].CreatedAtSecs
 
 	return time.Unix(oldestVersionTs, 0)
 }
@@ -2041,7 +2042,7 @@ func (tc *Catalog) DeleteVersionEntriesForCollection(ctx context.Context, tenant
 		}
 
 		numActiveVersions := tc.getNumberOfActiveVersions(versionFilePb)
-		if numActiveVersions <= 1 {
+		if numActiveVersions < 1 {
 			// No remaining valid versions after GC.
 			return errors.New("no valid versions after gc")
 		}
@@ -2091,11 +2092,15 @@ func (tc *Catalog) DeleteCollectionVersion(ctx context.Context, req *coordinator
 	result := coordinatorpb.DeleteCollectionVersionResponse{
 		CollectionIdToSuccess: make(map[string]bool),
 	}
+	var firstErr error
 	for _, collectionVersionList := range req.Versions {
 		err := tc.DeleteVersionEntriesForCollection(ctx, collectionVersionList.TenantId, collectionVersionList.CollectionId, collectionVersionList.Versions)
 		result.CollectionIdToSuccess[collectionVersionList.CollectionId] = err == nil
+		if firstErr == nil && err != nil {
+			firstErr = err
+		}
 	}
-	return &result, nil
+	return &result, firstErr
 }
 
 func (tc *Catalog) BatchGetCollectionVersionFilePaths(ctx context.Context, collectionIds []string) (*coordinatorpb.BatchGetCollectionVersionFilePathsResponse, error) {

go/pkg/sysdb/coordinator/table_catalog_test.go

@@ -759,7 +759,7 @@ func TestCatalog_DeleteCollectionVersion_CollectionNotFound(t *testing.T) {
 	resp, err := catalog.DeleteCollectionVersion(context.Background(), req)
 
 	// Verify results
-	assert.NoError(t, err)
+	assert.Error(t, err)
 	assert.NotNil(t, resp)
 	assert.False(t, resp.CollectionIdToSuccess[collectionID])
 

go/pkg/sysdb/grpc/proto_model_convert.go

@@ -1,6 +1,8 @@
 package grpc
 
 import (
+	"time"
+
 	"github.com/chroma-core/chroma/go/pkg/common"
 	"github.com/chroma-core/chroma/go/pkg/proto/coordinatorpb"
 	"github.com/chroma-core/chroma/go/pkg/sysdb/coordinator/model"
@@ -131,6 +133,7 @@ func convertToCreateCollectionModel(req *coordinatorpb.CreateCollectionRequest)
 		GetOrCreate:          req.GetGetOrCreate(),
 		TenantID:             req.GetTenant(),
 		DatabaseName:         req.GetDatabase(),
+		Ts:                   time.Now().Unix(),
 	}, nil
 }
 

rust/garbage_collector/src/operators/delete_versions_at_sysdb.rs

@@ -38,6 +38,8 @@ pub struct DeleteVersionsAtSysDbOutput {
 
 #[derive(Error, Debug)]
 pub enum DeleteVersionsAtSysDbError {
+    #[error("Unknown error occurred when deleting versions at sysdb")]
+    UnknownError,
     #[error("Error deleting versions in sysdb: {0}")]
     SysDBError(String),
     #[error("Error deleting version file {path}: {message}")]
@@ -157,7 +159,13 @@ impl Operator<DeleteVersionsAtSysDbInput, DeleteVersionsAtSysDbOutput>
                 .delete_collection_version(vec![input.versions_to_delete.clone()])
                 .await
             {
-                Ok(_) => {
+                Ok(results) => {
+                    for (_, was_successful) in results {
+                        if !was_successful {
+                            return Err(DeleteVersionsAtSysDbError::UnknownError);
+                        }
+                    }
+
                     tracing::info!(
                         versions = ?input.versions_to_delete.versions,
                         "Successfully deleted versions from SysDB"

rust/sysdb/src/sysdb.rs

@@ -1331,14 +1331,14 @@ impl ChromaError for MarkVersionForDeletionError {
 
 #[derive(Error, Debug)]
 pub enum DeleteCollectionVersionError {
-    #[error("Failed to delete version")]
+    #[error("Failed to delete version: {0}")]
     FailedToDeleteVersion(#[from] tonic::Status),
 }
 
 impl ChromaError for DeleteCollectionVersionError {
     fn code(&self) -> ErrorCodes {
         match self {
-            DeleteCollectionVersionError::FailedToDeleteVersion(_) => ErrorCodes::Internal,
+            DeleteCollectionVersionError::FailedToDeleteVersion(e) => e.code().into(),
         }
     }
 }