Repository: lf-edge/ekuiper
Pull request #2902 (2024-06-04 10:15:00)

feat(op): rate limit customized merger

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2902?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 `69.53125%` with `39 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@1bf084c`). [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/2902?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/rate\_limit.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2902?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Frate_limit.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL3JhdGVfbGltaXQuZ28=) | 63.10% | [27 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2902?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/planner/planner\_source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2902?src=pr&el=tree&filepath=internal%2Ftopo%2Fplanner%2Fplanner_source.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9wbGFubmVyL3BsYW5uZXJfc291cmNlLmdv) | 65.22% | [4 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2902?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    #2902   +/-   ##
============================================
  Coverage             ?   67.77%           
============================================
  Files                ?      293           
  Lines                ?    28231           
  Branches             ?        0           
============================================
  Hits                 ?    19133           
  Misses               ?     7388           
  Partials             ?     1710           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2902?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/binder/io/builtin.go b/internal/binder/io/builtin.go
index c021f983..fccf4f78 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -19,6 +19,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/io/file"
 	"github.com/lf-edge/ekuiper/v2/internal/io/memory"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt"
+	mqttCon "github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/io/sink"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
@@ -45,6 +46,8 @@ func init() {
 
 	// modules.RegisterLookupSource("memory", func() api.LookupSource { return memory.GetLookupSource() })
 	// modules.RegisterLookupSource("httppull", func() api.LookupSource { return http.GetLookUpSource() })
+
+	modules.RegisterConnection("mqtt", mqttCon.CreateConnection)
 }
 
 type Manager struct{}
diff --git a/internal/io/connection/connection_pool.go b/internal/io/connection/connection_pool.go
index a2ff1ce3..3da1c89d 100644
--- a/internal/io/connection/connection_pool.go
+++ b/internal/io/connection/connection_pool.go
@@ -200,16 +200,21 @@ var globalConnectionManager *ConnectionManager
 
 func InitConnectionManager4Test() error {
 	InitMockTest()
-	return InitConnectionManager()
+	InitConnectionManager()
+	return nil
 }
 
-func InitConnectionManager() error {
+func InitConnectionManager() {
 	globalConnectionManager = &ConnectionManager{
 		connectionPool: make(map[string]ConnectionMeta),
 	}
 	if conf.IsTesting {
-		return nil
+		return
 	}
+	DefaultBackoffMaxElapsedDuration = time.Duration(conf.Config.Connection.BackoffMaxElapsedDuration)
+}
+
+func ReloadConnection() error {
 	cfgs, err := conf.GetCfgFromKVStorage("connections", "", "")
 	if err != nil {
 		return err
@@ -228,12 +233,12 @@ func InitConnectionManager() error {
 		}
 		conn, err := createNamedConnection(context.Background(), meta)
 		if err != nil {
-			return fmt.Errorf("initialize connection:%v failed, err:%v", id, err)
+			conf.Log.Warnf("initialize connection:%v failed, err:%v", id, err)
+			continue
 		}
 		meta.conn = conn
 		globalConnectionManager.connectionPool[id] = meta
 	}
-	DefaultBackoffMaxElapsedDuration = time.Duration(conf.Config.Connection.BackoffMaxElapsedDuration)
 	return nil
 }
 
diff --git a/internal/io/mqtt/client/client.go b/internal/io/mqtt/client/client.go
index 92161fbf..fc006da1 100644
--- a/internal/io/mqtt/client/client.go
+++ b/internal/io/mqtt/client/client.go
@@ -32,10 +32,6 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
-func init() {
-	modules.ConnectionRegister["mqtt"] = CreateConnection
-}
-
 type Connection struct {
 	pahoMqtt.Client
 	selId     string
diff --git a/internal/pkg/filex/parser.go b/internal/pkg/filex/parser.go
index 4914508c..a3711c4e 100644
--- a/internal/pkg/filex/parser.go
+++ b/internal/pkg/filex/parser.go
@@ -16,6 +16,7 @@ package filex
 
 import (
 	"encoding/json"
+	"fmt"
 	"os"
 
 	"gopkg.in/yaml.v3"
@@ -41,7 +42,12 @@ func WriteYamlMarshal(path string, data interface{}) error {
 	return os.WriteFile(path, y, 0o666)
 }
 
-func ReadYamlUnmarshal(path string, ret interface{}) error {
+func ReadYamlUnmarshal(path string, ret interface{}) (err error) {
+	defer func() {
+		if err != nil {
+			err = fmt.Errorf("read %s unmarshal yaml failed, err:%v", path, err)
+		}
+	}()
 	sliByte, err := os.ReadFile(path)
 	if nil != err {
 		return err
diff --git a/internal/server/server.go b/internal/server/server.go
index cfaed7bd..b3930489 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -160,10 +160,6 @@ func StartUp(Version string) {
 	keyedstate.InitKeyedStateKV()
 
 	meta2.InitYamlConfigManager()
-	if err := connection.InitConnectionManager(); err != nil {
-		panic(err)
-	}
-
 	ruleProcessor = processor.NewRuleProcessor()
 	streamProcessor = processor.NewStreamProcessor()
 	rulesetProcessor = processor.NewRulesetProcessor(ruleProcessor, streamProcessor)
@@ -186,6 +182,10 @@ func StartUp(Version string) {
 	if err != nil {
 		panic(err)
 	}
+	connection.InitConnectionManager()
+	if err := connection.ReloadConnection(); err != nil {
+		conf.Log.Warn(err)
+	}
 	meta.Bind()
 	initRuleset()
 
diff --git a/internal/topo/node/rate_limit.go b/internal/topo/node/rate_limit.go
index 63e0d3e5..0c910dda 100644
--- a/internal/topo/node/rate_limit.go
+++ b/internal/topo/node/rate_limit.go
@@ -24,18 +24,22 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/converter"
+	schemaLayer "github.com/lf-edge/ekuiper/v2/internal/converter/schema"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 type limitConf struct {
-	Interval   time.Duration `json:"interval"`
-	MergeField string        `json:"mergeField"`
-	Format     string        `json:"format"`
+	Interval        time.Duration `json:"interval"`
+	MergeField      string        `json:"mergeField"`
+	Format          string        `json:"format"`
+	Merger          string        `json:"merger"`
+	PayloadSchemaId string        `json:"payloadSchemaId"`
 }
 
 // RateLimitOp handle messages at a regular rate, ignoring messages that arrive too quickly, only keep the most recent message. (default strategy)
@@ -47,17 +51,21 @@ type limitConf struct {
 type RateLimitOp struct {
 	*defaultSinkNode
 	// configs
-	c *limitConf
+	c             *limitConf
+	mergeStrategy int
 	// state
 	// keep last strategy
 	latest any
 	// merged items
 	frameSet map[any]map[string]any
-	// only when merge is set
+	// only when mergeField is set
 	decoder message.PartialDecoder
+	// only when using merger
+	merger modules.Merger
+	sLayer *schemaLayer.SchemaLayer
 }
 
-func NewRateLimitOp(ctx api.StreamContext, name string, rOpt *def.RuleOption, props map[string]any) (*RateLimitOp, error) {
+func NewRateLimitOp(ctx api.StreamContext, name string, rOpt *def.RuleOption, options *ast.Options, schema map[string]*ast.JsonStreamField, props map[string]any) (*RateLimitOp, error) {
 	c := &limitConf{}
 	err := cast.MapToStruct(props, c)
 	if err != nil {
@@ -77,6 +85,7 @@ func NewRateLimitOp(ctx api.StreamContext, name string, rOpt *def.RuleOption, pr
 		}
 		cv, err := converter.GetOrCreateConverter(ctx, &ast.Options{
 			FORMAT:       c.Format,
+			SCHEMAID:     options.SCHEMAID,
 			Schema:       nil,
 			IsSchemaLess: true,
 			IsWildCard:   true,
@@ -90,19 +99,66 @@ func NewRateLimitOp(ctx api.StreamContext, name string, rOpt *def.RuleOption, pr
 		} else {
 			return nil, fmt.Errorf("format %s does not support partial decode", c.Format)
 		}
+		o.mergeStrategy = 1
+	} else if c.Merger != "" {
+		if mp, ok := modules.Mergers[c.Merger]; ok {
+			cm, err := mp(ctx, c.PayloadSchemaId, schema)
+			if err != nil {
+				return nil, fmt.Errorf("fail to initiate merge %s: %v", c.Merger, err)
+			}
+			o.merger = cm
+		} else {
+			return nil, fmt.Errorf("merger %s not found", c.Merger)
+		}
+		o.mergeStrategy = 2
+		o.sLayer = schemaLayer.NewSchemaLayer(ctx.GetRuleId(), name, schema, schema != nil)
 	}
 	return o, nil
 }
 
+// Exec ratelimit op deal with 3 merge strategy
+// - latest
+// - merge by mergeField (when format and mergeField is set and no payload format)
+// - merge by merger (when format, payloadFormat and merger is set)
 func (o *RateLimitOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	o.prepareExec(ctx, errCh, "op")
 	ticker := timex.GetTicker(o.c.Interval)
+	ctx.GetLogger().Infof("rate limit run with interval %v", o.c.Interval)
 	go func() {
 		defer func() {
 			ticker.Stop()
 			o.Close()
 		}()
-		if o.c.MergeField != "" { // do merge
+		switch o.mergeStrategy {
+		case 0: // get the latest
+			ctx.GetLogger().Infof("rate limit run with merge strategy 0: latest")
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case d := <-o.input:
+					dd, processed := o.commonIngest(ctx, d)
+					if processed {
+						continue
+					}
+					o.statManager.IncTotalRecordsIn()
+					o.statManager.ProcessTimeStart()
+					o.latest = dd
+					o.statManager.ProcessTimeEnd()
+					o.statManager.IncTotalMessagesProcessed(1)
+					o.statManager.SetBufferLength(int64(len(o.input)))
+				case t := <-ticker.C:
+					if o.latest != nil {
+						o.Broadcast(o.latest)
+						o.latest = nil
+						o.statManager.IncTotalRecordsOut()
+					} else {
+						ctx.GetLogger().Debugf("ratelimit had nothing to sent at %d", t.UnixMilli())
+					}
+				}
+			}
+		case 1: // by mergeField
+			ctx.GetLogger().Infof("rate limit run with merge strategy 1: by field")
 			for {
 				select {
 				case <-ctx.Done():
@@ -173,7 +229,8 @@ func (o *RateLimitOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 					}
 				}
 			}
-		} else {
+		case 2:
+			ctx.GetLogger().Infof("rate limit run with merge strategy 2: by merger")
 			for {
 				select {
 				case <-ctx.Done():
@@ -185,13 +242,35 @@ func (o *RateLimitOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 					}
 					o.statManager.IncTotalRecordsIn()
 					o.statManager.ProcessTimeStart()
-					o.latest = dd
+					var err error
+					switch dt := dd.(type) {
+					case *xsql.RawTuple:
+						err = o.merger.Merging(ctx, dt.Raw())
+						if err == nil {
+							o.latest = dt
+						}
+					default:
+						err = fmt.Errorf("rate limit merge only supports raw but got %v", d)
+					}
+					if err != nil {
+						o.statManager.IncTotalExceptions(err.Error())
+						o.Broadcast(err)
+					}
 					o.statManager.ProcessTimeEnd()
 					o.statManager.IncTotalMessagesProcessed(1)
 					o.statManager.SetBufferLength(int64(len(o.input)))
 				case t := <-ticker.C:
-					if o.latest != nil {
-						o.Broadcast(o.latest)
+					frames, ok := o.merger.Trigger(ctx)
+					if ok {
+						rt := o.latest.(*xsql.RawTuple)
+						o.Broadcast(&xsql.Tuple{
+							Emitter:   rt.Emitter,
+							Timestamp: rt.Timestamp,
+							Metadata:  rt.Metadata,
+							Message: map[string]any{
+								"frames": frames,
+							},
+						})
 						o.latest = nil
 						o.statManager.IncTotalRecordsOut()
 					} else {
@@ -202,3 +281,34 @@ func (o *RateLimitOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 		}
 	}()
 }
+
+func (o *RateLimitOp) AttachSchema(ctx api.StreamContext, dataSource string, schema map[string]*ast.JsonStreamField, isWildcard bool) {
+	if o.mergeStrategy != 2 {
+		return
+	}
+	if fastDecoder, ok := o.merger.(message.SchemaResetAbleConverter); ok {
+		ctx.GetLogger().Infof("attach schema to shared stream")
+		// append payload field to schema
+		if err := o.sLayer.MergeSchema(ctx.GetRuleId(), dataSource, schema, isWildcard); err != nil {
+			ctx.GetLogger().Warnf("merge schema to shared stream failed, err: %v", err)
+		} else {
+			ctx.GetLogger().Infof("attach schema become %+v", o.sLayer.GetSchema())
+			fastDecoder.ResetSchema(o.sLayer.GetSchema())
+		}
+	}
+}
+
+func (o *RateLimitOp) DetachSchema(ctx api.StreamContext, ruleId string) {
+	if o.mergeStrategy != 2 {
+		return
+	}
+	if fastDecoder, ok := o.merger.(message.SchemaResetAbleConverter); ok {
+		ctx.GetLogger().Infof("detach schema for shared stream rule %v", ruleId)
+		if err := o.sLayer.DetachSchema(ruleId); err != nil {
+			ctx.GetLogger().Infof("detach schema for shared stream rule %v failed, err:%v", ruleId, err)
+		} else {
+			fastDecoder.ResetSchema(o.sLayer.GetSchema())
+			ctx.GetLogger().Infof("detach schema become %+v", o.sLayer.GetSchema())
+		}
+	}
+}
diff --git a/internal/topo/planner/dataSourcePlan.go b/internal/topo/planner/dataSourcePlan.go
index 8aa039c8..cb46e05a 100644
--- a/internal/topo/planner/dataSourcePlan.go
+++ b/internal/topo/planner/dataSourcePlan.go
@@ -377,6 +377,9 @@ func (p *DataSourcePlan) getAllFields() {
 	if !p.isWildCard {
 		p.streamFields = p.fields
 	} else {
+		if len(p.fields) > 0 && p.streamFields == nil {
+			p.streamFields = make(map[string]*ast.JsonStreamField, len(p.fields))
+		}
 		for name, fr := range p.fields {
 			p.streamFields[name] = fr
 		}
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index b96196f7..4b09b79a 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -104,13 +104,13 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 	var ops []node.OperatorNode
 
 	// Need to check after source has provisioned, so do not put it before provision
-	featureSet, err := checkFeatures(ss, sp, props)
+	featureSet, err := checkFeatures(ss, sp, props, t.streamStmt.Options)
 	if err != nil {
 		return nil, nil, 0, err
 	}
 
 	if featureSet.needRatelimit {
-		rlOp, err := node.NewRateLimitOp(ctx, fmt.Sprintf("%d_ratelimit", index), options, props)
+		rlOp, err := node.NewRateLimitOp(ctx, fmt.Sprintf("%d_ratelimit", index), options, t.streamStmt.Options, t.streamFields, props)
 		if err != nil {
 			return nil, nil, 0, err
 		}
@@ -137,6 +137,15 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 		ops = append(ops, decodeNode)
 	}
 
+	if featureSet.needRatelimitMerge {
+		rlOp, err := node.NewRateLimitOp(ctx, fmt.Sprintf("%d_ratelimit", index), options, t.streamStmt.Options, t.streamFields, props)
+		if err != nil {
+			return nil, nil, 0, err
+		}
+		index++
+		ops = append(ops, rlOp)
+	}
+
 	if featureSet.needPayloadDecode {
 		// Create the decode node
 		payloadDecodeNode, err := node.NewDecodeOp(ctx, true, fmt.Sprintf("%d_payload_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
@@ -176,8 +185,10 @@ type SourcePropsForSplit struct {
 	SelId         string        `json:"connectionSelector"`
 	PayloadFormat string        `json:"payloadFormat"`
 	Interval      time.Duration `json:"interval"`
-	MergeField    string        `json:"mergeField"`
-	Format        string        `json:"format"`
+	// merger and mergerField should only set one
+	MergeField string `json:"mergeField"`
+	Merger     string `json:"merger"`
+	Format     string `json:"format"`
 }
 
 type traits struct {
@@ -185,11 +196,22 @@ type traits struct {
 	needCompression   bool
 	needDecode        bool
 	needPayloadDecode bool
-	needRatelimit     bool
+	// rate limit will plan right after source read
+	needRatelimit bool
+	// rate limit merge will plan after decompress
+	needRatelimitMerge bool
 }
 
 // function to return if a sub node is needed
-func checkFeatures(ss api.Source, sp *SourcePropsForSplit, props map[string]any) (traits, error) {
+func checkFeatures(ss api.Source, sp *SourcePropsForSplit, props map[string]any, options *ast.Options) (traits, error) {
+	// validate merger
+	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")
+	}
+
 	info := checkByteSource(ss)
 	// TODO here is a hack for file source send interval. If it is sent in file, do not need to process sendInterval in decode
 	if info.HasInterval {
@@ -217,6 +239,19 @@ func checkFeatures(ss api.Source, sp *SourcePropsForSplit, props map[string]any)
 		r.needPayloadDecode = true
 		r.needDecode = false
 		props["payloadFormat"] = props["format"]
+		props["payloadSchemaId"] = options.SCHEMAID
+		props["payloadDelimiter"] = options.DELIMITER
+		props["payloadBatchField"] = "frames"
+		props["payloadField"] = "data"
+	}
+	if !r.needRatelimit && sp.Merger != "" {
+		return r, fmt.Errorf("merger is set but rate limit is not required")
+	}
+	// If rate limit merger is set, the first level decode will be done by merger
+	if r.needRatelimit && sp.Merger != "" {
+		r.needRatelimitMerge = true
+		r.needRatelimit = false
+		r.needDecode = false
 		props["payloadBatchField"] = "frames"
 		props["payloadField"] = "data"
 	}
diff --git a/pkg/ast/sourceStmt.go b/pkg/ast/sourceStmt.go
index 4baaa6f4..ca756bd1 100644
--- a/pkg/ast/sourceStmt.go
+++ b/pkg/ast/sourceStmt.go
@@ -92,11 +92,14 @@ func (sf *StreamFields) ToJsonSchema() map[string]*JsonStreamField {
 }
 
 func convertSchema(sfs StreamFields) map[string]*JsonStreamField {
-	result := make(map[string]*JsonStreamField, len(sfs))
-	for _, sf := range sfs {
-		result[sf.Name] = convertFieldType(sf.FieldType)
+	if len(sfs) > 0 {
+		result := make(map[string]*JsonStreamField, len(sfs))
+		for _, sf := range sfs {
+			result[sf.Name] = convertFieldType(sf.FieldType)
+		}
+		return result
 	}
-	return result
+	return nil
 }
 
 func convertFieldType(sf FieldType) *JsonStreamField {
diff --git a/pkg/message/mock.go b/pkg/message/mock.go
index f9f57441..477fbf14 100644
--- a/pkg/message/mock.go
+++ b/pkg/message/mock.go
@@ -14,7 +14,10 @@
 
 package message
 
-import "github.com/lf-edge/ekuiper/contract/v2/api"
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
 
 type MockPartialConverter struct {
 	i int
@@ -35,3 +38,27 @@ func (m *MockPartialConverter) DecodeField(ctx api.StreamContext, b []byte, f st
 	m.i++
 	return r, nil
 }
+
+type MockMerger struct {
+	count  int
+	frames []map[string]any
+}
+
+func (m *MockMerger) Merging(ctx api.StreamContext, b []byte) error {
+	if m.frames == nil {
+		m.frames = make([]map[string]any, 2)
+	}
+	m.frames[m.count%2] = map[string]any{"data": b}
+	m.count++
+	return nil
+}
+
+func (m *MockMerger) Trigger(ctx api.StreamContext) ([]any, bool) {
+	result := make([]any, len(m.frames))
+	for i, frame := range m.frames {
+		result[i] = frame
+	}
+	return result, true
+}
+
+func (m *MockMerger) ResetSchema(schema map[string]*ast.JsonStreamField) {}
diff --git a/pkg/modules/connection.go b/pkg/modules/connection.go
index feab1007..e693777b 100644
--- a/pkg/modules/connection.go
+++ b/pkg/modules/connection.go
@@ -25,10 +25,14 @@ type Connection interface {
 	Ref(ctx api.StreamContext) int
 }
 
-type RegisterConnection func(ctx api.StreamContext, id string, props map[string]any) (Connection, error)
+type ConnectionProvider func(ctx api.StreamContext, id string, props map[string]any) (Connection, error)
 
-var ConnectionRegister map[string]RegisterConnection
+var ConnectionRegister map[string]ConnectionProvider
 
 func init() {
-	ConnectionRegister = map[string]RegisterConnection{}
+	ConnectionRegister = map[string]ConnectionProvider{}
+}
+
+func RegisterConnection(name string, cp ConnectionProvider) {
+	ConnectionRegister[name] = cp
 }
diff --git a/pkg/modules/converter.go b/pkg/modules/converter.go
index 2f65ad55..22f47ff8 100644
--- a/pkg/modules/converter.go
+++ b/pkg/modules/converter.go
@@ -14,7 +14,11 @@
 
 package modules
 
-import "github.com/lf-edge/ekuiper/v2/pkg/message"
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+	"github.com/lf-edge/ekuiper/v2/pkg/message"
+)
 
 var Converters = map[string]message.ConverterProvider{}
 
@@ -27,3 +31,21 @@ func IsFormatSupported(format string) bool {
 	_, ok := Converters[format]
 	return ok
 }
+
+// Merger is used to merge multiple frames. It is currently called by rate limiter only
+type Merger interface {
+	// Merging is called when read in a new frame
+	Merging(ctx api.StreamContext, b []byte) error
+	// Trigger is called when rate limiter is trigger to send out a message
+	Trigger(ctx api.StreamContext) ([]any, bool)
+}
+
+type MergerProvider func(ctx api.StreamContext, payloadSchema string, logicalSchema map[string]*ast.JsonStreamField) (Merger, error)
+
+// Mergers list, the key is format + payload format such as "jsoncan"
+var Mergers = map[string]MergerProvider{}
+
+// RegisterMerger registers a merger with the format name and payload format name such as "jsoncan"
+func RegisterMerger(name string, provider MergerProvider) {
+	Mergers[name] = provider
+}


Test patch:
diff --git a/internal/io/mqtt/source_test.go b/internal/io/mqtt/source_test.go
index 95495b15..0fefb880 100644
--- a/internal/io/mqtt/source_test.go
+++ b/internal/io/mqtt/source_test.go
@@ -23,9 +23,11 @@ import (
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
+	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
 // NOTICE!!! Need to run a MQTT broker in localhost:1883 for this test or change the url to your broker
@@ -33,6 +35,7 @@ const url = "tcp://127.0.0.1:1883"
 
 func init() {
 	testx.InitEnv("mqtt_source_connector")
+	modules.RegisterConnection("mqtt", client.CreateConnection)
 }
 
 func TestProvision(t *testing.T) {
diff --git a/internal/pkg/filex/parser_test.go b/internal/pkg/filex/parser_test.go
new file mode 100644
index 00000000..c75a7981
--- /dev/null
+++ b/internal/pkg/filex/parser_test.go
@@ -0,0 +1,27 @@
+// 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 filex
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestReadYamlUnmarshalErr(t *testing.T) {
+	err := ReadYamlUnmarshal("1.yaml", map[string]interface{}{})
+	require.Error(t, err)
+	require.Equal(t, "read 1.yaml unmarshal yaml failed, err:open 1.yaml: no such file or directory", err.Error())
+}
diff --git a/internal/topo/node/rate_limit_test.go b/internal/topo/node/rate_limit_test.go
index 5e1ba99c..f05de6e1 100644
--- a/internal/topo/node/rate_limit_test.go
+++ b/internal/topo/node/rate_limit_test.go
@@ -15,6 +15,7 @@
 package node
 
 import (
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -29,24 +30,35 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 func TestNewRateLimit(t *testing.T) {
 	ctx := mockContext.NewMockContext("test1", "new_test")
-	_, err := NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{"interval": "1s"})
+	rl, err := NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{"interval": "1s"})
 	assert.NoError(t, err)
-	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{"interval": "1ns"})
+	assert.Equal(t, 0, rl.mergeStrategy)
+	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{"interval": "1ns"})
 	assert.Error(t, err)
 	assert.EqualError(t, err, "interval should be larger than 1ms")
-	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{"interval": "1s", "mergeField": "id"})
+	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{"interval": "1s", "mergeField": "id"})
 	assert.Error(t, err)
 	assert.EqualError(t, err, "rate limit merge must define format")
-	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{"interval": "1s", "mergeField": "id", "format": "none"})
+	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{"interval": "1s", "mergeField": "id", "format": "none"})
 	assert.Error(t, err)
 	assert.EqualError(t, err, "format type none not supported")
-	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{"interval": "1s", "mergeField": "id", "format": "delimited"})
+	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{"interval": "1s", "mergeField": "id", "format": "delimited"})
 	assert.Error(t, err)
 	assert.EqualError(t, err, "format delimited does not support partial decode")
+	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{"interval": "1s", "merger": "none", "format": "delimited"})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "merger none not found")
+	modules.RegisterMerger("none", func(ctx api.StreamContext, schemaId string, logicalSchema map[string]*ast.JsonStreamField) (modules.Merger, error) {
+		return nil, errors.New("mock error")
+	})
+	_, err = NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{"interval": "1s", "merger": "none", "format": "delimited"})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "fail to initiate merge none: mock error")
 }
 
 func TestRateLimit(t *testing.T) {
@@ -61,23 +73,26 @@ func TestRateLimit(t *testing.T) {
 			sendCount: 10,
 			interval:  time.Second,
 			expectItems: []any{
+				&xsql.WatermarkTuple{},
+				&xsql.WatermarkTuple{},
 				&xsql.RawTuple{
 					Rawdata: []byte{3},
 				},
+				&xsql.WatermarkTuple{},
 				&xsql.RawTuple{
 					Rawdata: []byte{6},
 				},
+				&xsql.WatermarkTuple{},
 				&xsql.RawTuple{
 					Rawdata: []byte{9},
 				},
 			},
 		},
 	}
-	mc := mockclock.GetMockClock()
 	for i, tc := range testcases {
 		t.Run(fmt.Sprintf("testcase %d", i), func(t *testing.T) {
 			ctx, cancel := mockContext.NewMockContext("test1", "batch_test").WithCancel()
-			op, err := NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{
+			op, err := NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{
 				"interval": "1s",
 			})
 			assert.NoError(t, err)
@@ -85,14 +100,22 @@ func TestRateLimit(t *testing.T) {
 			err = op.AddOutput(out, "test")
 			assert.NoError(t, err)
 
+			timex.Set(100)
 			errCh := make(chan error)
 			op.Exec(ctx, errCh)
 			for i := 0; i < tc.sendCount; i++ {
 				op.input <- &xsql.RawTuple{
 					Rawdata: []byte{uint8(i)},
 				}
-				mc.Add(300 * time.Millisecond)
+				fmt.Printf("send input at %d\n", timex.GetNowInMilli())
+				if i%3 == 0 {
+					op.input <- &xsql.WatermarkTuple{}
+					fmt.Printf("send watermark at %d\n", timex.GetNowInMilli())
+				}
+				time.Sleep(10 * time.Millisecond)
+				timex.Add(300 * time.Millisecond)
 			}
+			timex.Add(2 * time.Second)
 			cancel()
 			// make sure op has done all sending
 			for {
@@ -170,12 +193,13 @@ func TestRateLimitMerge(t *testing.T) {
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx, cancel := mockContext.NewMockContext("test1", "batch_test").WithCancel()
-			op, err := NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{
+			op, err := NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{
 				"interval":   "1s",
 				"mergeField": "id",
 				"format":     "mockp",
 			})
 			assert.NoError(t, err)
+			assert.Equal(t, 1, op.mergeStrategy)
 			out := make(chan any, 10)
 			err = op.AddOutput(out, "test")
 			assert.NoError(t, err)
@@ -206,3 +230,106 @@ func TestRateLimitMerge(t *testing.T) {
 		})
 	}
 }
+
+func TestRateLimitCustomMerge(t *testing.T) {
+	testcases := []struct {
+		name        string
+		sendCount   int
+		interval    time.Duration
+		expectItems []any
+	}{
+		{ // sending gap is 300ms
+			name:      "normal",
+			sendCount: 10,
+			interval:  time.Second,
+			expectItems: []any{
+				errors.New("rate limit merge only supports raw but got &{ map[test:1] 0001-01-01 00:00:00 +0000 UTC map[] map[] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}"),
+				&xsql.Tuple{
+					Message: map[string]any{
+						"frames": []any{
+							map[string]any{
+								"data": []byte(`{"id":0, "value":2}`),
+							},
+							map[string]any{
+								"data": []byte(`{"id":1, "value":3}`),
+							},
+						},
+					},
+				},
+				&xsql.Tuple{
+					Message: map[string]any{
+						"frames": []any{
+							map[string]any{
+								"data": []byte(`{"id":0, "value":6}`),
+							},
+							map[string]any{
+								"data": []byte(`{"id":1, "value":5}`),
+							},
+						},
+					},
+				},
+				&xsql.Tuple{
+					Message: map[string]any{
+						"frames": []any{
+							map[string]any{
+								"data": []byte(`{"id":0, "value":8}`),
+							},
+							map[string]any{
+								"data": []byte(`{"id":1, "value":9}`),
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	modules.RegisterMerger("mock", func(ctx api.StreamContext, schemaId string, logicalSchema map[string]*ast.JsonStreamField) (modules.Merger, error) {
+		return &message.MockMerger{}, nil
+	})
+	mc := mockclock.GetMockClock()
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, cancel := mockContext.NewMockContext("test1", "batch_test").WithCancel()
+			op, err := NewRateLimitOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, nil, map[string]any{
+				"interval":      "1s",
+				"merger":        "mock",
+				"format":        "json",
+				"payloadFormat": "json",
+			})
+			assert.NoError(t, err)
+			assert.Equal(t, 2, op.mergeStrategy)
+			out := make(chan any, 10)
+			err = op.AddOutput(out, "test")
+			assert.NoError(t, err)
+
+			errCh := make(chan error)
+			op.Exec(ctx, errCh)
+			op.input <- &xsql.Tuple{
+				Message: map[string]any{"test": 1},
+			}
+			op.AttachSchema(ctx, "source1", map[string]*ast.JsonStreamField{}, true)
+			for i := 0; i < tc.sendCount; i++ {
+				op.input <- &xsql.RawTuple{
+					Rawdata: []byte(fmt.Sprintf(`{"id":%d, "value":%d}`, i%2, i)),
+				}
+				mc.Add(300 * time.Millisecond)
+			}
+			cancel()
+			op.DetachSchema(ctx, "rule1")
+			// make sure op has done all sending
+			for {
+				processed := op.statManager.GetMetrics()[2]
+				if processed == int64(tc.sendCount)+1 {
+					close(out)
+					break
+				}
+				time.Sleep(10 * time.Millisecond)
+			}
+			r := make([]any, 0, len(tc.expectItems))
+			for ele := range out {
+				r = append(r, ele)
+			}
+			assert.Equal(t, tc.expectItems, r)
+		})
+	}
+}
diff --git a/internal/topo/planner/planner_source_test.go b/internal/topo/planner/planner_source_test.go
index 04177425..4d5c8b8a 100644
--- a/internal/topo/planner/planner_source_test.go
+++ b/internal/topo/planner/planner_source_test.go
@@ -42,11 +42,15 @@ func TestPlanTopo(t *testing.T) {
 	modules.RegisterConverter("mockp", func(ctx api.StreamContext, schemaFileName string, SchemaMessageName string, delimiter string, logicalSchema map[string]*ast.JsonStreamField) (message.Converter, error) {
 		return &message.MockPartialConverter{}, nil
 	})
+	modules.RegisterMerger("mock", func(ctx api.StreamContext, schemaId string, logicalSchema map[string]*ast.JsonStreamField) (modules.Merger, error) {
+		return &message.MockMerger{}, nil
+	})
 	streamSqls := map[string]string{
 		"src1":     `CREATE STREAM src1 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt");`,
 		"src2":     `CREATE STREAM src2 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt", SHARED="true");`,
 		"src3":     `CREATE STREAM src3 () WITH (DATASOURCE="topic1", FORMAT="mockp", TYPE="mqtt", CONF_KEY="testSelMock");`,
 		"src4":     `CREATE STREAM src4 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSel",SHARED="true");`,
+		"src5":     `CREATE STREAM src3 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSelMerger");`,
 		"filesrc1": `CREATE STREAM fs1 () WITH (FORMAT="json", TYPE="file",CONF_KEY="lines");`,
 		"filesrc2": `CREATE STREAM fs2 () WITH (FORMAT="delimited", TYPE="file",CONF_KEY="csv");`,
 		"filesrc3": `CREATE STREAM fs3 () WITH (FORMAT="json",TYPE="file",CONF_KEY="json");`,
@@ -82,6 +86,16 @@ func TestPlanTopo(t *testing.T) {
 			p: "mqtt",
 			k: "testSelMock",
 		},
+		{
+			conf: map[string]any{
+				"connectionSelector": "mqtt.localConnection",
+				"interval":           "1s",
+				"merger":             "mock",
+				"payloadFormat":      "json",
+			},
+			p: "mqtt",
+			k: "testSelMerger",
+		},
 		{
 			conf: map[string]any{
 				"fileType":      "csv",
@@ -309,6 +323,33 @@ func TestPlanTopo(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "test mqtt merger",
+			sql:  `SELECT * FROM src5`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_mqtt.localConnection/topic1"},
+				Edges: map[string][]any{
+					"source_mqtt.localConnection/topic1": {
+						"op_2_ratelimit",
+					},
+					"op_2_ratelimit": {
+						"op_3_payload_decoder",
+					},
+					"op_3_payload_decoder": {
+						"op_4_project",
+					},
+					"op_4_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -318,3 +359,108 @@ func TestPlanTopo(t *testing.T) {
 		})
 	}
 }
+
+func TestPlanError(t *testing.T) {
+	kv, err := store.GetKV("stream")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	modules.RegisterConverter("mockp", func(ctx api.StreamContext, schemaFileName string, SchemaMessageName string, delimiter string, logicalSchema map[string]*ast.JsonStreamField) (message.Converter, error) {
+		return &message.MockPartialConverter{}, nil
+	})
+	modules.RegisterMerger("mock", func(ctx api.StreamContext, schemaId string, logicalSchema map[string]*ast.JsonStreamField) (modules.Merger, error) {
+		return &message.MockMerger{}, nil
+	})
+	streamSqls := map[string]string{
+		"src1": `CREATE STREAM src1 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt", CONF_KEY="invalidMerge");`,
+		"src2": `CREATE STREAM src2 () WITH (DATASOURCE="src1", FORMAT="", TYPE="mqtt", CONF_KEY="invalidMerger");`,
+		"src3": `CREATE STREAM src2 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt", CONF_KEY="invalidMerger2");`,
+	}
+	for name, sql := range streamSqls {
+		s, err := json.Marshal(&xsql.StreamInfo{
+			StreamType: ast.TypeStream,
+			Statement:  sql,
+		})
+		assert.NoError(t, err)
+		err = kv.Set(name, string(s))
+		assert.NoError(t, err)
+	}
+	confs := []struct {
+		conf map[string]any
+		p    string
+		k    string
+	}{
+		{
+			conf: map[string]any{
+				"interval":   "1s",
+				"mergeField": "id",
+				"merger":     "mock",
+			},
+			p: "mqtt",
+			k: "invalidMerge",
+		},
+		{
+			conf: map[string]any{
+				"interval": "1s",
+				"merger":   "mock",
+			},
+			p: "mqtt",
+			k: "invalidMerger",
+		},
+		{
+			conf: map[string]any{
+				"merger":        "mock",
+				"payloadFormat": "mockp",
+			},
+			p: "mqtt",
+			k: "invalidMerger2",
+		},
+	}
+	meta.InitYamlConfigManager()
+	dataDir, _ := conf.GetDataLoc()
+	err = os.MkdirAll(filepath.Join(dataDir, "sources"), 0o755)
+	assert.NoError(t, err)
+
+	for _, cc := range confs {
+		p, k := cc.p, cc.k
+		bs, err := json.Marshal(cc.conf)
+		assert.NoError(t, err)
+		err = meta.AddSourceConfKey(p, k, "", bs)
+		assert.NoError(t, err)
+		// intended to run at last
+		defer func() {
+			err = meta.DelSourceConfKey(p, k, "")
+			assert.NoError(t, err)
+		}()
+	}
+
+	tests := []struct {
+		name string
+		sql  string
+		e    string
+	}{
+		{
+			name: "mergeField and merger mutual exclusive",
+			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`,
+			e:    "merger is set but rate limit is not required",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := PlanSQLWithSourcesAndSinks(def.GetDefaultRule(tt.name, tt.sql), nil)
+			assert.Error(t, err)
+			assert.EqualError(t, err, tt.e)
+		})
+	}
+}
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index e5e55e5b..a9b163c6 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -3069,7 +3069,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 															DataSourcePlan{
 																name:         "src1",
 																isWildCard:   true,
-																streamFields: map[string]*ast.JsonStreamField{},
+																streamFields: nil,
 																streamStmt:   streams["src1"],
 																metaFields:   []string{},
 																isSchemaless: true,
@@ -3765,7 +3765,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 						DataSourcePlan{
 							baseLogicalPlan: baseLogicalPlan{},
 							name:            "src1",
-							streamFields:    map[string]*ast.JsonStreamField{},
+							streamFields:    nil,
 							streamStmt:      streams["src1"],
 							metaFields:      []string{"device"},
 							isWildCard:      true,
@@ -3808,7 +3808,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 						DataSourcePlan{
 							baseLogicalPlan: baseLogicalPlan{},
 							name:            "src1",
-							streamFields:    map[string]*ast.JsonStreamField{},
+							streamFields:    nil,
 							streamStmt:      streams["src1"],
 							metaFields:      []string{"device"},
 							isWildCard:      true,
@@ -3877,7 +3877,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 									DataSourcePlan{
 										baseLogicalPlan: baseLogicalPlan{},
 										name:            "src1",
-										streamFields:    map[string]*ast.JsonStreamField{},
+										streamFields:    nil,
 										streamStmt:      streams["src1"],
 										metaFields:      []string{},
 										isWildCard:      true,
@@ -3926,7 +3926,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 									DataSourcePlan{
 										baseLogicalPlan: baseLogicalPlan{},
 										name:            "src1",
-										streamFields:    map[string]*ast.JsonStreamField{},
+										streamFields:    nil,
 										streamStmt:      streams["src1"],
 										metaFields:      []string{},
 										isWildCard:      true,
@@ -4574,7 +4574,7 @@ func Test_createLogicalPlan4Lookup(t *testing.T) {
 																baseLogicalPlan: baseLogicalPlan{},
 																name:            "src1",
 																streamStmt:      streams["src1"],
-																streamFields:    map[string]*ast.JsonStreamField{},
+																streamFields:    nil,
 																metaFields:      []string{},
 																isWildCard:      true,
 																isSchemaless:    true,
@@ -4663,7 +4663,7 @@ func Test_createLogicalPlan4Lookup(t *testing.T) {
 																baseLogicalPlan: baseLogicalPlan{},
 																name:            "src1",
 																streamStmt:      streams["src1"],
-																streamFields:    map[string]*ast.JsonStreamField{},
+																streamFields:    nil,
 																metaFields:      []string{},
 																isWildCard:      true,
 																isSchemaless:    true,