Repository: lf-edge/ekuiper
Pull request #3011 (2024-07-15 04:23:10)

feat: support load connections env

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3011?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 `81.25000%` with `9 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@64a121d`). [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/3011?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 |
|---|---|---|
| [pkg/env/env\_manager.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3011?src=pr&el=tree&filepath=pkg%2Fenv%2Fenv_manager.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-cGtnL2Vudi9lbnZfbWFuYWdlci5nbw==) | 82.98% | [5 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3011?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/server.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3011?src=pr&el=tree&filepath=internal%2Fserver%2Fserver.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3NlcnZlci5nbw==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3011?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    #3011   +/-   ##
============================================
  Coverage             ?   67.90%           
============================================
  Files                ?      354           
  Lines                ?    34254           
  Branches             ?        0           
============================================
  Hits                 ?    23258           
  Misses               ?     8921           
  Partials             ?     2075           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3011?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/conf/env_manager.go b/internal/conf/env_manager.go
new file mode 100644
index 00000000..1fef6ecb
--- /dev/null
+++ b/internal/conf/env_manager.go
@@ -0,0 +1,114 @@
+// 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 conf
+
+import (
+	"fmt"
+	"os"
+	"strings"
+)
+
+func init() {
+	globalEnvManager = &EnvManager{}
+	SetupEnv()
+}
+
+func SetupEnv() {
+	globalEnvManager.Setup()
+}
+
+func SetupConnectionProps() {
+	globalEnvManager.SetupConnectionProps()
+}
+
+func GetEnv() map[string]string {
+	return globalEnvManager.GetEnv()
+}
+
+var globalEnvManager *EnvManager
+
+type EnvManager struct {
+	env             map[string]string
+	connectionProps map[string]map[string]map[string]interface{}
+}
+
+func (e *EnvManager) Setup() {
+	e.loadEnv()
+}
+
+func (e *EnvManager) GetEnv() map[string]string {
+	if len(e.env) < 1 {
+		e.loadEnv()
+	}
+	return e.env
+}
+
+func (e *EnvManager) SetupConnectionProps() {
+	e.loadConnectionProps()
+	e.storeConnectionProps()
+	e.connectionProps = nil
+}
+
+func (e *EnvManager) loadEnv() {
+	got := os.Environ()
+	e.env = make(map[string]string)
+	for _, v := range got {
+		ss := strings.Split(v, "=")
+		if len(ss) != 2 {
+			continue
+		}
+		key := ss[0]
+		value := ss[1]
+		e.env[key] = value
+	}
+}
+
+func (e *EnvManager) loadConnectionProps() {
+	e.connectionProps = make(map[string]map[string]map[string]interface{})
+	for k, v := range e.env {
+		if !strings.HasPrefix(k, "CONNECTION") {
+			continue
+		}
+		ss := strings.Split(k, "__")
+		if len(ss) != 4 {
+			continue
+		}
+		pluginTyp := strings.ToLower(ss[1])
+		confName := strings.ToLower(ss[2])
+		confKey := strings.ToLower(ss[3])
+		v1, ok := e.connectionProps[pluginTyp]
+		if !ok {
+			v1 = make(map[string]map[string]interface{})
+			e.connectionProps[pluginTyp] = v1
+		}
+		v2, ok := v1[confName]
+		if !ok {
+			v2 = make(map[string]interface{})
+			v1[confName] = v2
+		}
+		v2[confKey] = v
+	}
+}
+
+func (e *EnvManager) storeConnectionProps() {
+	for pluginTyp, v := range e.connectionProps {
+		for confName, props := range v {
+			err := WriteCfgIntoKVStorage("connections", pluginTyp, confName, props)
+			if err != nil {
+				Log.Warn(fmt.Sprintf("load connections.%s.%s failed, err:%v", pluginTyp, confName, err))
+			}
+		}
+	}
+}
diff --git a/internal/conf/load.go b/internal/conf/load.go
index 567a3665..bf3a1383 100644
--- a/internal/conf/load.go
+++ b/internal/conf/load.go
@@ -71,7 +71,7 @@ func LoadConfigFromPath(p string, c interface{}) error {
 	}
 	// Make all keys to lowercase to match environment variables then revert it back by checking json defs
 	configs := normalize(configMap)
-	err = process(configs, os.Environ(), prefix)
+	err = process(configs, GetEnv(), prefix)
 	if err != nil {
 		return err
 	}
@@ -112,19 +112,15 @@ func getPrefix(p string) string {
 	return strings.ToUpper(strings.TrimSuffix(file, filepath.Ext(file)))
 }
 
-func process(configMap map[string]interface{}, variables []string, prefix string) error {
-	for _, e := range variables {
-		if !strings.HasPrefix(e, prefix) {
+func process(configMap map[string]interface{}, env map[string]string, prefix string) error {
+	for key, value := range env {
+		if !strings.HasPrefix(key, prefix) {
 			continue
 		}
-		pair := strings.SplitN(e, "=", 2)
-		if len(pair) != 2 {
-			return fmt.Errorf("wrong format of variable")
-		}
-		keys := nameToKeys(trimPrefix(pair[0], prefix))
-		handle(configMap, keys, pair[1])
+		keys := nameToKeys(trimPrefix(key, prefix))
+		handle(configMap, keys, value)
 		printableK := strings.Join(keys, ".")
-		printableV := pair[1]
+		printableV := value
 		if strings.Contains(strings.ToLower(printableK), "password") {
 			printableV = "*"
 		}
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/server/server.go b/internal/server/server.go
index e47ee266..7a257c48 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -138,6 +138,7 @@ func StartUp(Version string) {
 	version = Version
 	startTimeStamp = time.Now().Unix()
 	createPaths()
+	conf.SetupEnv()
 	conf.InitConf()
 
 	serverCtx, serverCancel := context.WithCancel(context.Background())
@@ -190,6 +191,7 @@ func StartUp(Version string) {
 	if err != nil {
 		panic(err)
 	}
+	conf.SetupConnectionProps()
 	connection.InitConnectionManager()
 	// reload with 3s timeout
 	if err := connection.ReloadConnection(3 * time.Second); err != nil {
diff --git a/internal/topo/node/compress_op.go b/internal/topo/node/compress_op.go
index 8b98e63c..c60367ac 100644
--- a/internal/topo/node/compress_op.go
+++ b/internal/topo/node/compress_op.go
@@ -60,11 +60,12 @@ func (o *CompressOp) Worker(_ api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
-	case []byte:
-		if r, err := o.tool.Compress(d); err != nil {
+	case api.RawTuple:
+		if r, err := o.tool.Compress(d.Raw()); err != nil {
 			return []any{err}
 		} else {
-			return []any{r}
+			d.Replace(r)
+			return []any{d}
 		}
 	default:
 		return []any{fmt.Errorf("unsupported data received: %v", d)}
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/node/window_op.go b/internal/topo/node/window_op.go
index 0eb3ee1e..eadbd1df 100644
--- a/internal/topo/node/window_op.go
+++ b/internal/topo/node/window_op.go
@@ -385,12 +385,12 @@ func (o *WindowOperator) execProcessingWindow(ctx api.StreamContext, inputs []*x
 							log.Debugf("Sent: %v", tsets)
 							o.Broadcast(tsets)
 							o.statManager.IncTotalRecordsOut()
-							o.statManager.IncTotalMessagesProcessed(int64(tsets.Len()))
 						}
 						inputs = tl.getRestTuples()
 					}
 				}
 				o.statManager.ProcessTimeEnd()
+				o.statManager.IncTotalMessagesProcessed(1)
 				o.statManager.SetBufferLength(int64(len(o.input)))
 				_ = ctx.PutState(WindowInputsKey, inputs)
 				_ = ctx.PutState(MsgCountKey, o.msgCount)
@@ -622,7 +622,6 @@ func (o *WindowOperator) scan(inputs []*xsql.Tuple, triggerTime time.Time, ctx a
 	log.Debugf("Sent: %v", results)
 	o.Broadcast(results)
 	o.statManager.IncTotalRecordsOut()
-	o.statManager.IncTotalMessagesProcessed(int64(results.Len()))
 
 	o.triggerTime = triggerTime
 	log.Debugf("new trigger time %d", o.triggerTime.UnixMilli())
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index d31388b4..8ac98bdf 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -225,8 +225,13 @@ func checkFeatures(ss api.Source, sp *SourcePropsForSplit, props map[string]any)
 	if sp.Merger != "" && sp.MergeField != "" {
 		return traits{}, fmt.Errorf("mergeField and merger cannot set together")
 	}
-	if sp.Merger != "" && (sp.Format == "" || sp.PayloadFormat == "") {
-		return traits{}, fmt.Errorf("merger must work with both format and payloadFormat")
+	// Let merger payload format defaults to format
+	if sp.Merger != "" && sp.PayloadFormat == "" {
+		sp.PayloadFormat = sp.Format
+		if sp.PayloadFormat == "" {
+			sp.PayloadFormat = "json"
+		}
+		props["payloadFormat"] = sp.PayloadFormat
 	}
 
 	info := checkByteSource(ss)
@@ -271,6 +276,10 @@ func checkFeatures(ss api.Source, sp *SourcePropsForSplit, props map[string]any)
 		r.needDecode = false
 		props["payloadBatchField"] = "frames"
 		props["payloadField"] = "data"
+		if _, ok := props["payloadSchemaId"]; !ok {
+			props["payloadSchemaId"] = props["schemaId"]
+			props["payloadDelimiter"] = props["delimiter"]
+		}
 	}
 	return r, nil
 }
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:
diff --git a/pkg/nng/sock.go b/pkg/nng/sock.go
index 3c311d98..b99a8905 100644
--- a/pkg/nng/sock.go
+++ b/pkg/nng/sock.go
@@ -94,6 +94,7 @@ func CreateConnection(ctx api.StreamContext, props map[string]any) (modules.Conn
 		mangos.OptionDialAsynch:       true, // will not report error and keep connecting
 		mangos.OptionMaxReconnectTime: 5 * time.Second,
 		mangos.OptionReconnectTime:    100 * time.Millisecond,
+		mangos.OptionMaxRecvSize:      0,
 	}); err != nil {
 		return nil, fmt.Errorf("please make sure nng server side has started and configured, can't dial: %s", err.Error())
 	}


Test patch:
diff --git a/internal/conf/env_manager_test.go b/internal/conf/env_manager_test.go
new file mode 100644
index 00000000..d3cd0399
--- /dev/null
+++ b/internal/conf/env_manager_test.go
@@ -0,0 +1,38 @@
+// 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 conf
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLoadEnv(t *testing.T) {
+	IsTesting = true
+	em := &EnvManager{}
+	require.NoError(t, os.Setenv("CONNECTION__MQTT__ENVDEMO__SERVER", "tcp://127.0.0.1:1883"))
+	require.NoError(t, os.Setenv("CONNECTION__MQTT__ENVDEMO__TOPIC", "mock"))
+	em.Setup()
+	em.SetupConnectionProps()
+	g, err := GetCfgFromKVStorage("connections", "mqtt", "envdemo")
+	require.NoError(t, err)
+	require.Equal(t, map[string]interface{}{
+		"server": "tcp://127.0.0.1:1883",
+		"topic":  "mock",
+	}, g["connections.mqtt.envdemo"])
+	require.Nil(t, em.connectionProps)
+}
diff --git a/internal/conf/load_test.go b/internal/conf/load_test.go
index 1ee03990..cdd2ba33 100644
--- a/internal/conf/load_test.go
+++ b/internal/conf/load_test.go
@@ -29,7 +29,7 @@ func TestEnv(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-
+	SetupEnv()
 	c := KuiperConf{}
 	err = LoadConfig(&c)
 	if err != nil {
@@ -49,7 +49,7 @@ func TestJsonCamelCase(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-
+	SetupEnv()
 	const ConfigName = "sources/httppull.yaml"
 	c := make(map[string]interface{})
 	err = LoadConfigByName(ConfigName, &c)
@@ -74,7 +74,7 @@ func TestNestedFields(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-
+	SetupEnv()
 	const ConfigName = "sources/edgex.yaml"
 	c := make(map[string]interface{})
 	err = LoadConfigByName(ConfigName, &c)
diff --git a/internal/conf/redis_store_config_test.go b/internal/conf/redis_store_config_test.go
index a66d7e85..6f412b14 100644
--- a/internal/conf/redis_store_config_test.go
+++ b/internal/conf/redis_store_config_test.go
@@ -63,7 +63,7 @@ func TestRedisStorageConSelector(t *testing.T) {
 			t.Error(err)
 		}
 	}
-
+	SetupEnv()
 	InitConf()
 
 	if Config.Store.Type != "redis" {
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/compress_op_test.go b/internal/topo/node/compress_op_test.go
index 5c637875..ca29d5ba 100644
--- a/internal/topo/node/compress_op_test.go
+++ b/internal/topo/node/compress_op_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
@@ -44,12 +45,12 @@ func TestCompressOp_Exec(t *testing.T) {
 	op.Exec(ctx, errCh)
 
 	cases := []any{
-		[]byte("{\"a\":1,\"b\":2}"),
+		&xsql.RawTuple{Rawdata: []byte("{\"a\":1,\"b\":2}")},
 		errors.New("go through error"),
 		"invalid",
 	}
 	expects := [][]any{
-		{[]byte("mock compress")},
+		{&xsql.RawTuple{Rawdata: []byte("mock compress")}},
 		{errors.New("go through error")},
 		{errors.New("unsupported data received: invalid")},
 	}
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/planner/planner_source_test.go b/internal/topo/planner/planner_source_test.go
index b5b6efc1..e1274227 100644
--- a/internal/topo/planner/planner_source_test.go
+++ b/internal/topo/planner/planner_source_test.go
@@ -94,7 +94,6 @@ func TestPlanTopo(t *testing.T) {
 				"connectionSelector": "mqtt.localConnection",
 				"interval":           "1s",
 				"merger":             "mock",
-				"payloadFormat":      "json",
 			},
 			p: "mqtt",
 			k: "testSelMerger",
@@ -491,11 +490,6 @@ func TestPlanError(t *testing.T) {
 			sql:  `SELECT * FROM src1`,
 			e:    "mergeField and merger cannot set together",
 		},
-		{
-			name: "merger need decoder",
-			sql:  `SELECT * FROM src2`,
-			e:    "merger must work with both format and payloadFormat",
-		},
 		{
 			name: "merger need rate limit",
 			sql:  `SELECT * FROM src3`,
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