[ENH]: sysdb changes to support moving collection hard deletes to garbage collector

Author: codetheweb

go/cmd/coordinator/cmd.go

@@ -44,13 +44,6 @@ func init() {
 	Cmd.Flags().IntVar(&conf.DBConfig.MaxOpenConns, "max-open-conns", 10, "MetaTable max open connections")
 	Cmd.Flags().StringVar(&conf.DBConfig.SslMode, "ssl-mode", "disable", "SSL mode for database connection")
 
-	// Soft deletes
-	Cmd.Flags().BoolVar(&conf.SoftDeleteEnabled, "soft-delete-enabled", true, "Enable soft deletes")
-	Cmd.Flags().DurationVar(&conf.SoftDeleteCleanupInterval, "soft-delete-cleanup-interval", 30*time.Second, "Soft delete cleanup interval")
-	Cmd.Flags().DurationVar(&conf.SoftDeleteMaxAge, "soft-delete-max-age", 72*time.Hour, "Soft delete max age")
-	Cmd.Flags().UintVar(&conf.SoftDeleteCleanupBatchSize, "soft-delete-cleanup-batch-size", 100, "Soft delete cleanup batch size")
-	// With above values, the soft delete cleaner can remove around 1000 collections in 5 minutes.
-
 	// Memberlist
 	Cmd.Flags().StringVar(&conf.KubernetesNamespace, "kubernetes-namespace", "chroma", "Kubernetes namespace")
 	Cmd.Flags().DurationVar(&conf.ReconcileInterval, "reconcile-interval", 100*time.Millisecond, "Reconcile interval")

go/pkg/common/errors.go

@@ -30,6 +30,7 @@ var (
 	ErrCollectionTooManyFork                 = errors.New("collection entry has too many forks")
 	ErrCollectionDeletedWithLocksHeld        = errors.New("collection got deleted concurrently even though select for update locks were held. Not possible unless corruption somehow")
 	ErrMissingLineageFileName                = errors.New("missing lineage file name in root collection entry")
+	ErrCollectionWasNotSoftDeleted           = errors.New("collection was not soft deleted")
 
 	// Collection metadata errors
 	ErrUnknownCollectionMetadataType = errors.New("collection metadata value type not supported")

go/pkg/sysdb/coordinator/coordinator.go

@@ -15,30 +15,18 @@ import (
 	"go.uber.org/zap"
 )
 
-// DeleteMode represents whether to perform a soft or hard delete
-type DeleteMode int
-
-const (
-	// SoftDelete marks records as deleted but keeps them in the database
-	SoftDelete DeleteMode = iota
-	// HardDelete permanently removes records from the database
-	HardDelete
-)
-
 // Coordinator is the top level component.
 // Currently, it only has the system catalog related APIs and will be extended to
 // support other functionalities such as membership managed and propagation.
 type Coordinator struct {
 	ctx         context.Context
 	catalog     Catalog
-	deleteMode  DeleteMode
 	objectStore *s3metastore.S3MetaStore
 }
 
-func NewCoordinator(ctx context.Context, deleteMode DeleteMode, objectStore *s3metastore.S3MetaStore, versionFileEnabled bool) (*Coordinator, error) {
+func NewCoordinator(ctx context.Context, objectStore *s3metastore.S3MetaStore, versionFileEnabled bool) (*Coordinator, error) {
 	s := &Coordinator{
 		ctx:         ctx,
-		deleteMode:  deleteMode,
 		objectStore: objectStore,
 	}
 
@@ -142,14 +130,11 @@ func (s *Coordinator) GetSoftDeletedCollections(ctx context.Context, collectionI
 	return s.catalog.GetSoftDeletedCollections(ctx, collectionID, tenantID, databaseName, limit)
 }
 
-func (s *Coordinator) DeleteCollection(ctx context.Context, deleteCollection *model.DeleteCollection) error {
-	if s.deleteMode == SoftDelete {
-		return s.catalog.DeleteCollection(ctx, deleteCollection, true)
-	}
-	return s.catalog.DeleteCollection(ctx, deleteCollection, false)
+func (s *Coordinator) SoftDeleteCollection(ctx context.Context, deleteCollection *model.DeleteCollection) error {
+	return s.catalog.DeleteCollection(ctx, deleteCollection, true)
 }
 
-func (s *Coordinator) CleanupSoftDeletedCollection(ctx context.Context, deleteCollection *model.DeleteCollection) error {
+func (s *Coordinator) FinishCollectionDeletion(ctx context.Context, deleteCollection *model.DeleteCollection) error {
 	return s.catalog.DeleteCollection(ctx, deleteCollection, false)
 }
 
@@ -267,7 +252,6 @@ func (s *Coordinator) BatchGetCollectionVersionFilePaths(ctx context.Context, re
 	return s.catalog.BatchGetCollectionVersionFilePaths(ctx, req.CollectionIds)
 }
 
-// SetDeleteMode sets the delete mode for testing
-func (c *Coordinator) SetDeleteMode(mode DeleteMode) {
-	c.deleteMode = mode
+func (s *Coordinator) BatchGetCollectionSoftDeleteStatus(ctx context.Context, req *coordinatorpb.BatchGetCollectionSoftDeleteStatusRequest) (*coordinatorpb.BatchGetCollectionSoftDeleteStatusResponse, error) {
+	return s.catalog.BatchGetCollectionSoftDeleteStatus(ctx, req.CollectionIds)
 }

go/pkg/sysdb/coordinator/table_catalog.go

@@ -555,7 +555,7 @@ func (tc *Catalog) hardDeleteCollection(ctx context.Context, deleteCollection *m
 	return tc.txImpl.Transaction(ctx, func(txCtx context.Context) error {
 		collectionID := deleteCollection.ID
 
-		collectionEntry, err := tc.metaDomain.CollectionDb(txCtx).GetCollectionWithoutMetadata(types.FromUniqueID(collectionID), &deleteCollection.DatabaseName, nil)
+		collectionEntry, err := tc.metaDomain.CollectionDb(txCtx).GetCollectionWithoutMetadata(types.FromUniqueID(collectionID), &deleteCollection.DatabaseName, nil) // todo: should filter for soft deleted
 		if err != nil {
 			return err
 		}
@@ -564,6 +564,51 @@ func (tc *Catalog) hardDeleteCollection(ctx context.Context, deleteCollection *m
 			return common.ErrCollectionDeleteNonExistingCollection
 		}
 
+		if !collectionEntry.IsDeleted {
+			return common.ErrCollectionWasNotSoftDeleted
+		}
+
+		if collectionEntry.RootCollectionId != nil {
+			rootCollection, err := tc.metaDomain.CollectionDb(txCtx).GetCollectionWithoutMetadata(collectionEntry.RootCollectionId, nil, nil)
+			if err != nil {
+				return err
+			}
+			if rootCollection == nil {
+				return errors.New("root collection not found")
+			}
+
+			if rootCollection.LineageFileName == nil {
+				return errors.New("lineage file name is nil on root collection")
+			}
+
+			lineageFile, err := tc.getLineageFile(txCtx, rootCollection.LineageFileName)
+			if err != nil {
+				return err
+			}
+			// filter out this collection
+			updatedDependencies := make([]*coordinatorpb.CollectionVersionDependency, 0)
+			for _, dependency := range lineageFile.Dependencies {
+				if dependency.TargetCollectionId != deleteCollection.ID.String() {
+					updatedDependencies = append(updatedDependencies, dependency)
+				}
+			}
+			lineageFile.Dependencies = updatedDependencies
+
+			newLineageFileId, err := uuid.NewV7()
+			if err != nil {
+				return err
+			}
+
+			log.Info("new lineage file id", zap.String("newLineageFileId", newLineageFileId.String()))
+
+			newLineageFileFullName, err := tc.s3Store.PutLineageFile(collectionEntry.Tenant, collectionEntry.DatabaseID, rootCollection.ID, fmt.Sprintf("%s.binpb", newLineageFileId.String()), lineageFile)
+			if err != nil {
+				return err
+			}
+
+			tc.metaDomain.CollectionDb(txCtx).UpdateCollectionLineageFilePath(rootCollection.ID, rootCollection.LineageFileName, newLineageFileFullName)
+		}
+
 		// Delete collection and collection metadata.
 		collectionDeletedCount, err := tc.metaDomain.CollectionDb(txCtx).DeleteCollectionByID(collectionID.String())
 		if err != nil {
@@ -1960,7 +2005,7 @@ func (tc *Catalog) MarkVersionForDeletion(ctx context.Context, req *coordinatorp
 
 func (tc *Catalog) updateProtoRemoveVersionEntries(versionFilePb *coordinatorpb.CollectionVersionFile, versions []int64) error {
 	// Check if version history exists
-	if versionFilePb.GetVersionHistory() == nil || len(versionFilePb.GetVersionHistory().Versions) == 0 {
+	if versionFilePb.GetVersionHistory() == nil {
 		log.Error("version history not found")
 		return errors.New("version history not found")
 	}
@@ -1996,14 +2041,14 @@ func (tc *Catalog) getNumberOfActiveVersions(versionFilePb *coordinatorpb.Collec
 	return len(activeVersions)
 }
 
-func (tc *Catalog) getOldestVersionTs(versionFilePb *coordinatorpb.CollectionVersionFile) time.Time {
+func (tc *Catalog) getOldestVersionTs(versionFilePb *coordinatorpb.CollectionVersionFile) *time.Time {
 	if versionFilePb.GetVersionHistory() == nil || len(versionFilePb.GetVersionHistory().Versions) == 0 {
-		// Returning a zero timestamp that represents an unset value.
-		return time.Time{}
+		return nil
 	}
 	oldestVersionTs := versionFilePb.GetVersionHistory().Versions[0].CreatedAtSecs
 
-	return time.Unix(oldestVersionTs, 0)
+	ts := time.Unix(oldestVersionTs, 0)
+	return &ts
 }
 
 func (tc *Catalog) DeleteVersionEntriesForCollection(ctx context.Context, tenantID string, collectionID string, versions []int64) error {
@@ -2017,8 +2062,8 @@ func (tc *Catalog) DeleteVersionEntriesForCollection(ctx context.Context, tenant
 
 		// Read the existing version file
 		collectionIDPtr := &collectionID
-		isDeleted := false
-		collectionEntry, err := tc.metaDomain.CollectionDb(ctx).GetCollectionWithoutMetadata(collectionIDPtr, nil, &isDeleted)
+		// isDeleted := true // todo
+		collectionEntry, err := tc.metaDomain.CollectionDb(ctx).GetCollectionWithoutMetadata(collectionIDPtr, nil, nil) // todo
 		if err != nil {
 			return err
 		}
@@ -2038,14 +2083,19 @@ func (tc *Catalog) DeleteVersionEntriesForCollection(ctx context.Context, tenant
 		}
 
 		numActiveVersions := tc.getNumberOfActiveVersions(versionFilePb)
-		if numActiveVersions < 1 {
+		if numActiveVersions < 1 && !collectionEntry.IsDeleted {
 			// No remaining valid versions after GC.
 			return errors.New("no valid versions after gc")
 		}
 
 		// Get the creation time of the oldest version.
 		oldestVersionTs := tc.getOldestVersionTs(versionFilePb)
-		if oldestVersionTs.IsZero() {
+		if oldestVersionTs == nil {
+			if !collectionEntry.IsDeleted {
+				// todo
+				return errors.New("oldest version timestamp is nil after GC")
+			}
+		} else if oldestVersionTs.IsZero() {
 			// This should never happen.
 			log.Error("oldest version timestamp is zero after GC.", zap.String("collection_id", collectionID))
 			// No versions to delete.
@@ -2065,7 +2115,7 @@ func (tc *Catalog) DeleteVersionEntriesForCollection(ctx context.Context, tenant
 		}
 
 		// Update the version file name in Postgres table as a CAS operation
-		rowsAffected, err := tc.metaDomain.CollectionDb(ctx).UpdateVersionRelatedFields(collectionID, existingVersionFileName, newVerFileFullPath, &oldestVersionTs, &numActiveVersions)
+		rowsAffected, err := tc.metaDomain.CollectionDb(ctx).UpdateVersionRelatedFields(collectionID, existingVersionFileName, newVerFileFullPath, oldestVersionTs, &numActiveVersions)
 		if err != nil {
 			// Delete the newly created version file from S3 since it is not needed
 			tc.s3Store.DeleteVersionFile(tenantID, collectionEntry.DatabaseID, collectionID, newVersionFileName)
@@ -2113,6 +2163,20 @@ func (tc *Catalog) BatchGetCollectionVersionFilePaths(ctx context.Context, colle
 	return &result, nil
 }
 
+func (tc *Catalog) BatchGetCollectionSoftDeleteStatus(ctx context.Context, collectionIds []string) (*coordinatorpb.BatchGetCollectionSoftDeleteStatusResponse, error) {
+	result := coordinatorpb.BatchGetCollectionSoftDeleteStatusResponse{
+		CollectionIdToIsSoftDeleted: make(map[string]bool),
+	}
+
+	status, err := tc.metaDomain.CollectionDb(ctx).BatchGetCollectionSoftDeleteStatus(collectionIds)
+	if err != nil {
+		return nil, err
+	}
+	result.CollectionIdToIsSoftDeleted = status
+
+	return &result, nil
+}
+
 func (tc *Catalog) GetVersionFileNamesForCollection(ctx context.Context, tenantID string, collectionID string) (string, error) {
 	collectionIDPtr := &collectionID
 	isDeleted := false