[ENH] Return better error from Sysdb for flush compaction. (#3244)

## Description of changes

*Summarize the changes made by this PR.*
 - Improvements & Bug fixes
- Check if collection exists or is soft deleted and return appropriate
error. Will allow compaction to differentiate between different error
conditions.
 - 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)?*
n/a

go/pkg/common/errors.go

@@ -16,6 +16,7 @@ var (
 
 	// Collection errors
 	ErrCollectionNotFound                    = errors.New("collection not found")
+	ErrCollectionSoftDeleted                 = errors.New("collection soft deleted")
 	ErrCollectionIDFormat                    = errors.New("collection id format error")
 	ErrCollectionNameEmpty                   = errors.New("collection name is empty")
 	ErrCollectionUniqueConstraintViolation   = errors.New("collection unique constraint violation")

go/pkg/sysdb/coordinator/coordinator_test.go

@@ -2,6 +2,7 @@ package coordinator
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sort"
 	"strconv"
@@ -1135,6 +1136,18 @@ func (suite *APIsTestSuite) TestSoftAndHardDeleteCollection() {
 	suite.NoError(err)
 	suite.Empty(results)
 
+	// Do a flush collection compaction
+	flushCollectionInfo, err := suite.coordinator.FlushCollectionCompaction(ctx, &model.FlushCollectionCompaction{
+		ID:       testCollection.ID,
+		TenantID: testCollection.TenantID,
+	})
+	// The flush collection compaction should fail because the collection is soft deleted.
+	suite.Error(err)
+	// Check for ErrCollectionSoftDeleted error.
+	suite.True(errors.Is(err, common.ErrCollectionSoftDeleted))
+	// Check that the flush collection info is nil.
+	suite.Nil(flushCollectionInfo)
+
 	// Verify collection appears in soft deleted list
 	id = testCollection.ID.String()
 	softDeletedResults, err = suite.coordinator.GetSoftDeletedCollections(ctx, &id, suite.tenantName, suite.databaseName, 10)

go/pkg/sysdb/coordinator/table_catalog.go

@@ -754,8 +754,20 @@ func (tc *Catalog) FlushCollectionCompaction(ctx context.Context, flushCollectio
 	}
 
 	err := tc.txImpl.Transaction(ctx, func(txCtx context.Context) error {
+		// Check if collection exists.
+		collection, err := tc.metaDomain.CollectionDb(txCtx).GetCollectionEntry(types.FromUniqueID(flushCollectionCompaction.ID), nil)
+		if err != nil {
+			return err
+		}
+		if collection == nil {
+			return common.ErrCollectionNotFound
+		}
+		if collection.IsDeleted {
+			return common.ErrCollectionSoftDeleted
+		}
+
 		// register files to Segment metadata
-		err := tc.metaDomain.SegmentDb(txCtx).RegisterFilePaths(flushCollectionCompaction.FlushSegmentCompactions)
+		err = tc.metaDomain.SegmentDb(txCtx).RegisterFilePaths(flushCollectionCompaction.FlushSegmentCompactions)
 		if err != nil {
 			return err
 		}

go/pkg/sysdb/metastore/db/dao/collection.go

@@ -27,10 +27,16 @@ func (s *collectionDb) DeleteAll() error {
 
 func (s *collectionDb) GetCollectionEntry(collectionID *string, databaseName *string) (*dbmodel.Collection, error) {
 	var collections []*dbmodel.Collection
-	err := s.db.Table("collections").
+	query := s.db.Table("collections").
 		Select("collections.id, collections.name, collections.database_id, collections.is_deleted, databases.name, databases.tenant_id").
 		Joins("INNER JOIN databases ON collections.database_id = databases.id").
-		Where("collections.id = ? AND databases.name = ?", collectionID, databaseName).Find(&collections).Error
+		Where("collections.id = ?", collectionID)
+
+	if databaseName != nil && *databaseName != "" {
+		query = query.Where("databases.name = ?", databaseName)
+	}
+
+	err := query.Find(&collections).Error
 	if err != nil {
 		return nil, err
 	}