Repository: lf-edge/ekuiper
Pull request #3010 (2024-07-15 03:30:48)

fix(topo): detach subtopo from closed rule

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3010?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `8.33333%` with `11 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@462e712`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3010?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/topo/node/node.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3010?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Fnode.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL25vZGUuZ28=) | 0.00% | [9 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3010?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/subtopo.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3010?src=pr&el=tree&filepath=internal%2Ftopo%2Fsubtopo.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9zdWJ0b3BvLmdv) | 33.33% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3010?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #3010   +/-   ##
============================================
  Coverage             ?   67.86%           
============================================
  Files                ?      353           
  Lines                ?    34218           
  Branches             ?        0           
============================================
  Hits                 ?    23222           
  Misses               ?     8930           
  Partials             ?     2066           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3010?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/internal/converter/delimited/converter.go b/internal/converter/delimited/converter.go
index b70f024e..65730d46 100644
--- a/internal/converter/delimited/converter.go
+++ b/internal/converter/delimited/converter.go
@@ -16,6 +16,7 @@ package delimited
 
 import (
 	"bytes"
+	"encoding/binary"
 	"fmt"
 	"sort"
 	"strconv"
@@ -29,7 +30,8 @@ import (
 
 type Converter struct {
 	Delimiter string   `json:"delimiter"`
-	Cols      []string `json:"cols"`
+	Cols      []string `json:"fields"`
+	HasHeader bool     `json:"hasHeader"`
 }
 
 func NewConverter(props map[string]any) (message.Converter, error) {
@@ -56,7 +58,7 @@ func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 		}
 	}()
 	switch m := d.(type) {
-	case map[string]interface{}:
+	case map[string]any:
 		sb := &bytes.Buffer{}
 		if len(c.Cols) == 0 {
 			keys := make([]string, 0, len(m))
@@ -65,8 +67,14 @@ func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 			}
 			sort.Strings(keys)
 			c.Cols = keys
+			if len(c.Cols) > 0 && c.HasHeader {
+				hb := []byte(strings.Join(c.Cols, c.Delimiter))
+				sb.WriteString(c.Delimiter)
+				_ = binary.Write(sb, binary.BigEndian, uint32(len(hb)))
+				sb.Write(hb)
+				ctx.GetLogger().Infof("delimiter header %s", hb)
+			}
 		}
-
 		for i, v := range c.Cols {
 			if i > 0 {
 				sb.WriteString(c.Delimiter)
@@ -74,6 +82,34 @@ func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 			fmt.Fprintf(sb, "%v", m[v])
 		}
 		return sb.Bytes(), nil
+	case []map[string]any:
+		sb := &bytes.Buffer{}
+		var cols []string
+		for i, mm := range m {
+			if i > 0 {
+				sb.WriteString("\n")
+			}
+			if len(cols) == 0 {
+				keys := make([]string, 0, len(mm))
+				for k := range mm {
+					keys = append(keys, k)
+				}
+				sort.Strings(keys)
+				cols = keys
+				if len(cols) > 0 && c.HasHeader {
+					hb := []byte(strings.Join(cols, c.Delimiter))
+					sb.Write(hb)
+					sb.WriteString("\n")
+				}
+			}
+			for j, v := range cols {
+				if j > 0 {
+					sb.WriteString(c.Delimiter)
+				}
+				fmt.Fprintf(sb, "%v", mm[v])
+			}
+		}
+		return sb.Bytes(), nil
 	default:
 		return nil, fmt.Errorf("unsupported type %v, must be a map", d)
 	}
diff --git a/internal/io/file/sink.go b/internal/io/file/sink.go
index f0669e5a..07497b07 100644
--- a/internal/io/file/sink.go
+++ b/internal/io/file/sink.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,10 +15,11 @@
 package file
 
 import (
+	"bytes"
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"path/filepath"
-	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -55,6 +56,7 @@ type fileSink struct {
 	mux      sync.Mutex
 	fws      map[string]*fileWriter
 	rollHook modules.RollHook
+	headers  string
 }
 
 func (m *fileSink) Provision(ctx api.StreamContext, props map[string]interface{}) error {
@@ -166,7 +168,7 @@ func (m *fileSink) Collect(ctx api.StreamContext, tuple api.RawTuple) error {
 		}
 	}
 	ctx.GetLogger().Debugf("writing to file path %s", fn)
-	fw, err := m.GetFws(ctx, fn, item)
+	fw, item, err := m.GetFws(ctx, fn, item)
 	if err != nil {
 		return err
 	}
@@ -234,41 +236,31 @@ func (m *fileSink) roll(ctx api.StreamContext, k string, v *fileWriter) error {
 
 // GetFws returns the file writer for the given file name, if the file writer does not exist, it will create one
 // The item is used to get the csv header if needed
-func (m *fileSink) GetFws(ctx api.StreamContext, fn string, item interface{}) (*fileWriter, error) {
+func (m *fileSink) GetFws(ctx api.StreamContext, fn string, item []byte) (*fileWriter, []byte, error) {
 	m.mux.Lock()
 	defer m.mux.Unlock()
+	if m.c.FileType == CSV_TYPE && m.c.HasHeader && m.headers == "" {
+		if len(m.c.Fields) > 0 {
+			m.headers = strings.Join(m.c.Fields, m.c.Delimiter)
+		} else {
+			db := []byte(m.c.Delimiter)
+			if bytes.HasPrefix(item, db) {
+				cursor := len(db)
+				nextCursor := cursor + 4
+				hl := binary.BigEndian.Uint32(item[cursor:nextCursor])
+				cursor = nextCursor
+				nextCursor = cursor + int(hl)
+				hb := item[cursor:nextCursor]
+				m.headers = string(hb)
+				ctx.GetLogger().Infof("csv header %s", hb)
+				cursor = nextCursor
+				item = item[nextCursor:]
+			}
+		}
+	}
 	fws, ok := m.fws[fn]
 	if !ok {
 		var e error
-		// extract header for csv
-		var headers string
-		if m.c.FileType == CSV_TYPE && m.c.HasHeader {
-			var header []string
-			if len(m.c.Fields) > 0 {
-				header = m.c.Fields
-			} else {
-				switch v := item.(type) {
-				case map[string]interface{}:
-					header = make([]string, len(v))
-					i := 0
-					for k := range item.(map[string]interface{}) {
-						header[i] = k
-						i++
-					}
-				case []map[string]interface{}:
-					if len(v) > 0 {
-						header = make([]string, len(v[0]))
-						i := 0
-						for k := range v[0] {
-							header[i] = k
-							i++
-						}
-					}
-				}
-				sort.Strings(header)
-			}
-			headers = strings.Join(header, m.c.Delimiter)
-		}
 		nfn := fn
 		if m.c.RollingNamePattern != "" {
 			newFile := ""
@@ -286,13 +278,13 @@ func (m *fileSink) GetFws(ctx api.StreamContext, fn string, item interface{}) (*
 			nfn = filepath.Join(fileDir, newFile)
 		}
 
-		fws, e = m.createFileWriter(ctx, nfn, m.c.FileType, headers, m.c.Compression, m.c.Encryption)
+		fws, e = m.createFileWriter(ctx, nfn, m.c.FileType, m.headers, m.c.Compression, m.c.Encryption)
 		if e != nil {
-			return nil, e
+			return nil, item, e
 		}
 		m.fws[fn] = fws
 	}
-	return fws, nil
+	return fws, item, nil
 }
 
 func GetSink() api.Sink {
diff --git a/internal/topo/node/contract.go b/internal/topo/node/contract.go
index 2a10ad12..007a0ff2 100644
--- a/internal/topo/node/contract.go
+++ b/internal/topo/node/contract.go
@@ -23,6 +23,7 @@ import (
 
 type Emitter interface {
 	AddOutput(chan<- interface{}, string) error
+	RemoveOutput(string) error
 }
 
 type Collector interface {
@@ -83,7 +84,7 @@ type MergeableTopo interface {
 	// SubMetrics return the metrics of the sub nodes
 	SubMetrics() ([]string, []any)
 	// Close notifies subtopo to deref
-	Close(ctx api.StreamContext, ruleId string)
+	Close(ctx api.StreamContext, ruleId string, runId int)
 }
 
 type SchemaNode interface {
diff --git a/internal/topo/node/encode_op.go b/internal/topo/node/encode_op.go
index 5ba4ae33..696d9b4e 100644
--- a/internal/topo/node/encode_op.go
+++ b/internal/topo/node/encode_op.go
@@ -35,7 +35,7 @@ type EncodeOp struct {
 }
 
 func NewEncodeOp(ctx api.StreamContext, name string, rOpt *def.RuleOption, sc *SinkConf) (*EncodeOp, error) {
-	c, err := converter.GetOrCreateConverter(ctx, sc.Format, sc.SchemaId, nil, map[string]any{"delimiter": sc.Delimiter})
+	c, err := converter.GetOrCreateConverter(ctx, sc.Format, sc.SchemaId, nil, map[string]any{"delimiter": sc.Delimiter, "hasHeader": sc.HasHeader})
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
index eb8945e4..8ebb7f5f 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -17,6 +17,7 @@ package node
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
@@ -56,13 +57,28 @@ func newDefaultNode(name string, options *def.RuleOption) *defaultNode {
 	}
 }
 
-func (o *defaultNode) AddOutput(output chan<- interface{}, name string) error {
+func (o *defaultNode) AddOutput(output chan<- any, name string) error {
 	o.outputMu.Lock()
 	defer o.outputMu.Unlock()
 	o.outputs[name] = output
 	return nil
 }
 
+func (o *defaultNode) RemoveOutput(name string) error {
+	o.outputMu.Lock()
+	defer o.outputMu.Unlock()
+	namePre := name + "_"
+	for n := range o.outputs {
+		if strings.HasPrefix(n, namePre) {
+			delete(o.outputs, n)
+			if o.ctx != nil {
+				o.ctx.GetLogger().Infof("Remove output %s from %s", n, o.name)
+			}
+		}
+	}
+	return nil
+}
+
 func (o *defaultNode) GetName() string {
 	return o.name
 }
@@ -84,11 +100,11 @@ func (o *defaultNode) RemoveMetrics(ruleId string) {
 	}
 }
 
-func (o *defaultNode) Broadcast(val interface{}) {
+func (o *defaultNode) Broadcast(val any) {
 	o.BroadcastCustomized(val, o.doBroadcast)
 }
 
-func (o *defaultNode) BroadcastCustomized(val interface{}, broadcastFunc func(val any)) {
+func (o *defaultNode) BroadcastCustomized(val any, broadcastFunc func(val any)) {
 	if _, ok := val.(error); ok && !o.sendError {
 		return
 	}
@@ -104,7 +120,7 @@ func (o *defaultNode) BroadcastCustomized(val interface{}, broadcastFunc func(va
 	return
 }
 
-func (o *defaultNode) doBroadcast(val interface{}) {
+func (o *defaultNode) doBroadcast(val any) {
 	o.outputMu.RLock()
 	defer o.outputMu.RUnlock()
 	l := len(o.outputs)
@@ -151,7 +167,7 @@ func newDefaultSinkNode(name string, options *def.RuleOption) *defaultSinkNode {
 	}
 }
 
-func (o *defaultSinkNode) GetInput() (chan<- interface{}, string) {
+func (o *defaultSinkNode) GetInput() (chan<- any, string) {
 	return o.input, o.name
 }
 
@@ -236,7 +252,7 @@ func (o *defaultSinkNode) handleEof(ctx api.StreamContext, d xsql.EOFTuple) {
 	}
 }
 
-func SourcePing(sourceType string, config map[string]interface{}) error {
+func SourcePing(sourceType string, config map[string]any) error {
 	source, err := io.Source(sourceType)
 	if err != nil {
 		return err
diff --git a/internal/topo/node/props.go b/internal/topo/node/props.go
index 698f10d8..6e843412 100644
--- a/internal/topo/node/props.go
+++ b/internal/topo/node/props.go
@@ -38,6 +38,7 @@ type SinkConf struct {
 	LingerInterval cast.DurationConf `json:"lingerInterval"`
 	Compression    string            `json:"compression"`
 	Encryption     string            `json:"encryption"`
+	HasHeader      bool              `json:"hasHeader"`
 	conf.SinkConf
 }
 
diff --git a/internal/topo/subtopo.go b/internal/topo/subtopo.go
index 8ce506f2..f8ab13e0 100644
--- a/internal/topo/subtopo.go
+++ b/internal/topo/subtopo.go
@@ -65,6 +65,10 @@ func (s *SrcSubTopo) AddOutput(output chan<- interface{}, name string) error {
 	return s.tail.AddOutput(output, name)
 }
 
+func (s *SrcSubTopo) RemoveOutput(name string) error {
+	return s.tail.RemoveOutput(name)
+}
+
 func (s *SrcSubTopo) Open(ctx api.StreamContext, parentErrCh chan<- error) {
 	// Update the ref count
 	if _, loaded := s.refRules.LoadOrStore(ctx.GetRuleId(), parentErrCh); !loaded {
@@ -171,7 +175,7 @@ func (s *SrcSubTopo) StoreSchema(ruleID, dataSource string, schema map[string]*a
 	}
 }
 
-func (s *SrcSubTopo) Close(ctx api.StreamContext, ruleId string) {
+func (s *SrcSubTopo) Close(ctx api.StreamContext, ruleId string, runId int) {
 	if _, ok := s.refRules.LoadAndDelete(ruleId); ok {
 		s.refCount.Add(-1)
 		if s.refCount.Load() == 0 {
@@ -186,6 +190,7 @@ func (s *SrcSubTopo) Close(ctx api.StreamContext, ruleId string) {
 			}
 		}
 	}
+	_ = s.RemoveOutput(fmt.Sprintf("%s.%d", ruleId, runId))
 }
 
 // RemoveMetrics is called when the rule is deleted
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index 019c7fd0..c498e3e0 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -39,6 +39,10 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
+var uid atomic.Uint32
+
+// Topo is the runtime DAG for a rule
+// It only run once. If the rule restarts, another topo is created.
 type Topo struct {
 	streams     []string
 	sources     []node.DataSourceNode
@@ -48,6 +52,7 @@ type Topo struct {
 	drain       chan error
 	ops         []node.OperatorNode
 	name        string
+	runId       int
 	options     *def.RuleOption
 	store       api.Store
 	coordinator *checkpoint.Coordinator
@@ -59,8 +64,10 @@ type Topo struct {
 }
 
 func NewWithNameAndOptions(name string, options *def.RuleOption) (*Topo, error) {
+	id := uid.Add(1)
 	tp := &Topo{
 		name:    name,
+		runId:   int(id),
 		options: options,
 		topo: &def.PrintableTopo{
 			Sources: make([]string, 0),
@@ -115,7 +122,7 @@ func (s *Topo) Cancel() {
 	s.coordinator = nil
 	for _, src := range s.sources {
 		if rt, ok := src.(node.MergeableTopo); ok {
-			rt.Close(s.ctx, s.name)
+			rt.Close(s.ctx, s.name, s.runId)
 		}
 	}
 }
@@ -158,7 +165,7 @@ func (s *Topo) AddOperator(inputs []node.Emitter, operator node.OperatorNode) *T
 	ch, opName := operator.GetInput()
 	for _, input := range inputs {
 		// add rule id to make operator name unique
-		_ = input.AddOutput(ch, fmt.Sprintf("%s_%s", s.name, opName))
+		_ = input.AddOutput(ch, fmt.Sprintf("%s.%d_%s", s.name, s.runId, opName))
 		operator.AddInputCount()
 		switch rt := input.(type) {
 		case node.MergeableTopo:


Test patch:
diff --git a/internal/converter/delimited/converter_test.go b/internal/converter/delimited/converter_test.go
index a0c0b5fa..387fccce 100644
--- a/internal/converter/delimited/converter_test.go
+++ b/internal/converter/delimited/converter_test.go
@@ -19,6 +19,7 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
@@ -28,17 +29,21 @@ import (
 
 func TestEncode(t *testing.T) {
 	tests := []struct {
-		m map[string]interface{}
-		r []byte
-		e string
+		name string
+		m    any
+		r    []byte
+		e    string
 	}{
 		{
+			name: "normal",
 			m: map[string]interface{}{
 				"id":   12,
 				"name": "test",
 			},
 			r: []byte(`12:test`),
-		}, {
+		},
+		{
+			name: "embedded",
 			m: map[string]interface{}{
 				"id":   7,
 				"name": "John Doe",
@@ -54,20 +59,97 @@ func TestEncode(t *testing.T) {
 			},
 			r: []byte(`22:map[indoor:[Chess] outdoor:[Basketball]]:7:John Doe`),
 		},
+		{
+			name: "list",
+			m: []map[string]interface{}{
+				{
+					"id":   12,
+					"name": "test",
+				},
+				{
+					"id":   14,
+					"name": "test2",
+				},
+			},
+			r: []byte("12:test\n14:test2"),
+		},
 	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	ctx := mockContext.NewMockContext("test", "op1")
-	for i, tt := range tests {
-		c, err := NewConverter(map[string]any{"delimiter": ":"})
-		if err != nil {
-			t.Fatal(err)
-		}
-		a, err := c.Encode(ctx, tt.m)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%s\n\ngot=%s\n\n", i, tt.r, a)
-		}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c, err := NewConverter(map[string]any{"delimiter": ":"})
+			assert.NoError(t, err)
+			a, err := c.Encode(ctx, tt.m)
+			if tt.e != "" {
+				assert.EqualError(t, err, tt.e)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.r, a)
+			}
+		})
+	}
+}
+
+func TestEncodeWithHeader(t *testing.T) {
+	tests := []struct {
+		name string
+		m    any
+		r    []byte
+		e    string
+	}{
+		{
+			name: "normal",
+			m: map[string]interface{}{
+				"id":   12,
+				"name": "test",
+			},
+			r: []byte{0x3a, 0x0, 0x0, 0x0, 0x7, 0x69, 0x64, 0x3a, 0x6e, 0x61, 0x6d, 0x65, 0x31, 0x32, 0x3a, 0x74, 0x65, 0x73, 0x74},
+		},
+		{
+			name: "embedded",
+			m: map[string]interface{}{
+				"id":   7,
+				"name": "John Doe",
+				"age":  22,
+				"hobbies": map[string]interface{}{
+					"indoor": []string{
+						"Chess",
+					},
+					"outdoor": []string{
+						"Basketball",
+					},
+				},
+			},
+			r: []byte{0x3a, 0x0, 0x0, 0x0, 0x13, 0x61, 0x67, 0x65, 0x3a, 0x68, 0x6f, 0x62, 0x62, 0x69, 0x65, 0x73, 0x3a, 0x69, 0x64, 0x3a, 0x6e, 0x61, 0x6d, 0x65, 0x32, 0x32, 0x3a, 0x6d, 0x61, 0x70, 0x5b, 0x69, 0x6e, 0x64, 0x6f, 0x6f, 0x72, 0x3a, 0x5b, 0x43, 0x68, 0x65, 0x73, 0x73, 0x5d, 0x20, 0x6f, 0x75, 0x74, 0x64, 0x6f, 0x6f, 0x72, 0x3a, 0x5b, 0x42, 0x61, 0x73, 0x6b, 0x65, 0x74, 0x62, 0x61, 0x6c, 0x6c, 0x5d, 0x5d, 0x3a, 0x37, 0x3a, 0x4a, 0x6f, 0x68, 0x6e, 0x20, 0x44, 0x6f, 0x65},
+		},
+		{
+			name: "list",
+			m: []map[string]interface{}{
+				{
+					"id":   12,
+					"name": "test",
+				},
+				{
+					"id":   14,
+					"name": "test2",
+				},
+			},
+			r: []byte("id:name\n12:test\n14:test2"),
+		},
+	}
+	ctx := mockContext.NewMockContext("test", "op1")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c, err := NewConverter(map[string]any{"delimiter": ":", "hasHeader": true})
+			assert.NoError(t, err)
+			a, err := c.Encode(ctx, tt.m)
+			if tt.e != "" {
+				assert.EqualError(t, err, tt.e)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.r, a)
+			}
+		})
 	}
 }
 
diff --git a/internal/io/file/sink_test.go b/internal/io/file/sink_test.go
index 02e8224e..97157e29 100644
--- a/internal/io/file/sink_test.go
+++ b/internal/io/file/sink_test.go
@@ -1,4 +1,4 @@
-// Copyright 2023 EMQ Technologies Co., Ltd.
+// Copyright 2023-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -353,6 +353,69 @@ func TestFileSink_Collect(t *testing.T) {
 	}
 }
 
+// Test single message with header defined
+func TestCSVSingMessHeader(t *testing.T) {
+	tests := []struct {
+		name    string
+		fname   string
+		content []byte
+	}{
+		{
+			name:    "csv",
+			fname:   "test_csvh",
+			content: []byte("id:name\n12:test\n11:value2"),
+		},
+	}
+
+	ctx := mockContext.NewMockContext("test1", "test")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a temporary file for testing
+			tmpfile, err := os.CreateTemp("", tt.fname)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer os.Remove(tmpfile.Name())
+			// Create a file sink with the temporary file path
+			sink := &fileSink{}
+			f := message.FormatDelimited
+			err = sink.Provision(ctx, map[string]interface{}{
+				"path":               tmpfile.Name(),
+				"fileType":           CSV_TYPE,
+				"hasHeader":          true,
+				"delimiter":          ":",
+				"format":             f,
+				"rollingNamePattern": "none",
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = sink.Connect(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if err := sink.Collect(ctx, &xsql.RawTuple{Rawdata: []byte{0x3a, 0x0, 0x0, 0x0, 0x7, 0x69, 0x64, 0x3a, 0x6e, 0x61, 0x6d, 0x65, 0x31, 0x32, 0x3a, 0x74, 0x65, 0x73, 0x74}}); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			// Test collecting another map item
+			if err := sink.Collect(ctx, &xsql.RawTuple{Rawdata: []byte("11:value2")}); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+			if err = sink.Close(ctx); err != nil {
+				t.Errorf("unexpected close error: %s", err)
+			}
+			// Read the contents of the temporary file and check if they match the collected items
+			contents, err := os.ReadFile(tmpfile.Name())
+			if err != nil {
+				t.Fatal(err)
+			}
+			assert.Equal(t, tt.content, contents)
+		})
+	}
+}
+
 // Test file rolling by time
 func TestFileSinkRolling_Collect(t *testing.T) {
 	// Remove existing files
diff --git a/internal/topo/node/node_test.go b/internal/topo/node/node_test.go
new file mode 100644
index 00000000..9cd73f92
--- /dev/null
+++ b/internal/topo/node/node_test.go
@@ -0,0 +1,36 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed 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 node
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+)
+
+func TestOutputs(t *testing.T) {
+	n := newDefaultNode("test", &def.RuleOption{})
+	err := n.AddOutput(make(chan<- any), "rule.1_test")
+	assert.NoError(t, err)
+	err = n.AddOutput(make(chan<- any), "rule.2_test")
+	assert.NoError(t, err)
+	err = n.RemoveOutput("rule.1")
+	assert.NoError(t, err)
+	err = n.RemoveOutput("rule.4")
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(n.outputs))
+}
diff --git a/internal/topo/subtopo_test.go b/internal/topo/subtopo_test.go
index ec31f206..9f9030a9 100644
--- a/internal/topo/subtopo_test.go
+++ b/internal/topo/subtopo_test.go
@@ -108,10 +108,10 @@ func TestSubtopoLC(t *testing.T) {
 	assert.Equal(t, []checkpoint.StreamTask{srcNode}, sources)
 	assert.Equal(t, []checkpoint.NonSourceTask{opNode}, ops)
 	// Stop
-	subTopo.Close(ctx1, "rule1")
+	subTopo.Close(ctx1, "rule1", 1)
 	assert.Equal(t, int32(1), subTopo.refCount.Load())
 	assert.Equal(t, 1, mlen(&subTopoPool))
-	subTopo2.Close(ctx2, "rule2")
+	subTopo2.Close(ctx2, "rule2", 2)
 	assert.Equal(t, int32(0), subTopo.refCount.Load())
 	assert.Equal(t, 0, mlen(&subTopoPool))
 	assert.Equal(t, 2, len(subTopo.schemaReg))
@@ -138,7 +138,7 @@ func TestSubtopoRunError(t *testing.T) {
 	subTopo.Open(ctx1, make(chan error))
 	assert.Equal(t, int32(1), subTopo.refCount.Load())
 	assert.Equal(t, true, subTopo.opened.Load())
-	subTopo.Close(ctx1, "rule1")
+	subTopo.Close(ctx1, "rule1", 1)
 	assert.Equal(t, int32(0), subTopo.refCount.Load())
 	assert.Equal(t, 0, mlen(&subTopoPool))
 	time.Sleep(10 * time.Millisecond)
@@ -156,14 +156,14 @@ func TestSubtopoRunError(t *testing.T) {
 	select {
 	case err := <-errCh1:
 		assert.Equal(t, assert.AnError, err)
-		subTopo.Close(ctx1, "rule1")
+		subTopo.Close(ctx1, "rule1", 1)
 	case <-time.After(1 * time.Second):
 		assert.Fail(t, "Should receive error")
 	}
 	select {
 	case err := <-errCh2:
 		assert.Equal(t, assert.AnError, err)
-		subTopo2.Close(ctx2, "rule2")
+		subTopo2.Close(ctx2, "rule2", 2)
 	case <-time.After(1 * time.Second):
 		assert.Fail(t, "Should receive error")
 	}
@@ -239,6 +239,11 @@ func (m *mockSrc) AddOutput(c chan<- interface{}, s string) error {
 	return nil
 }
 
+func (m *mockSrc) RemoveOutput(s string) error {
+	m.outputs = m.outputs[1:]
+	return nil
+}
+
 func (m *mockSrc) Open(ctx api.StreamContext, errCh chan<- error) {
 	if m.runCount%3 != 0 {
 		fmt.Printf("sent error for %d \n", m.runCount)
@@ -273,6 +278,13 @@ type mockOp struct {
 	schemaCount int
 }
 
+func (m *mockOp) RemoveOutput(s string) error {
+	if len(m.outputs) > 0 {
+		m.outputs = m.outputs[1:]
+	}
+	return nil
+}
+
 func (m *mockOp) AddOutput(c chan<- interface{}, s string) error {
 	m.outputs = append(m.outputs, c)
 	return nil