milvus/internal/proxy/function_task.go

// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package proxy

import (
	"context"

	"google.golang.org/protobuf/proto"

	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
	"github.com/milvus-io/milvus-proto/go-api/v3/milvuspb"
	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
	"github.com/milvus-io/milvus/internal/types"
	"github.com/milvus-io/milvus/internal/util/function/validator"
	"github.com/milvus-io/milvus/pkg/v3/mlog"
	"github.com/milvus-io/milvus/pkg/v3/util/commonpbutil"
	"github.com/milvus-io/milvus/pkg/v3/util/merr"
	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
)

const externalCollectionFunctionMutationUnsupportedMsg = "external collection does not support altering or dropping functions"

func rejectExternalCollectionFunctionMutation(schema *schemapb.CollectionSchema) error {
	if typeutil.IsExternalCollection(schema) {
		return merr.WrapErrParameterInvalidMsg(externalCollectionFunctionMutationUnsupportedMsg)
	}
	return nil
}

type alterCollectionFunctionTask struct {
	baseTask
	Condition
	*milvuspb.AlterCollectionFunctionRequest
	ctx      context.Context
	mixCoord types.MixCoordClient
	result   *commonpb.Status
}

func (t *alterCollectionFunctionTask) TraceCtx() context.Context {
	return t.ctx
}

func (t *alterCollectionFunctionTask) ID() UniqueID {
	return t.Base.MsgID
}

func (t *alterCollectionFunctionTask) SetID(uid UniqueID) {
	t.Base.MsgID = uid
}

func (t *alterCollectionFunctionTask) Type() commonpb.MsgType {
	return t.Base.MsgType
}

func (t *alterCollectionFunctionTask) BeginTs() Timestamp {
	return t.Base.Timestamp
}

func (t *alterCollectionFunctionTask) EndTs() Timestamp {
	return t.Base.Timestamp
}

func (t *alterCollectionFunctionTask) SetTs(ts Timestamp) {
	t.Base.Timestamp = ts
}

func (t *alterCollectionFunctionTask) OnEnqueue() error {
	if t.Base == nil {
		t.Base = commonpbutil.NewMsgBase()
	}
	t.Base.MsgType = commonpb.MsgType_AlterCollectionFunction
	t.Base.SourceID = paramtable.GetNodeID()
	return nil
}

func (t *alterCollectionFunctionTask) Name() string {
	return AlterCollectionFunctionTask
}

func (t *alterCollectionFunctionTask) PreExecute(ctx context.Context) error {
	if t.FunctionSchema == nil {
		return merr.WrapErrParameterInvalidMsg("function schema is empty")
	}
	if t.FunctionSchema.Type == schemapb.FunctionType_BM25 {
		return merr.WrapErrParameterInvalidMsg("currently does not support alter BM25 function")
	}
	if t.FunctionName != t.FunctionSchema.Name {
		return merr.WrapErrParameterInvalidMsg("invalid function config, name not match")
	}
	coll, err := getCollectionInfo(ctx, t.GetDbName(), t.GetCollectionName())
	if err != nil {
		mlog.Error(t.ctx, "AddCollectionTask, get collection info failed",
			mlog.String("dbName", t.GetDbName()),
			mlog.String("collectionName", t.GetCollectionName()),
			mlog.Err(err))
		return err
	}
	if err := rejectExternalCollectionFunctionMutation(coll.schema.CollectionSchema); err != nil {
		return err
	}
	funcExist := false
	newFunctions := []*schemapb.FunctionSchema{}
	for _, fSchema := range coll.schema.Functions {
		if t.FunctionName == fSchema.Name {
			if fSchema.Type == schemapb.FunctionType_BM25 {
				return merr.WrapErrParameterInvalidMsg("currently does not support alter BM25 function")
			}
			if err := validator.CheckFunctionAlterAllowed(fSchema, t.FunctionSchema); err != nil {
				return err
			}
			newFunctions = append(newFunctions, t.FunctionSchema)
			funcExist = true
		} else {
			newFunctions = append(newFunctions, fSchema)
		}
	}
	if !funcExist {
		return merr.WrapErrParameterInvalidMsg("function %s not found", t.FunctionName)
	}

	newColl := proto.Clone(coll.schema.CollectionSchema).(*schemapb.CollectionSchema)
	newColl.Functions = newFunctions
	if err := validator.ValidateFunction(newColl, t.FunctionName, false); err != nil {
		return err
	}
	return nil
}

func (t *alterCollectionFunctionTask) Execute(ctx context.Context) error {
	var err error
	t.result, err = t.mixCoord.AlterCollectionFunction(ctx, t.AlterCollectionFunctionRequest)
	if err = merr.CheckRPCCall(t.result, err); err != nil {
		return err
	}
	return nil
}

func (t *alterCollectionFunctionTask) PostExecute(ctx context.Context) error {
	return nil
}

func getCollectionInfo(ctx context.Context, dbName string, collectionName string) (*collectionInfo, error) {
	collID, err := globalMetaCache.GetCollectionID(ctx, dbName, collectionName)
	if err != nil {
		return nil, err
	}
	coll, err := globalMetaCache.GetCollectionInfo(ctx, dbName, collectionName, collID)
	if err != nil {
		return nil, err
	}
	coll.schema.DbName = dbName
	return coll, nil
}