fix analyzer memroy leak because function runner not close

Signed-off-by: aoiasd <[email protected]>

Author: aoiasd

internal/datanode/importv2/util.go

@@ -233,6 +233,8 @@ func RunBm25Function(task *ImportTask, data *storage.InsertData) error {
 			continue
 		}
 
+		defer runner.Close()
+
 		inputFieldIDs := lo.Map(runner.GetInputFields(), func(field *schemapb.FieldSchema, _ int) int64 { return field.GetFieldID() })
 		inputDatas := make([]any, 0, len(inputFieldIDs))
 		for _, inputFieldID := range inputFieldIDs {

internal/flushcommon/pipeline/flow_graph_embedding_node.go

@@ -168,6 +168,12 @@ func (eNode *embeddingNode) Operate(in []Msg) []Msg {
 	return []Msg{fgMsg}
 }
 
+func (eNode *embeddingNode) Close() {
+	for _, runner := range eNode.functionRunners {
+		runner.Close()
+	}
+}
+
 func BuildSparseFieldData(array *schemapb.SparseFloatArray) storage.FieldData {
 	return &storage.SparseFloatVectorFieldData{
 		SparseFloatArray: schemapb.SparseFloatArray{

internal/querynodev2/pipeline/embedding_node.go

@@ -206,6 +206,12 @@ func (eNode *embeddingNode) Operate(in Msg) Msg {
 	return nodeMsg
 }
 
+func (eNode *embeddingNode) Close() {
+	for _, functionRunner := range eNode.functionRunners {
+		functionRunner.Close()
+	}
+}
+
 func getEmbeddingFieldDatas(datas []*schemapb.FieldData, fieldIDs ...int64) ([]any, error) {
 	result := []any{}
 	for _, fieldID := range fieldIDs {

internal/util/function/bm25_function.go

@@ -171,6 +171,10 @@ func (v *BM25FunctionRunner) GetInputFields() []*schemapb.FieldSchema {
 	return []*schemapb.FieldSchema{v.inputField}
 }
 
+func (v *BM25FunctionRunner) Close() {
+	v.tokenizer.Destroy()
+}
+
 func buildSparseFloatArray(mapdata []map[uint32]float32) *schemapb.SparseFloatArray {
 	dim := int64(0)
 	bytes := lo.Map(mapdata, func(sparseMap map[uint32]float32, _ int) []byte {

internal/util/function/function.go

@@ -30,6 +30,8 @@ type FunctionRunner interface {
 	GetSchema() *schemapb.FunctionSchema
 	GetOutputFields() []*schemapb.FieldSchema
 	GetInputFields() []*schemapb.FieldSchema
+
+	Close()
 }
 
 func NewFunctionRunner(coll *schemapb.CollectionSchema, schema *schemapb.FunctionSchema) (FunctionRunner, error) {

internal/util/function/mock_function.go

@@ -235,3 +235,9 @@ func (v *MultiAnalyzerBM25FunctionRunner) GetOutputFields() []*schemapb.FieldSch
 func (v *MultiAnalyzerBM25FunctionRunner) GetInputFields() []*schemapb.FieldSchema {
 	return v.inputFields
 }
+
+func (v *MultiAnalyzerBM25FunctionRunner) Close() {
+	for _, analyzer := range v.analyzers {
+		analyzer.Destroy()
+	}
+}

internal/util/function/multi_analyzer_bm25_function.go

@@ -24,6 +24,8 @@ type Node interface {
 	Name() string
 	MaxQueueLength() int32
 	Operate(in Msg) Msg
+
+	Close()
 }
 
 type nodeCtx struct {
@@ -56,6 +58,9 @@ func (node *BaseNode) MaxQueueLength() int32 {
 	return node.maxQueueLength
 }
 
+func (node *BaseNode) Close() {
+}
+
 func NewBaseNode(name string, maxQueryLength int32) *BaseNode {
 	return &BaseNode{
 		name:           name,

internal/util/pipeline/node.go

@@ -70,6 +70,7 @@ func (p *pipeline) Start() error {
 
 func (p *pipeline) Close() {
 	for _, node := range p.nodes {
+		node.node.Close()
 		if node.Checker != nil {
 			node.Checker.Close()
 		}