fix:[2.5] analyzer memory leak because function runner not close

internal/datanode/importv2/util.go

@@ -214,6 +214,7 @@
 		if err != nil {
 			return err
 		}
+		defer runner.Close()
 
 		inputFieldIDs := lo.Map(runner.GetInputFields(), func(field *schemapb.FieldSchema, _ int) int64 { return field.GetFieldID() })
 		inputDatas := make([]any, 0, len(inputFieldIDs))

internal/flushcommon/pipeline/flow_graph_embedding_node.go

@@ -165,6 +165,12 @@
 	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

@@ -203,6 +203,12 @@
 	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 @@
 	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/multi_analyzer_bm25_function.go

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

internal/util/pipeline/node.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/pipeline.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()
 		}