retrieve any (ready/not-ready) task

Author: rescrv

go/pkg/sysdb/coordinator/task.go

@@ -92,7 +92,7 @@ func (s *Coordinator) AttachFunction(ctx context.Context, req *coordinatorpb.Att
 	err := s.catalog.txImpl.Transaction(ctx, func(txCtx context.Context) error {
 		// Check if there's any active (ready, non-deleted) attached function for this collection
 		// We only allow one active attached function per collection
-		existingAttachedFunctions, err := s.catalog.metaDomain.AttachedFunctionDb(txCtx).GetByCollectionID(req.InputCollectionId)
+		existingAttachedFunctions, err := s.catalog.metaDomain.AttachedFunctionDb(txCtx).GetAnyByCollectionID(req.InputCollectionId)
 		if err != nil {
 			log.Error("AttachFunction: failed to check for existing attached function", zap.Error(err))
 			return err

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

@@ -153,6 +153,23 @@ func (s *attachedFunctionDb) GetByCollectionID(inputCollectionID string) ([]*dbm
 	return attachedFunctions, nil
 }
 
+// Returns the non-deleted functions, without regard for `is_ready`.  Deleted functions will still
+// be excluded.
+func (s *attachedFunctionDb) GetAnyByCollectionID(inputCollectionID string) ([]*dbmodel.AttachedFunction, error) {
+	var attachedFunctions []*dbmodel.AttachedFunction
+	err := s.db.
+		Where("input_collection_id = ?", inputCollectionID).
+		Where("is_deleted = ?", false).
+		Find(&attachedFunctions).Error
+
+	if err != nil {
+		log.Error("GetByCollectionID failed", zap.Error(err), zap.String("input_collection_id", inputCollectionID))
+		return nil, err
+	}
+
+	return attachedFunctions, nil
+}
+
 func (s *attachedFunctionDb) SoftDelete(inputCollectionID string, name string) error {
 	// Update name and is_deleted in a single query
 	// Format: _deleted_<original_name>_<id>

go/pkg/sysdb/metastore/db/dbmodel/task.go

@@ -42,6 +42,7 @@ type IAttachedFunctionDb interface {
 	GetByID(id uuid.UUID) (*AttachedFunction, error)
 	GetAnyByID(id uuid.UUID) (*AttachedFunction, error) // TODO(tanujnay112): Consolidate all the getters.
 	GetByCollectionID(inputCollectionID string) ([]*AttachedFunction, error)
+	GetAnyByCollectionID(inputCollectionID string) ([]*AttachedFunction, error)
 	Update(attachedFunction *AttachedFunction) error
 	Finish(id uuid.UUID) error
 	SoftDelete(inputCollectionID string, name string) error