Repository: lf-edge/ekuiper
Pull request #2906 (2024-06-05 06:28:17)

feat: support connection status api

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?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 `43.75000%` with `27 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@0753030`). [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.

> :exclamation: **Current head 2f7c51c differs from pull request most recent head c41ef1b**
> 
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) reports for the commit c41ef1b to get more accurate results.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?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/io/connection/connection\_mock.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?src=pr&el=tree&filepath=internal%2Fio%2Fconnection%2Fconnection_mock.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vY29ubmVjdGlvbi9jb25uZWN0aW9uX21vY2suZ28=) | 20.00% | [12 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/connection/connection\_pool.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?src=pr&el=tree&filepath=internal%2Fio%2Fconnection%2Fconnection_pool.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vY29ubmVjdGlvbi9jb25uZWN0aW9uX3Bvb2wuZ28=) | 72.00% | [4 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/connection.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?src=pr&el=tree&filepath=internal%2Fserver%2Fconnection.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL2Nvbm5lY3Rpb24uZ28=) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/rest.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?src=pr&el=tree&filepath=internal%2Fserver%2Frest.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3Jlc3QuZ28=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?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    #2906   +/-   ##
============================================
  Coverage             ?   67.85%           
============================================
  Files                ?      294           
  Lines                ?    28243           
  Branches             ?        0           
============================================
  Hits                 ?    19164           
  Misses               ?     7369           
  Partials             ?     1710           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2906?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/contract/api/source.go b/contract/api/source.go
index 64c02e8e..9a2c1e04 100644
--- a/contract/api/source.go
+++ b/contract/api/source.go
@@ -68,14 +68,9 @@ type Rewindable interface {
 
 // LookupSource is a source feature to query the source on demand (TO be modified later)
 type LookupSource interface {
-	// Open creates the connection to the external data source
-	Open(ctx StreamContext) error
-	// Configure Called during initialization. Configure the source with the data source(e.g. topic for mqtt) and the properties
-	// read from the yaml
-	Configure(datasource string, props map[string]interface{}) error
+	Source
 	// Lookup receive lookup values to construct the query and return query results
-	Lookup(ctx StreamContext, fields []string, keys []string, values []interface{}) (MessageTupleList, error)
-	Closable
+	Lookup(ctx StreamContext, fields []string, keys []string, values []any) ([]map[string]any, error)
 }
 
 /// helper function definition
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index fccf4f78..7bb314ef 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -44,7 +44,7 @@ func init() {
 	modules.RegisterSink("file", file.GetSink)
 	// modules.RegisterSink("websocket", func() api.Sink { return &websocket.WebSocketSink{} })
 
-	// modules.RegisterLookupSource("memory", func() api.LookupSource { return memory.GetLookupSource() })
+	modules.RegisterLookupSource("memory", memory.GetLookupSource)
 	// modules.RegisterLookupSource("httppull", func() api.LookupSource { return http.GetLookUpSource() })
 
 	modules.RegisterConnection("mqtt", mqttCon.CreateConnection)
diff --git a/internal/io/connection/connection_mock.go b/internal/io/connection/connection_mock.go
index 0930f41d..863ff238 100644
--- a/internal/io/connection/connection_mock.go
+++ b/internal/io/connection/connection_mock.go
@@ -15,6 +15,9 @@
 package connection
 
 import (
+	"errors"
+	"strings"
+
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
@@ -23,6 +26,7 @@ import (
 func InitMockTest() {
 	conf.IsTesting = true
 	modules.ConnectionRegister["mock"] = CreateMockConnection
+	modules.ConnectionRegister[strings.ToLower("mockErr")] = CreateMockErrConnection
 }
 
 type mockConnection struct {
@@ -61,3 +65,33 @@ func CreateMockConnection(ctx api.StreamContext, id string, props map[string]any
 	m := &mockConnection{id: id, ref: 0}
 	return m, nil
 }
+
+type mockErrConnection struct{}
+
+func (m mockErrConnection) Ping(ctx api.StreamContext) error {
+	return errors.New("mockErr")
+}
+
+func (m mockErrConnection) Close(ctx api.StreamContext) {
+	return
+}
+
+func (m mockErrConnection) Attach(ctx api.StreamContext) {
+	return
+}
+
+func (m mockErrConnection) DetachSub(ctx api.StreamContext, props map[string]any) {
+	return
+}
+
+func (m mockErrConnection) DetachPub(ctx api.StreamContext, props map[string]any) {
+	return
+}
+
+func (m mockErrConnection) Ref(ctx api.StreamContext) int {
+	return 0
+}
+
+func CreateMockErrConnection(ctx api.StreamContext, id string, props map[string]any) (modules.Connection, error) {
+	return nil, errors.New("mockErr")
+}
diff --git a/internal/io/connection/connection_pool.go b/internal/io/connection/connection_pool.go
index 3da1c89d..c5e3c33e 100644
--- a/internal/io/connection/connection_pool.go
+++ b/internal/io/connection/connection_pool.go
@@ -47,6 +47,31 @@ func dropConnectionStore(plugin, id string) error {
 	return err
 }
 
+func GetAllConnectionStatus(ctx api.StreamContext) map[string]ConnectionStatus {
+	globalConnectionManager.RLock()
+	defer globalConnectionManager.RUnlock()
+	s := make(map[string]ConnectionStatus)
+	for id, err := range globalConnectionManager.failConnection {
+		status := ConnectionStatus{
+			Status: ConnectionFail,
+			ErrMsg: err,
+		}
+		s[id] = status
+	}
+	for id, meta := range globalConnectionManager.connectionPool {
+		status := ConnectionStatus{
+			Status: ConnectionRunning,
+		}
+		err := meta.conn.Ping(ctx)
+		if err != nil {
+			status.Status = ConnectionFail
+			status.ErrMsg = err.Error()
+		}
+		s[id] = status
+	}
+	return s
+}
+
 func GetAllConnectionsID() []string {
 	globalConnectionManager.RLock()
 	defer globalConnectionManager.RUnlock()
@@ -102,6 +127,9 @@ func CreateNamedConnection(ctx api.StreamContext, id, typ string, props map[stri
 	}
 	meta.conn = conn
 	globalConnectionManager.connectionPool[id] = meta
+	if _, ok := globalConnectionManager.failConnection[id]; ok {
+		delete(globalConnectionManager.failConnection, id)
+	}
 	return conn, nil
 }
 
@@ -181,6 +209,10 @@ func DropNameConnection(ctx api.StreamContext, selId string) error {
 	defer globalConnectionManager.Unlock()
 	meta, ok := globalConnectionManager.connectionPool[selId]
 	if !ok {
+		_, ok := globalConnectionManager.failConnection[selId]
+		if ok {
+			delete(globalConnectionManager.failConnection, selId)
+		}
 		return nil
 	}
 	conn := meta.conn
@@ -207,6 +239,7 @@ func InitConnectionManager4Test() error {
 func InitConnectionManager() {
 	globalConnectionManager = &ConnectionManager{
 		connectionPool: make(map[string]ConnectionMeta),
+		failConnection: make(map[string]string),
 	}
 	if conf.IsTesting {
 		return
@@ -234,6 +267,7 @@ func ReloadConnection() error {
 		conn, err := createNamedConnection(context.Background(), meta)
 		if err != nil {
 			conf.Log.Warnf("initialize connection:%v failed, err:%v", id, err)
+			globalConnectionManager.failConnection[id] = err.Error()
 			continue
 		}
 		meta.conn = conn
@@ -245,6 +279,7 @@ func ReloadConnection() error {
 type ConnectionManager struct {
 	sync.RWMutex
 	connectionPool map[string]ConnectionMeta
+	failConnection map[string]string
 }
 
 type ConnectionMeta struct {
@@ -254,10 +289,6 @@ type ConnectionMeta struct {
 	conn  modules.Connection `json:"-"`
 }
 
-func init() {
-	modules.ConnectionRegister["mock"] = CreateMockConnection
-}
-
 func NewExponentialBackOff() *backoff.ExponentialBackOff {
 	return backoff.NewExponentialBackOff(
 		backoff.WithInitialInterval(DefaultInitialInterval),
@@ -272,3 +303,13 @@ const (
 )
 
 var DefaultBackoffMaxElapsedDuration = 3 * time.Minute
+
+const (
+	ConnectionRunning = "running"
+	ConnectionFail    = "fail"
+)
+
+type ConnectionStatus struct {
+	Status string
+	ErrMsg string
+}
diff --git a/internal/io/file/sink.go b/internal/io/file/sink.go
index 4b0787b4..f0669e5a 100644
--- a/internal/io/file/sink.go
+++ b/internal/io/file/sink.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
@@ -122,22 +123,28 @@ func (m *fileSink) Connect(ctx api.StreamContext) error {
 	// Check if the files have opened longer than the rolling interval, if so close it and create a new one
 	if m.c.CheckInterval > 0 {
 		t := timex.GetTicker(m.c.CheckInterval)
-		go func() {
+		go func() { // this will never panic
 			defer t.Stop()
 			for {
 				select {
 				case now := <-t.C:
-					m.mux.Lock()
-					for k, v := range m.fws {
-						if now.Sub(v.Start) > m.c.RollingInterval {
-							err := m.roll(ctx, k, v)
-							// TODO how to deal with this error
-							if err != nil {
-								ctx.GetLogger().Errorf("file sink fails to close file %s with error %s.", k, err)
+					e := infra.SafeRun(func() error {
+						m.mux.Lock()
+						defer m.mux.Unlock()
+						for k, v := range m.fws {
+							if now.Sub(v.Start) > m.c.RollingInterval {
+								err := m.roll(ctx, k, v)
+								// TODO how to deal with this error
+								if err != nil {
+									return fmt.Errorf("file sink fails to close file %s with error %s.", k, err)
+								}
 							}
 						}
+						return nil
+					})
+					if e != nil {
+						ctx.GetLogger().Error(e)
 					}
-					m.mux.Unlock()
 				case <-ctx.Done():
 					ctx.GetLogger().Info("file sink done")
 					return
diff --git a/internal/io/file/source.go b/internal/io/file/source.go
index b5c5fea2..8e55a920 100644
--- a/internal/io/file/source.go
+++ b/internal/io/file/source.go
@@ -28,6 +28,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	_ "github.com/lf-edge/ekuiper/v2/internal/io/file/reader"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
@@ -179,8 +180,14 @@ func (fs *Source) Load(ctx api.StreamContext, ingest api.TupleIngest, ingestErro
 				}
 				wg.Add(1)
 				go func(file string) {
-					defer wg.Done()
-					fs.parseFile(ctx, file, ingest, ingestError)
+					e := infra.SafeRun(func() error {
+						defer wg.Done()
+						fs.parseFile(ctx, file, ingest, ingestError)
+						return nil
+					})
+					if e != nil {
+						ingestError(ctx, e)
+					}
 				}(filepath.Join(fs.file, entry.Name()))
 			}
 			wg.Wait()
@@ -278,24 +285,38 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Tuple
 func ignoreLines(ctx api.StreamContext, reader io.Reader, ignoreStartLines int, ignoreEndLines int) io.Reader {
 	r, w := io.Pipe()
 	go func() {
-		defer func() {
-			w.Close()
-			reader.(io.ReadCloser).Close()
-		}()
-		scanner := bufio.NewScanner(reader)
-		scanner.Split(bufio.ScanLines)
+		e := infra.SafeRun(func() error {
+			defer func() {
+				w.Close()
+				reader.(io.ReadCloser).Close()
+			}()
+			scanner := bufio.NewScanner(reader)
+			scanner.Split(bufio.ScanLines)
 
-		ln := 0
-		// This is a queue to store the lines that should be ignored
-		tempLines := make([][]byte, 0, ignoreEndLines)
-		for scanner.Scan() {
-			if ln >= ignoreStartLines {
-				if ignoreEndLines > 0 { // the last n line are left in the tempLines
-					slot := (ln - ignoreStartLines) % ignoreEndLines
-					if len(tempLines) <= slot { // first round
-						tempLines = append(tempLines, scanner.Bytes())
+			ln := 0
+			// This is a queue to store the lines that should be ignored
+			tempLines := make([][]byte, 0, ignoreEndLines)
+			for scanner.Scan() {
+				if ln >= ignoreStartLines {
+					if ignoreEndLines > 0 { // the last n line are left in the tempLines
+						slot := (ln - ignoreStartLines) % ignoreEndLines
+						if len(tempLines) <= slot { // first round
+							tempLines = append(tempLines, scanner.Bytes())
+						} else {
+							_, err := w.Write(tempLines[slot])
+							if err != nil {
+								ctx.GetLogger().Error(err)
+								break
+							}
+							_, err = w.Write([]byte{'\n'})
+							if err != nil {
+								ctx.GetLogger().Error(err)
+								break
+							}
+							tempLines[slot] = scanner.Bytes()
+						}
 					} else {
-						_, err := w.Write(tempLines[slot])
+						_, err := w.Write(scanner.Bytes())
 						if err != nil {
 							ctx.GetLogger().Error(err)
 							break
@@ -305,22 +326,14 @@ func ignoreLines(ctx api.StreamContext, reader io.Reader, ignoreStartLines int,
 							ctx.GetLogger().Error(err)
 							break
 						}
-						tempLines[slot] = scanner.Bytes()
-					}
-				} else {
-					_, err := w.Write(scanner.Bytes())
-					if err != nil {
-						ctx.GetLogger().Error(err)
-						break
-					}
-					_, err = w.Write([]byte{'\n'})
-					if err != nil {
-						ctx.GetLogger().Error(err)
-						break
 					}
 				}
+				ln++
 			}
-			ln++
+			return nil
+		})
+		if e != nil {
+			ctx.GetLogger().Error(e)
 		}
 	}()
 	return r
diff --git a/internal/io/memory/lookupsource.go b/internal/io/memory/lookupsource.go
new file mode 100644
index 00000000..2bc080d3
--- /dev/null
+++ b/internal/io/memory/lookupsource.go
@@ -0,0 +1,92 @@
+// Copyright 2022-2023 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 memory
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/store"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+type lc struct {
+	Topic string `json:"datasource"`
+	Key   string `json:"key"`
+}
+
+// lookupsource is a lookup source that reads data from memory
+// The memory lookup table reads a global memory store for data
+type lookupsource struct {
+	topic      string
+	topicRegex *regexp.Regexp
+	table      *store.Table
+	key        string
+}
+
+func (s *lookupsource) Connect(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("lookup source %s is opened with key %v", s.topic, s.key)
+	var err error
+	s.table, err = store.Reg(s.topic, s.topicRegex, s.key)
+	return err
+}
+
+func (s *lookupsource) Provision(ctx api.StreamContext, props map[string]any) error {
+	cfg := &lc{}
+	err := cast.MapToStruct(props, cfg)
+	if err != nil {
+		return fmt.Errorf("read properties %v fail with error: %v", props, err)
+	}
+	if cfg.Topic == "" {
+		return fmt.Errorf("datasource(topic) is required")
+	}
+	if strings.ContainsAny(cfg.Topic, "+#") {
+		r, err := getRegexp(cfg.Topic)
+		if err != nil {
+			return err
+		}
+		s.topicRegex = r
+	}
+	if cfg.Key == "" {
+		return fmt.Errorf("key is required for lookup source")
+	}
+	s.topic = cfg.Topic
+	s.key = cfg.Key
+	return nil
+}
+
+func (s *lookupsource) Lookup(ctx api.StreamContext, _ []string, keys []string, values []interface{}) ([]map[string]any, error) {
+	ctx.GetLogger().Debugf("lookup source %s is looking up keys %v with values %v", s.topic, keys, values)
+	tuples, err := s.table.Read(keys, values)
+	if err != nil {
+		return nil, err
+	}
+	r := make([]map[string]any, len(tuples))
+	for i, t := range tuples {
+		r[i] = t.ToMap()
+	}
+	return r, nil
+}
+
+func (s *lookupsource) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("lookup source %s is closing", s.topic)
+	return store.Unreg(s.topic, s.key)
+}
+
+func GetLookupSource() api.LookupSource {
+	return &lookupsource{}
+}
diff --git a/internal/io/memory/pubsub/manager.go b/internal/io/memory/pubsub/manager.go
index 1b725205..ea004fc6 100644
--- a/internal/io/memory/pubsub/manager.go
+++ b/internal/io/memory/pubsub/manager.go
@@ -20,7 +20,6 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 )
 
 const IdProperty = "topic"
@@ -110,11 +109,11 @@ func RemovePub(topic string) {
 	}
 }
 
-func Produce(ctx api.StreamContext, topic string, data *xsql.Tuple) {
+func Produce(ctx api.StreamContext, topic string, data MemTuple) {
 	doProduce(ctx, topic, data)
 }
 
-func ProduceList(ctx api.StreamContext, topic string, list []*xsql.Tuple) {
+func ProduceList(ctx api.StreamContext, topic string, list []MemTuple) {
 	doProduce(ctx, topic, list)
 }
 
diff --git a/internal/io/memory/pubsub/tuple.go b/internal/io/memory/pubsub/tuple.go
index c6cbc44e..9fe38e72 100644
--- a/internal/io/memory/pubsub/tuple.go
+++ b/internal/io/memory/pubsub/tuple.go
@@ -14,12 +14,13 @@
 
 package pubsub
 
-import (
-	"github.com/lf-edge/ekuiper/v2/pkg/model"
-)
-
 type UpdatableTuple struct {
-	*model.DefaultSourceTuple
+	MemTuple
 	Rowkind string
 	Keyval  interface{}
 }
+
+type MemTuple interface {
+	Value(key, table string) (any, bool)
+	ToMap() map[string]any
+}
diff --git a/internal/io/memory/sink.go b/internal/io/memory/sink.go
index 5250d49c..25303216 100644
--- a/internal/io/memory/sink.go
+++ b/internal/io/memory/sink.go
@@ -21,28 +21,21 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
 	"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/timex"
 )
 
 type config struct {
-	Topic        string   `json:"topic"`
-	DataTemplate string   `json:"dataTemplate"`
-	RowkindField string   `json:"rowkindField"`
-	KeyField     string   `json:"keyField"`
-	Fields       []string `json:"fields"`
-	DataField    string   `json:"dataField"`
-	ResendTopic  string   `json:"resendDestination"`
+	Topic        string `json:"topic"`
+	RowkindField string `json:"rowkindField"`
+	KeyField     string `json:"keyField"`
 }
 
 type sink struct {
 	topic        string
-	hasTransform bool
 	keyField     string
 	rowkindField string
-	fields       []string
-	dataField    string
-	resendTopic  string
 	meta         map[string]any
 }
 
@@ -56,20 +49,11 @@ func (s *sink) Provision(_ api.StreamContext, props map[string]any) error {
 		return fmt.Errorf("invalid memory topic %s: wildcard found", cfg.Topic)
 	}
 	s.topic = cfg.Topic
-	if cfg.DataTemplate != "" {
-		s.hasTransform = true
-	}
-	s.dataField = cfg.DataField
-	s.fields = cfg.Fields
 	s.rowkindField = cfg.RowkindField
 	s.keyField = cfg.KeyField
 	if s.rowkindField != "" && s.keyField == "" {
 		return fmt.Errorf("keyField is required when rowkindField is set")
 	}
-	s.resendTopic = cfg.ResendTopic
-	if s.resendTopic == "" {
-		s.resendTopic = s.topic
-	}
 	s.meta = map[string]any{
 		"topic": cfg.Topic,
 	}
@@ -91,10 +75,45 @@ func (s *sink) Collect(ctx api.StreamContext, data api.MessageTuple) error {
 		}
 	}
 	ctx.GetLogger().Debugf("publishing to topic %s", topic)
-	pubsub.Produce(ctx, topic, &xsql.Tuple{Message: data.ToMap(), Metadata: s.meta, Timestamp: timex.GetNow()})
+	var (
+		t   pubsub.MemTuple = &xsql.Tuple{Message: data.ToMap(), Metadata: s.meta, Timestamp: timex.GetNow()}
+		err error
+	)
+	if s.rowkindField != "" {
+		t, err = s.wrapUpdatable(t)
+		if err != nil {
+			return err
+		}
+	}
+	pubsub.Produce(ctx, topic, t)
 	return nil
 }
 
+func (s *sink) wrapUpdatable(el pubsub.MemTuple) (pubsub.MemTuple, error) {
+	c, ok := el.Value(s.rowkindField, "")
+	var rowkind string
+	if !ok {
+		rowkind = ast.RowkindUpsert
+	} else {
+		rowkind, ok = c.(string)
+		if !ok {
+			return nil, fmt.Errorf("rowkind field %s is not a string in data %v", s.rowkindField, el)
+		}
+		if rowkind != ast.RowkindInsert && rowkind != ast.RowkindUpdate && rowkind != ast.RowkindDelete && rowkind != ast.RowkindUpsert {
+			return nil, fmt.Errorf("invalid rowkind %s", rowkind)
+		}
+	}
+	key, ok := el.Value(s.keyField, "")
+	if !ok {
+		return nil, fmt.Errorf("key field %s not found in data %v", s.keyField, el)
+	}
+	return &pubsub.UpdatableTuple{
+		MemTuple: el,
+		Rowkind:  rowkind,
+		Keyval:   key,
+	}, nil
+}
+
 func (s *sink) CollectList(ctx api.StreamContext, tuples api.MessageTupleList) error {
 	topic := s.topic
 	if dp, ok := tuples.(api.HasDynamicProps); ok {
@@ -103,20 +122,25 @@ func (s *sink) CollectList(ctx api.StreamContext, tuples api.MessageTupleList) e
 			topic = temp
 		}
 	}
-	result := make([]*xsql.Tuple, tuples.Len())
+	result := make([]pubsub.MemTuple, tuples.Len())
 	tuples.RangeOfTuples(func(index int, tuple api.MessageTuple) bool {
-		result[index] = &xsql.Tuple{Message: tuple.ToMap(), Metadata: s.meta, Timestamp: timex.GetNow()}
+		t := &xsql.Tuple{Message: tuple.ToMap(), Metadata: s.meta, Timestamp: timex.GetNow()}
+		if s.rowkindField != "" {
+			st, err := s.wrapUpdatable(t)
+			if err != nil {
+				ctx.GetLogger().Errorf("cannot convert %v to updatable %v", t, err)
+			} else {
+				result[index] = st
+				return true
+			}
+		}
+		result[index] = t
 		return true
 	})
 	pubsub.ProduceList(ctx, topic, result)
 	return nil
 }
 
-//func (s *sink) CollectResend(ctx api.StreamContext, data interface{}) error {
-//	ctx.GetLogger().Debugf("resend %+v", data)
-//	return s.collectWithTopic(ctx, data, s.resendTopic)
-//}
-
 func (s *sink) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Debugf("closing memory sink")
 	pubsub.RemovePub(s.topic)
diff --git a/internal/io/memory/source.go b/internal/io/memory/source.go
index 37b81418..c04846eb 100644
--- a/internal/io/memory/source.go
+++ b/internal/io/memory/source.go
@@ -22,6 +22,7 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
@@ -63,14 +64,20 @@ func (s *source) Connect(_ api.StreamContext) error {
 }
 
 // Subscribe For memory source, it can receive a source tuple directly. So just pass it through
-func (s *source) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, _ api.ErrorIngest) error {
+func (s *source) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, ingestErr api.ErrorIngest) error {
 	ch := pubsub.CreateSub(s.c.Topic, s.topicRegex, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), s.c.BufferLength)
 	ctx.GetLogger().Infof("Subscribe to topic %s", s.c.Topic)
 	go func() {
 		for {
 			select {
 			case v := <-ch:
-				ingest(ctx, v, nil, timex.GetNow())
+				e := infra.SafeRun(func() error {
+					ingest(ctx, v, nil, timex.GetNow())
+					return nil
+				})
+				if e != nil {
+					ingestErr(ctx, e)
+				}
 			case <-ctx.Done():
 				return
 			}
diff --git a/internal/io/memory/store/db.go b/internal/io/memory/store/db.go
new file mode 100644
index 00000000..328c4c15
--- /dev/null
+++ b/internal/io/memory/store/db.go
@@ -0,0 +1,181 @@
+// Copyright 2022-2023 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 store
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+)
+
+type tableCount struct {
+	sync.RWMutex
+	count int
+	t     *Table
+}
+
+func (tc *tableCount) Increase() int {
+	tc.Lock()
+	defer tc.Unlock()
+	tc.count++
+	return tc.count
+}
+
+func (tc *tableCount) Decrease() int {
+	tc.Lock()
+	defer tc.Unlock()
+	tc.count--
+	if tc.count < 0 {
+		conf.Log.Errorf("Table count is less than 0: %d", tc.count)
+	}
+	return tc.count
+}
+
+type database struct {
+	sync.RWMutex
+	tables map[string]*tableCount // topic_key: table
+}
+
+// getTable return the table of the topic.
+func (db *database) getTable(topic string, key string) (*Table, bool) {
+	db.RLock()
+	defer db.RUnlock()
+	tableId := fmt.Sprintf("%s_%s", topic, key)
+	tc, ok := db.tables[tableId]
+	if ok {
+		return tc.t, true
+	} else {
+		return nil, false
+	}
+}
+
+// addTable add a table to the database
+// If the table already exists, return the existing table;
+// otherwise, create a new table and return it.
+// The second argument is to indicate if the table is newly created
+func (db *database) addTable(topic string, key string) (*Table, bool) {
+	db.Lock()
+	defer db.Unlock()
+	tableId := fmt.Sprintf("%s_%s", topic, key)
+	tc, ok := db.tables[tableId]
+	if ok {
+		tc.Increase()
+	} else {
+		t := createTable(topic, key)
+		tc = &tableCount{
+			count: 1,
+			t:     t,
+		}
+		db.tables[tableId] = tc
+	}
+	return tc.t, !ok
+}
+
+// dropTable drop the table of the topic/values
+// stops to accumulate job
+// deletes the cache data
+func (db *database) dropTable(topic string, key string) error {
+	tableId := fmt.Sprintf("%s_%s", topic, key)
+	db.Lock()
+	defer db.Unlock()
+	if tc, ok := db.tables[tableId]; ok {
+		if tc.Decrease() == 0 {
+			if tc.t != nil && tc.t.cancel != nil {
+				tc.t.cancel()
+			}
+			delete(db.tables, tableId)
+		}
+		return nil
+	}
+	return fmt.Errorf("Table %s not found", tableId)
+}
+
+// Table has one writer and multiple reader
+type Table struct {
+	sync.RWMutex
+	topic string
+	key   string
+	// datamap is the overall data indexed by primary key
+	datamap map[any]pubsub.MemTuple
+	cancel  context.CancelFunc
+}
+
+func createTable(topic string, key string) *Table {
+	t := &Table{topic: topic, key: key, datamap: make(map[any]pubsub.MemTuple)}
+	return t
+}
+
+func (t *Table) add(value pubsub.MemTuple) {
+	t.Lock()
+	defer t.Unlock()
+	keyval, ok := value.Value(t.key, "")
+	if !ok {
+		conf.Log.Errorf("add to table %s omitted, value not found for key %s", t.topic, t.key)
+	}
+	t.datamap[keyval] = value
+}
+
+func (t *Table) delete(key interface{}) {
+	t.Lock()
+	defer t.Unlock()
+	delete(t.datamap, key)
+}
+
+func (t *Table) Read(keys []string, values []interface{}) ([]pubsub.MemTuple, error) {
+	t.RLock()
+	defer t.RUnlock()
+	// Find the primary key
+	var matched pubsub.MemTuple
+	for i, k := range keys {
+		if k == t.key {
+			matched = t.datamap[values[i]]
+		}
+	}
+	if matched != nil {
+		match := true
+		for i, k := range keys {
+			if val, ok := matched.Value(k, ""); !ok || val != values[i] {
+				match = false
+				break
+			}
+		}
+		if match {
+			return []pubsub.MemTuple{matched}, nil
+		} else {
+			return nil, nil
+		}
+	}
+	var result []pubsub.MemTuple
+	for _, v := range t.datamap {
+		match := true
+		for i, k := range keys {
+			if val, ok := v.Value(k, ""); !ok || val != values[i] {
+				match = false
+				break
+			}
+		}
+		if match {
+			result = append(result, v)
+		}
+	}
+	return result, nil
+}
+
+var db = &database{
+	tables: make(map[string]*tableCount),
+}
diff --git a/internal/io/memory/store/store.go b/internal/io/memory/store/store.go
new file mode 100644
index 00000000..d545cdf8
--- /dev/null
+++ b/internal/io/memory/store/store.go
@@ -0,0 +1,84 @@
+// Copyright 2022-2023 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 store
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+// Reg registers a topic to save it to memory store
+// Create a new go routine to listen to the topic and save the data to memory
+func Reg(topic string, topicRegex *regexp.Regexp, key string) (*Table, error) {
+	t, isNew := db.addTable(topic, key)
+	if isNew {
+		go runTable(topic, topicRegex, t)
+	}
+	return t, nil
+}
+
+// runTable should only run in a single instance.
+// This go routine is used to accumulate data in memory
+// If the go routine close, the go routine exits but the data will be kept until table dropped
+func runTable(topic string, topicRegex *regexp.Regexp, t *Table) {
+	conf.Log.Infof("runTable %s", topic)
+	ch := pubsub.CreateSub(topic, topicRegex, fmt.Sprintf("store_%s", topic), 1024)
+	ctx, cancel := context.WithCancel(context.Background())
+	t.cancel = cancel
+	for {
+		select {
+		case v := <-ch:
+			switch vv := v.(type) {
+			case pubsub.MemTuple:
+				ingestMemTuple(t, vv)
+			case []pubsub.MemTuple:
+				for _, vvv := range vv {
+					ingestMemTuple(t, vvv)
+				}
+			default:
+				// should never happen
+				conf.Log.Errorf("add wrong data %v for table %s", v, topic)
+			}
+			conf.Log.Debugf("receive data %v for %s", v, topic)
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func ingestMemTuple(t *Table, tuple pubsub.MemTuple) {
+	switch tt := tuple.(type) {
+	case *pubsub.UpdatableTuple:
+		switch tt.Rowkind {
+		case ast.RowkindInsert, ast.RowkindUpdate, ast.RowkindUpsert:
+			t.add(tt.MemTuple)
+		case ast.RowkindDelete:
+			t.delete(tt.Keyval)
+		}
+	default:
+		t.add(tt)
+	}
+}
+
+// Unreg unregisters a topic to remove it from memory store
+func Unreg(topic string, key string) error {
+	// Must be an atomic operation
+	return db.dropTable(topic, key)
+}
diff --git a/internal/plugin/portable/manager.go b/internal/plugin/portable/manager.go
index 578aa11c..dc542351 100644
--- a/internal/plugin/portable/manager.go
+++ b/internal/plugin/portable/manager.go
@@ -162,7 +162,7 @@ func (m *Manager) doRegister(name string, pi *PluginInfo, isInit bool) error {
 		}
 	}
 	conf.Log.Infof("Installed portable plugin %s successfully", name)
-	runtime.GetPluginInsManager().CreateIns(&pi.PluginMeta)
+	runtime.GetPluginInsManager().CreateIns(&pi.PluginMeta, isInit)
 	return nil
 }
 
diff --git a/internal/plugin/portable/runtime/connection.go b/internal/plugin/portable/runtime/connection.go
index ba51feef..02474131 100644
--- a/internal/plugin/portable/runtime/connection.go
+++ b/internal/plugin/portable/runtime/connection.go
@@ -15,10 +15,12 @@
 package runtime
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"go.nanomsg.org/mangos/v3"
 	"go.nanomsg.org/mangos/v3/protocol/pull"
 	"go.nanomsg.org/mangos/v3/protocol/push"
@@ -245,6 +247,9 @@ func CreateSinkAckChannel(ctx api.StreamContext) (DataInChannel, error) {
 }
 
 func CreateControlChannel(pluginName string) (ControlChannel, error) {
+	failpoint.Inject("CreateControlChannelErr", func() {
+		failpoint.Return(nil, errors.New("CreateControlChannelErr"))
+	})
 	var (
 		sock mangos.Socket
 		err  error
diff --git a/internal/plugin/portable/runtime/function.go b/internal/plugin/portable/runtime/function.go
index 40f0ab43..95919259 100644
--- a/internal/plugin/portable/runtime/function.go
+++ b/internal/plugin/portable/runtime/function.go
@@ -41,7 +41,7 @@ func NewPortableFunc(symbolName string, reg *PluginMeta) (_ *PortableFunc, e err
 	// Setup channel and route the data
 	conf.Log.Infof("Start running portable function meta %+v", reg)
 	pm := GetPluginInsManager()
-	ins, err := pm.getOrStartProcess(reg, PortbleConf)
+	ins, err := pm.GetOrStartProcess(reg, PortbleConf)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager.go b/internal/plugin/portable/runtime/plugin_ins_manager.go
index 3ef4bbf3..a9dd60ca 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager.go
@@ -16,6 +16,7 @@ package runtime
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"os/exec"
@@ -23,6 +24,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pingcap/failpoint"
+
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
@@ -49,6 +52,7 @@ type PluginIns struct {
 	// audit the commands, so that when restarting the plugin, we can replay the commands
 	commands map[Meta][]byte
 	process  *os.Process // created when used by rule and deleted when no rule uses it
+	Status   *PluginStatus
 }
 
 func NewPluginIns(name string, ctrlChan ControlChannel, process *os.Process) *PluginIns {
@@ -57,6 +61,7 @@ func NewPluginIns(name string, ctrlChan ControlChannel, process *os.Process) *Pl
 		ctrlChan: ctrlChan,
 		name:     name,
 		commands: make(map[Meta][]byte),
+		Status:   NewPluginStatus(),
 	}
 }
 
@@ -151,12 +156,19 @@ func (i *PluginIns) Stop() error {
 	var err error
 	i.RLock()
 	defer i.RUnlock()
+	i.Status.Stop()
 	if i.process != nil { // will also trigger process exit clean up
 		err = i.process.Kill()
 	}
 	return err
 }
 
+func (i *PluginIns) GetStatus() *PluginStatus {
+	i.RLock()
+	defer i.RUnlock()
+	return i.Status
+}
+
 // Manager plugin process and control socket
 type pluginInsManager struct {
 	instances map[string]*PluginIns
@@ -172,17 +184,42 @@ func GetPluginInsManager() *pluginInsManager {
 	return pm
 }
 
-func (p *pluginInsManager) getPluginIns(name string) (*PluginIns, bool) {
+func GetPluginInsManager4Test() *pluginInsManager {
+	testPM := &pluginInsManager{
+		instances: make(map[string]*PluginIns),
+	}
+	return testPM
+}
+
+func (p *pluginInsManager) GetPluginInsStatus(name string) (*PluginStatus, bool) {
+	p.RLock()
+	defer p.RUnlock()
+	ins, ok := p.getPluginIns(name)
+	if !ok {
+		return nil, false
+	}
+	return ins.GetStatus(), true
+}
+
+func (p *pluginInsManager) GetPluginIns(name string) (*PluginIns, bool) {
 	p.RLock()
 	defer p.RUnlock()
+	return p.getPluginIns(name)
+}
+
+func (p *pluginInsManager) getPluginIns(name string) (*PluginIns, bool) {
 	ins, ok := p.instances[name]
 	return ins, ok
 }
 
-// deletePluginIns should only run when there is no state aka. commands
-func (p *pluginInsManager) deletePluginIns(name string) {
+func (p *pluginInsManager) DeletePluginIns(name string) {
 	p.Lock()
 	defer p.Unlock()
+	p.deletePluginIns(name)
+}
+
+// deletePluginIns should only run when there is no state aka. commands
+func (p *pluginInsManager) deletePluginIns(name string) {
 	delete(p.instances, name)
 }
 
@@ -194,16 +231,25 @@ func (p *pluginInsManager) AddPluginIns(name string, ins *PluginIns) {
 }
 
 // CreateIns Run when plugin is created/updated
-func (p *pluginInsManager) CreateIns(pluginMeta *PluginMeta) {
+func (p *pluginInsManager) CreateIns(pluginMeta *PluginMeta, isInit bool) {
 	p.Lock()
 	defer p.Unlock()
-	if ins, ok := p.instances[pluginMeta.Name]; ok {
-		if len(ins.commands) != 0 {
-			go p.getOrStartProcess(pluginMeta, PortbleConf)
+	if !isInit {
+		if ins, ok := p.instances[pluginMeta.Name]; ok {
+			if len(ins.commands) != 0 {
+				go p.GetOrStartProcess(pluginMeta, PortbleConf)
+			}
 		}
+	} else {
+		go p.GetOrStartProcess(pluginMeta, PortbleConf)
 	}
 }
 
+// GetOrStartProcess Control the plugin process lifecycle.
+func (p *pluginInsManager) GetOrStartProcess(pluginMeta *PluginMeta, pconf *PortableConfig) (_ *PluginIns, e error) {
+	return p.getOrStartProcess(pluginMeta, pconf)
+}
+
 // getOrStartProcess Control the plugin process lifecycle.
 // Need to manage the resources: instances map, control socket, plugin process
 // May be called at plugin creation or restart with previous state(ctrlCh, commands)
@@ -233,6 +279,7 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 		conf.Log.Infof("create control channel")
 		ctrlChan, err := CreateControlChannel(pluginMeta.Name)
 		if err != nil {
+			ins.Status.StatusErr(err)
 			return nil, fmt.Errorf("can't create new control channel: %s", err.Error())
 		}
 		ins.ctrlChan = ctrlChan
@@ -240,7 +287,11 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 	// init or restart all need to run the process
 	conf.Log.Infof("executing plugin")
 	jsonArg, err := json.Marshal(pconf)
+	failpoint.Inject("confErr", func() {
+		err = errors.New("confErr")
+	})
 	if err != nil {
+		ins.Status.StatusErr(err)
 		return nil, fmt.Errorf("invalid conf: %v", pconf)
 	}
 	var cmd *exec.Cmd
@@ -249,14 +300,14 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 		case "go":
 			conf.Log.Printf("starting go plugin executable %s", pluginMeta.Executable)
 			cmd = exec.Command(pluginMeta.Executable, string(jsonArg))
-
 		case "python":
 			if pluginMeta.VirtualType != nil {
 				switch *pluginMeta.VirtualType {
 				case "conda":
 					cmd = exec.Command("conda", "run", "-n", *pluginMeta.Env, conf.Config.Portable.PythonBin, pluginMeta.Executable, string(jsonArg))
 				default:
-					return fmt.Errorf("unsupported virtual type: %s", *pluginMeta.VirtualType)
+					err = fmt.Errorf("unsupported virtual type: %s", *pluginMeta.VirtualType)
+					return err
 				}
 			}
 			if cmd == nil {
@@ -264,11 +315,13 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 			}
 			conf.Log.Infof("starting python plugin: %s", cmd)
 		default:
-			return fmt.Errorf("unsupported language: %s", pluginMeta.Language)
+			err := fmt.Errorf("unsupported language: %s", pluginMeta.Language)
+			return err
 		}
 		return nil
 	})
 	if err != nil {
+		ins.Status.StatusErr(err)
 		return nil, fmt.Errorf("fail to start plugin %s: %v", pluginMeta.Name, err)
 	}
 	cmd.Stdout = conf.Log.Out
@@ -277,7 +330,12 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 
 	conf.Log.Println("plugin starting")
 	err = cmd.Start()
+	failpoint.Inject("cmdStartErr", func() {
+		cmd.Process.Kill()
+		err = errors.New("cmdStartErr")
+	})
 	if err != nil {
+		ins.Status.StatusErr(err)
 		return nil, fmt.Errorf("plugin executable %s stops with error %v", pluginMeta.Executable, err)
 	}
 	process := cmd.Process
@@ -290,37 +348,41 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 	go infra.SafeRun(func() error { // just print out error inside
 		err = cmd.Wait()
 		if err != nil {
+			ins.Status.StatusErr(err)
 			conf.Log.Printf("plugin executable %s stops with error %v", pluginMeta.Executable, err)
 		}
 		// must make sure the plugin ins is not cleaned up yet by checking the process identity
 		// clean up for stop unintentionally
-		if ins, ok := p.getPluginIns(pluginMeta.Name); ok && ins.process == cmd.Process {
+		if ins, ok := p.GetPluginIns(pluginMeta.Name); ok && ins.process == cmd.Process {
 			ins.Lock()
 			if len(ins.commands) == 0 {
 				if ins.ctrlChan != nil {
 					_ = ins.ctrlChan.Close()
 				}
-				p.deletePluginIns(pluginMeta.Name)
 			}
 			ins.process = nil
 			ins.Unlock()
+			p.DeletePluginIns(pluginMeta.Name)
 		}
 		return nil
 	})
 	conf.Log.Println("waiting handshake")
 	err = ins.ctrlChan.Handshake()
 	if err != nil {
+		ins.Status.StatusErr(err)
 		return nil, fmt.Errorf("plugin %s control handshake error: %v", pluginMeta.Executable, err)
 	}
 	ins.process = process
 	p.instances[pluginMeta.Name] = ins
 	conf.Log.Println("plugin start running")
+	ins.Status.StartRunning()
 	// restore symbols by sending commands when restarting plugin
 	conf.Log.Info("restore plugin symbols")
 	for m, c := range ins.commands {
 		go func(key Meta, jsonArg []byte) {
 			e := ins.sendCmd(jsonArg)
 			if e != nil {
+				ins.Status.StatusErr(err)
 				conf.Log.Errorf("send command to %v error: %v", key, e)
 			}
 		}(m, c)
@@ -359,3 +421,36 @@ type PluginMeta struct {
 	VirtualType *string `json:"virtualEnvType,omitempty"`
 	Env         *string `json:"env,omitempty"`
 }
+
+const (
+	PluginStatusRunning = "running"
+	PluginStatusInit    = "initializing"
+	PluginStatusErr     = "error"
+	PluginStatusStop    = "stop"
+)
+
+type PluginStatus struct {
+	Status string `json:"status"`
+	ErrMsg string `json:"errMsg"`
+}
+
+func NewPluginStatus() *PluginStatus {
+	return &PluginStatus{
+		Status: PluginStatusInit,
+	}
+}
+
+func (s *PluginStatus) StatusErr(err error) {
+	s.Status = PluginStatusErr
+	s.ErrMsg = err.Error()
+}
+
+func (s *PluginStatus) StartRunning() {
+	s.Status = PluginStatusRunning
+	s.ErrMsg = ""
+}
+
+func (s *PluginStatus) Stop() {
+	s.Status = PluginStatusStop
+	s.ErrMsg = ""
+}
diff --git a/internal/plugin/portable/runtime/sink.go b/internal/plugin/portable/runtime/sink.go
index ca32c4a8..a96d04fe 100644
--- a/internal/plugin/portable/runtime/sink.go
+++ b/internal/plugin/portable/runtime/sink.go
@@ -51,7 +51,7 @@ func (ps *PortableSink) Provision(ctx api.StreamContext, configs map[string]any)
 func (ps *PortableSink) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Start running portable sink %s with conf %+v", ps.symbolName, ps.props)
 	pm := GetPluginInsManager()
-	ins, err := pm.getOrStartProcess(ps.reg, PortbleConf)
+	ins, err := pm.GetOrStartProcess(ps.reg, PortbleConf)
 	if err != nil {
 		return err
 	}
diff --git a/internal/plugin/portable/runtime/source.go b/internal/plugin/portable/runtime/source.go
index 8ee45a2d..a4e5cf57 100644
--- a/internal/plugin/portable/runtime/source.go
+++ b/internal/plugin/portable/runtime/source.go
@@ -44,7 +44,7 @@ func (ps *PortableSource) Provision(ctx api.StreamContext, configs map[string]an
 func (ps *PortableSource) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Start running portable source %s with datasource %s and conf %+v", ps.symbolName, ps.topic, ps.props)
 	pm := GetPluginInsManager()
-	ins, err := pm.getOrStartProcess(ps.reg, PortbleConf)
+	ins, err := pm.GetOrStartProcess(ps.reg, PortbleConf)
 	if err != nil {
 		return err
 	}
diff --git a/internal/server/connection.go b/internal/server/connection.go
index ee60d7a2..841c2c1c 100644
--- a/internal/server/connection.go
+++ b/internal/server/connection.go
@@ -31,6 +31,16 @@ type ConnectionRequest struct {
 	Props map[string]interface{} `json:"props"`
 }
 
+func connectionsStatusHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	switch r.Method {
+	case http.MethodGet:
+		allStatus := connection.GetAllConnectionStatus(context.Background())
+		w.WriteHeader(http.StatusOK)
+		jsonResponse(allStatus, w, logger)
+	}
+}
+
 func connectionsHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	switch r.Method {
diff --git a/internal/server/portable_init.go b/internal/server/portable_init.go
index f92712c2..81d9fdc7 100644
--- a/internal/server/portable_init.go
+++ b/internal/server/portable_init.go
@@ -28,6 +28,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/plugin"
 	"github.com/lf-edge/ekuiper/v2/internal/plugin/portable"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 )
 
@@ -51,6 +52,7 @@ func (p portableComp) register() {
 func (p portableComp) rest(r *mux.Router) {
 	r.HandleFunc("/plugins/portables", portablesHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/plugins/portables/{name}", portableHandler).Methods(http.MethodGet, http.MethodDelete, http.MethodPut)
+	r.HandleFunc("/plugins/portables/{name}/status", portableStatusHandler).Methods(http.MethodGet)
 }
 
 func (p portableComp) exporter() ConfManager {
@@ -81,6 +83,19 @@ func portablesHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func portableStatusHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+	name := vars["name"]
+	status, ok := runtime.GetPluginInsManager().GetPluginInsStatus(name)
+	if !ok {
+		handleError(w, errorx.NewWithCode(errorx.NOT_FOUND, "not found"), fmt.Sprintf("portable plugin %s not found", name), logger)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	jsonResponse(status, w, logger)
+}
+
 func portableHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	vars := mux.Vars(r)
diff --git a/internal/server/rest.go b/internal/server/rest.go
index 069cb6a8..cf484184 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -177,6 +177,7 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/data/import", configurationImportHandler).Methods(http.MethodPost)
 	r.HandleFunc("/data/import/status", configurationStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/connections/status", connectionsStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	// Register extended routes
diff --git a/internal/topo/lookup/cache/cache.go b/internal/topo/lookup/cache/cache.go
index a050f6b8..45de5750 100644
--- a/internal/topo/lookup/cache/cache.go
+++ b/internal/topo/lookup/cache/cache.go
@@ -19,13 +19,12 @@ import (
 	"sync"
 	"time"
 
-	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 type item struct {
-	data       api.MessageTupleList
+	data       []map[string]any
 	expiration time.Time
 }
 
@@ -76,8 +75,8 @@ func (c *Cache) deleteExpired() {
 	c.Unlock()
 }
 
-func (c *Cache) Set(key string, value api.MessageTupleList) {
-	if (value == nil || value.Len() == 0) && !c.cacheMissingKey {
+func (c *Cache) Set(key string, value []map[string]any) {
+	if len(value) == 0 && !c.cacheMissingKey {
 		return
 	}
 	c.Lock()
@@ -89,7 +88,7 @@ func (c *Cache) Set(key string, value api.MessageTupleList) {
 	}
 }
 
-func (c *Cache) Get(key string) (api.MessageTupleList, bool) {
+func (c *Cache) Get(key string) ([]map[string]any, bool) {
 	c.RLock()
 	defer c.RUnlock()
 	if v, ok := c.items[key]; ok {
diff --git a/internal/topo/lookup/table.go b/internal/topo/lookup/table.go
index 6316280b..e2c12abf 100644
--- a/internal/topo/lookup/table.go
+++ b/internal/topo/lookup/table.go
@@ -77,12 +77,12 @@ func CreateInstance(name string, sourceType string, options *ast.Options) error
 		return err
 	}
 	ctx.GetLogger().Debugf("lookup source %s is created", sourceType)
-	err = ns.Configure(options.DATASOURCE, props)
+	err = ns.Provision(ctx, props)
 	if err != nil {
 		return err
 	}
 	ctx.GetLogger().Debugf("lookup source %s is configured", sourceType)
-	err = ns.Open(ctx)
+	err = ns.Connect(ctx)
 	if err != nil {
 		return err
 	}
diff --git a/internal/topo/node/batch_op.go b/internal/topo/node/batch_op.go
index b5523b17..2c40535d 100644
--- a/internal/topo/node/batch_op.go
+++ b/internal/topo/node/batch_op.go
@@ -21,6 +21,7 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
@@ -57,30 +58,35 @@ func (b *BatchOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	b.prepareExec(ctx, errCh, "op")
 	switch {
 	case b.batchSize > 0 && b.lingerInterval > 0:
-		b.runWithTickerAndBatchSize(ctx)
+		b.runWithTickerAndBatchSize(ctx, errCh)
 	case b.batchSize > 0 && b.lingerInterval == 0:
-		b.runWithBatchSize(ctx)
+		b.runWithBatchSize(ctx, errCh)
 	case b.batchSize == 0 && b.lingerInterval > 0:
-		b.runWithTicker(ctx)
+		b.runWithTicker(ctx, errCh)
 	}
 }
 
-func (b *BatchOp) runWithTickerAndBatchSize(ctx api.StreamContext) {
+func (b *BatchOp) runWithTickerAndBatchSize(ctx api.StreamContext, errCh chan<- error) {
 	ticker := timex.GetTicker(b.lingerInterval)
 	go func() {
-		defer func() {
-			ticker.Stop()
-			b.Close()
-		}()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case d := <-b.input:
-				b.ingest(ctx, d, true)
-			case <-ticker.C:
-				b.send()
+		err := infra.SafeRun(func() error {
+			defer func() {
+				ticker.Stop()
+				b.Close()
+			}()
+			for {
+				select {
+				case <-ctx.Done():
+					return nil
+				case d := <-b.input:
+					b.ingest(ctx, d, true)
+				case <-ticker.C:
+					b.send()
+				}
 			}
+		})
+		if err != nil {
+			infra.DrainError(ctx, err, errCh)
 		}
 	}()
 }
@@ -123,38 +129,48 @@ func (b *BatchOp) send() {
 	b.currIndex = 0
 }
 
-func (b *BatchOp) runWithBatchSize(ctx api.StreamContext) {
+func (b *BatchOp) runWithBatchSize(ctx api.StreamContext, errCh chan<- error) {
 	go func() {
-		defer func() {
-			b.Close()
-		}()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case d := <-b.input:
-				b.ingest(ctx, d, true)
+		err := infra.SafeRun(func() error {
+			defer func() {
+				b.Close()
+			}()
+			for {
+				select {
+				case <-ctx.Done():
+					return nil
+				case d := <-b.input:
+					b.ingest(ctx, d, true)
+				}
 			}
+		})
+		if err != nil {
+			infra.DrainError(ctx, err, errCh)
 		}
 	}()
 }
 
-func (b *BatchOp) runWithTicker(ctx api.StreamContext) {
+func (b *BatchOp) runWithTicker(ctx api.StreamContext, errCh chan<- error) {
 	ticker := timex.GetTicker(b.lingerInterval)
 	go func() {
-		defer func() {
-			ticker.Stop()
-			b.Close()
-		}()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case d := <-b.input:
-				b.ingest(ctx, d, false)
-			case <-ticker.C:
-				b.send()
+		err := infra.SafeRun(func() error {
+			defer func() {
+				ticker.Stop()
+				b.Close()
+			}()
+			for {
+				select {
+				case <-ctx.Done():
+					return nil
+				case d := <-b.input:
+					b.ingest(ctx, d, false)
+				case <-ticker.C:
+					b.send()
+				}
 			}
+		})
+		if err != nil {
+			infra.DrainError(ctx, err, errCh)
 		}
 	}()
 }
diff --git a/internal/topo/node/cache_op.go b/internal/topo/node/cache_op.go
index 74a33d24..cedf2e88 100644
--- a/internal/topo/node/cache_op.go
+++ b/internal/topo/node/cache_op.go
@@ -67,57 +67,62 @@ func (s *CacheOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}
 	s.prepareExec(ctx, errCh, "op")
 	go func() {
-		defer func() {
-			s.Close()
-		}()
-		for {
-			select {
-			case <-ctx.Done():
-				s.cache.Flush(ctx)
-				return
-			case d := <-s.input:
-				data, processed := s.commonIngest(ctx, d)
-				if processed {
-					break
-				}
-				// If already have the cache, append this to cache and send the currItem
-				// Otherwise, send out the new data. If blocked, make it currItem
-				s.statManager.IncTotalRecordsIn()
-				s.statManager.ProcessTimeStart()
-
-				if s.hasCache { // already have cache, add current data to cache and send out the cache
-					err := s.cache.AddCache(ctx, data)
-					ctx.GetLogger().Debugf("add data %v to cache", data)
-					if err != nil {
-						s.statManager.IncTotalExceptions(err.Error())
-						s.Broadcast(err)
-						s.statManager.ProcessTimeEnd()
-						s.statManager.IncTotalMessagesProcessed(1)
+		err := infra.SafeRun(func() error {
+			defer func() {
+				s.Close()
+			}()
+			for {
+				select {
+				case <-ctx.Done():
+					s.cache.Flush(ctx)
+					return nil
+				case d := <-s.input:
+					data, processed := s.commonIngest(ctx, d)
+					if processed {
 						break
 					}
-				} else {
-					s.currItem = data
-				}
-				s.send()
+					// If already have the cache, append this to cache and send the currItem
+					// Otherwise, send out the new data. If blocked, make it currItem
+					s.statManager.IncTotalRecordsIn()
+					s.statManager.ProcessTimeStart()
 
-				s.statManager.ProcessTimeEnd()
-				s.statManager.IncTotalMessagesProcessed(1)
-				l := int64(len(s.input) + s.cache.CacheLength)
-				if s.currItem != nil {
-					l += 1
-				}
-				s.statManager.SetBufferLength(l)
-			case <-s.resendTimerCh:
-				ctx.GetLogger().Debugf("ticker is triggered")
-				s.statManager.ProcessTimeStart()
-				s.send()
-				s.statManager.ProcessTimeEnd()
-				l := int64(len(s.input) + s.cache.CacheLength)
-				if s.currItem != nil {
-					l += 1
+					if s.hasCache { // already have cache, add current data to cache and send out the cache
+						err := s.cache.AddCache(ctx, data)
+						ctx.GetLogger().Debugf("add data %v to cache", data)
+						if err != nil {
+							s.statManager.IncTotalExceptions(err.Error())
+							s.Broadcast(err)
+							s.statManager.ProcessTimeEnd()
+							s.statManager.IncTotalMessagesProcessed(1)
+							break
+						}
+					} else {
+						s.currItem = data
+					}
+					s.send()
+
+					s.statManager.ProcessTimeEnd()
+					s.statManager.IncTotalMessagesProcessed(1)
+					l := int64(len(s.input) + s.cache.CacheLength)
+					if s.currItem != nil {
+						l += 1
+					}
+					s.statManager.SetBufferLength(l)
+				case <-s.resendTimerCh:
+					ctx.GetLogger().Debugf("ticker is triggered")
+					s.statManager.ProcessTimeStart()
+					s.send()
+					s.statManager.ProcessTimeEnd()
+					l := int64(len(s.input) + s.cache.CacheLength)
+					if s.currItem != nil {
+						l += 1
+					}
+					s.statManager.SetBufferLength(l)
 				}
-				s.statManager.SetBufferLength(l)
 			}
+		})
+		if err != nil {
+			infra.DrainError(ctx, err, errCh)
 		}
 	}()
 }
diff --git a/internal/topo/node/concurrent.go b/internal/topo/node/concurrent.go
index f15131ba..9625be4f 100644
--- a/internal/topo/node/concurrent.go
+++ b/internal/topo/node/concurrent.go
@@ -64,12 +64,7 @@ func merge(ctx api.StreamContext, node *defaultSinkNode, sendInterval time.Durat
 						continue
 					}
 					node.Broadcast(dd)
-					switch dt := dd.(type) {
-					case error:
-						node.statManager.IncTotalExceptions(dt.Error())
-					default:
-						node.statManager.IncTotalRecordsOut()
-					}
+					node.statManager.IncTotalRecordsOut()
 					if sendInterval > 0 {
 						time.Sleep(sendInterval)
 					}
diff --git a/internal/topo/node/event_window_trigger.go b/internal/topo/node/event_window_trigger.go
index f206e41a..6b18fcc6 100644
--- a/internal/topo/node/event_window_trigger.go
+++ b/internal/topo/node/event_window_trigger.go
@@ -122,9 +122,6 @@ func (o *WindowOperator) execEventWindow(ctx api.StreamContext, inputs []*xsql.T
 				break
 			}
 			switch d := data.(type) {
-			case error:
-				o.Broadcast(d)
-				o.statManager.IncTotalExceptions(d.Error())
 			case *xsql.WatermarkTuple:
 				ctx.GetLogger().Debug("WatermarkTuple", d.GetTimestamp())
 				watermarkTs := d.GetTimestamp()
@@ -225,7 +222,6 @@ func (o *WindowOperator) ingest(ctx api.StreamContext, item any) (any, bool) {
 	}
 	switch d := item.(type) {
 	case error:
-		o.statManager.IncTotalExceptions(d.Error())
 		if o.sendError {
 			o.Broadcast(d)
 		}
diff --git a/internal/topo/node/lookup_node.go b/internal/topo/node/lookup_node.go
index e592c0e2..a12759fa 100644
--- a/internal/topo/node/lookup_node.go
+++ b/internal/topo/node/lookup_node.go
@@ -175,7 +175,7 @@ func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.Row, fv *xsql.Function
 		}
 	}
 	var (
-		r  api.MessageTupleList
+		r  []map[string]any
 		e  error
 		ok bool
 	)
@@ -197,7 +197,7 @@ func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.Row, fv *xsql.Function
 	if e != nil {
 		return e
 	} else {
-		if r != nil && r.Len() == 0 {
+		if len(r) == 0 {
 			if n.joinType == ast.LEFT_JOIN {
 				merged := &xsql.JoinTuple{}
 				merged.AddTuple(d)
@@ -208,23 +208,17 @@ func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.Row, fv *xsql.Function
 			}
 		}
 		if r != nil {
-			r.RangeOfTuples(func(index int, tuple api.MessageTuple) bool {
+			for _, mm := range r {
 				merged := &xsql.JoinTuple{}
 				merged.AddTuple(d)
-				var meta map[string]any
-				if mi, ok := tuple.(api.MetaInfo); ok {
-					meta = mi.AllMeta()
-				}
 				t := &xsql.Tuple{
 					Emitter:   n.name,
-					Message:   tuple.ToMap(),
-					Metadata:  meta,
+					Message:   mm,
 					Timestamp: timex.GetNow(),
 				}
 				merged.AddTuple(t)
 				tuples.Content = append(tuples.Content, merged)
-				return true
-			})
+			}
 		}
 		return nil
 	}
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
index a7512dd0..eb8945e4 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -216,7 +216,6 @@ func (o *defaultSinkNode) commonIngest(ctx api.StreamContext, item any) (any, bo
 	}
 	switch d := item.(type) {
 	case error:
-		o.statManager.IncTotalExceptions(d.Error())
 		if o.sendError {
 			o.Broadcast(d)
 		}
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/node/sink_node.go b/internal/topo/node/sink_node.go
index 9719a895..af71fa1a 100644
--- a/internal/topo/node/sink_node.go
+++ b/internal/topo/node/sink_node.go
@@ -67,62 +67,67 @@ func newSinkNode(ctx api.StreamContext, name string, rOpt def.RuleOption, eoflim
 func (s *SinkNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	s.prepareExec(ctx, errCh, "sink")
 	go func() {
-		err := s.sink.Connect(ctx)
-		if err != nil {
-			infra.DrainError(ctx, err, errCh)
-		}
-		defer func() {
-			s.sink.Close(ctx)
-			s.Close()
-		}()
-		s.currentEof = 0
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case d := <-s.input:
-				data, processed := s.ingest(ctx, d)
-				if processed {
-					break
-				}
+		err := infra.SafeRun(func() error {
+			err := s.sink.Connect(ctx)
+			if err != nil {
+				infra.DrainError(ctx, err, errCh)
+			}
+			defer func() {
+				s.sink.Close(ctx)
+				s.Close()
+			}()
+			s.currentEof = 0
+			for {
+				select {
+				case <-ctx.Done():
+					return nil
+				case d := <-s.input:
+					data, processed := s.ingest(ctx, d)
+					if processed {
+						break
+					}
 
-				s.statManager.IncTotalRecordsIn()
-				s.statManager.ProcessTimeStart()
-				err = s.doCollect(ctx, s.sink, data)
-				if err != nil { // resend handling when enabling cache. Two cases: 1. send to alter queue with resendOUt. 2. retry (blocking) until success or unrecoverable error if resendInterval is set
-					ctx.GetLogger().Error(err)
-					s.statManager.IncTotalExceptions(err.Error())
-					if s.resendOut != nil {
-						s.BroadcastCustomized(data, func(val any) {
-							select {
-							case s.resendOut <- val:
-								// do nothing
-							case <-ctx.Done():
-								// rule stop so stop waiting
-							default:
-								s.statManager.IncTotalExceptions(fmt.Sprintf("buffer full, drop message from %s to resend sink", s.name))
+					s.statManager.IncTotalRecordsIn()
+					s.statManager.ProcessTimeStart()
+					err = s.doCollect(ctx, s.sink, data)
+					if err != nil { // resend handling when enabling cache. Two cases: 1. send to alter queue with resendOUt. 2. retry (blocking) until success or unrecoverable error if resendInterval is set
+						ctx.GetLogger().Error(err)
+						s.statManager.IncTotalExceptions(err.Error())
+						if s.resendOut != nil {
+							s.BroadcastCustomized(data, func(val any) {
+								select {
+								case s.resendOut <- val:
+									// do nothing
+								case <-ctx.Done():
+									// rule stop so stop waiting
+								default:
+									s.statManager.IncTotalExceptions(fmt.Sprintf("buffer full, drop message from %s to resend sink", s.name))
+								}
+							})
+						} else if s.resendInterval > 0 {
+							for err != nil && errorx.IsIOError(err) {
+								timex.Sleep(s.resendInterval)
+								err = s.doCollect(ctx, s.sink, data)
+								ctx.GetLogger().Debugf("resending, got err? %v", err)
+								s.statManager.SetBufferLength(int64(len(s.input)))
+							}
+							if err == nil {
+								s.statManager.IncTotalRecordsOut()
+							} else {
+								ctx.GetLogger().Errorf("resend error %v", err)
 							}
-						})
-					} else if s.resendInterval > 0 {
-						for err != nil && errorx.IsIOError(err) {
-							timex.Sleep(s.resendInterval)
-							err = s.doCollect(ctx, s.sink, data)
-							ctx.GetLogger().Debugf("resending, got err? %v", err)
-							s.statManager.SetBufferLength(int64(len(s.input)))
-						}
-						if err == nil {
-							s.statManager.IncTotalRecordsOut()
-						} else {
-							ctx.GetLogger().Errorf("resend error %v", err)
 						}
+					} else {
+						s.statManager.IncTotalRecordsOut()
 					}
-				} else {
-					s.statManager.IncTotalRecordsOut()
+					s.statManager.ProcessTimeEnd()
+					s.statManager.IncTotalMessagesProcessed(1)
+					s.statManager.SetBufferLength(int64(len(s.input)))
 				}
-				s.statManager.ProcessTimeEnd()
-				s.statManager.IncTotalMessagesProcessed(1)
-				s.statManager.SetBufferLength(int64(len(s.input)))
 			}
+		})
+		if err != nil {
+			infra.DrainError(ctx, err, errCh)
 		}
 	}()
 }
@@ -139,7 +144,6 @@ func (s *SinkNode) ingest(ctx api.StreamContext, item any) (any, bool) {
 	}
 	switch d := item.(type) {
 	case error:
-		s.statManager.IncTotalExceptions(d.Error())
 		if s.sendError {
 			return d, false
 		}
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index 4f9b02e7..79c705ac 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -176,7 +176,10 @@ func (m *SourceNode) Run(ctx api.StreamContext, ctrlCh chan<- error) {
 }
 
 func (m *SourceNode) runPull(ctx api.StreamContext) error {
-	m.doPull(ctx, timex.GetNow())
+	err := m.doPull(ctx, timex.GetNow())
+	if err != nil {
+		return err
+	}
 	ticker := timex.GetTicker(m.interval)
 	go func() {
 		defer ticker.Stop()
@@ -184,7 +187,10 @@ func (m *SourceNode) runPull(ctx api.StreamContext) error {
 			select {
 			case tc := <-ticker.C:
 				ctx.GetLogger().Debugf("source pull at %v", tc.UnixMilli())
-				m.doPull(ctx, tc)
+				e := m.doPull(ctx, tc)
+				if e != nil {
+					m.ingestError(ctx, e)
+				}
 			case <-ctx.Done():
 				return
 			}
@@ -193,13 +199,16 @@ func (m *SourceNode) runPull(ctx api.StreamContext) error {
 	return nil
 }
 
-func (m *SourceNode) doPull(ctx api.StreamContext, tc time.Time) {
-	switch ss := m.s.(type) {
-	case api.PullBytesSource:
-		ss.Pull(ctx, tc, m.ingestBytes, m.ingestError)
-	case api.PullTupleSource:
-		ss.Pull(ctx, tc, m.ingestAnyTuple, m.ingestError)
-	}
+func (m *SourceNode) doPull(ctx api.StreamContext, tc time.Time) error {
+	return infra.SafeRun(func() error {
+		switch ss := m.s.(type) {
+		case api.PullBytesSource:
+			ss.Pull(ctx, tc, m.ingestBytes, m.ingestError)
+		case api.PullTupleSource:
+			ss.Pull(ctx, tc, m.ingestAnyTuple, m.ingestError)
+		}
+		return nil
+	})
 }
 
 func (m *SourceNode) Close() {
diff --git a/internal/topo/node/watermark_op.go b/internal/topo/node/watermark_op.go
index 7b2b83cb..9cced422 100644
--- a/internal/topo/node/watermark_op.go
+++ b/internal/topo/node/watermark_op.go
@@ -113,9 +113,6 @@ func (w *WatermarkOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 						break
 					}
 					switch d := data.(type) {
-					case error:
-						w.Broadcast(d)
-						w.statManager.IncTotalExceptions(d.Error())
 					case *xsql.Tuple:
 						w.statManager.IncTotalRecordsIn()
 						// Start the first event processing.
diff --git a/internal/topo/node/window_op.go b/internal/topo/node/window_op.go
index 5e16d7f9..0eb3ee1e 100644
--- a/internal/topo/node/window_op.go
+++ b/internal/topo/node/window_op.go
@@ -325,9 +325,6 @@ func (o *WindowOperator) execProcessingWindow(ctx api.StreamContext, inputs []*x
 			o.statManager.IncTotalRecordsIn()
 			o.statManager.ProcessTimeStart()
 			switch d := data.(type) {
-			case error:
-				o.Broadcast(d)
-				o.statManager.IncTotalExceptions(d.Error())
 			case *xsql.Tuple:
 				log.Debugf("Event window receive tuple %s", d.Message)
 				inputs = append(inputs, d)
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/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/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index 4cf85c80..3ca6488a 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -59,6 +59,7 @@ jobs:
         cd ../../../../
         cp -r sdk/python/example/pysam plugins/portable/pysam
         cp -r sdk/python/ekuiper plugins/portable/pysam/
+        cp test/lookup.json data/lookup.json
     - name: Run test case
       run: |
         make failpoint-enable
diff --git a/internal/io/connection/connection_pool_test.go b/internal/io/connection/connection_pool_test.go
index 2b4ccd55..f9019bb8 100644
--- a/internal/io/connection/connection_pool_test.go
+++ b/internal/io/connection/connection_pool_test.go
@@ -113,3 +113,27 @@ func TestConnectionErr(t *testing.T) {
 	require.Error(t, err)
 	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/connection/dropConnectionStoreErr")
 }
+
+func TestConnectionStatus(t *testing.T) {
+	dataDir, err := conf.GetDataLoc()
+	require.NoError(t, err)
+	require.NoError(t, store.SetupDefault(dataDir))
+	require.NoError(t, InitConnectionManager4Test())
+
+	conf.WriteCfgIntoKVStorage("connections", "mockErr", "a1", map[string]interface{}{})
+	conf.WriteCfgIntoKVStorage("connections", "mock", "a2", map[string]interface{}{})
+	require.NoError(t, ReloadConnection())
+	ctx := context.Background()
+	allStatus := GetAllConnectionStatus(ctx)
+	s, ok := allStatus["a1"]
+	require.True(t, ok)
+	require.Equal(t, ConnectionStatus{
+		Status: ConnectionFail,
+		ErrMsg: "mockErr",
+	}, s)
+	s, ok = allStatus["a2"]
+	require.True(t, ok)
+	require.Equal(t, ConnectionStatus{
+		Status: ConnectionRunning,
+	}, s)
+}
diff --git a/internal/io/memory/lookupsource_test.go b/internal/io/memory/lookupsource_test.go
new file mode 100644
index 00000000..571e9140
--- /dev/null
+++ b/internal/io/memory/lookupsource_test.go
@@ -0,0 +1,173 @@
+// Copyright 2022-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 memory
+
+import (
+	gocontext "context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func produceUpdatable(ctx api.StreamContext, topic string, data pubsub.MemTuple, rowkind string, keyval any) {
+	pubsub.Produce(ctx, topic, &pubsub.UpdatableTuple{
+		MemTuple: data,
+		Rowkind:  rowkind,
+		Keyval:   keyval,
+	})
+}
+
+func TestLookupProvision(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "Test")
+	tests := []struct {
+		name  string
+		props map[string]any
+		err   string
+	}{
+		{
+			name: "invalid prop type",
+			props: map[string]any{
+				"key": 1,
+			},
+			err: "read properties map[key:1] fail with error: 1 error(s) decoding:\n\n* 'key' expected type 'string', got unconvertible type 'int', value: '1'",
+		},
+		{
+			name: "missing topic",
+			props: map[string]any{
+				"key": "test",
+			},
+			err: "datasource(topic) is required",
+		},
+		{
+			name: "wrong topic regex",
+			props: map[string]any{
+				"key":        "test",
+				"datasource": "test/#/abc",
+			},
+			err: "invalid topic test/#/abc: # must at the last level",
+		},
+		{
+			name: "missing key",
+			props: map[string]any{
+				"datasource": "test/#",
+			},
+			err: "key is required for lookup source",
+		},
+	}
+	ls := &lookupsource{}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := ls.Provision(ctx, tt.props)
+			assert.Error(t, err)
+			assert.EqualError(t, err, tt.err)
+		})
+	}
+}
+
+func TestUpdateLookup(t *testing.T) {
+	contextLogger := conf.Log.WithField("rule", "test")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	ls := GetLookupSource()
+	err := ls.Provision(ctx, map[string]interface{}{"datasource": "test", "key": "ff"})
+	assert.NoError(t, err)
+	err = ls.Connect(ctx)
+	assert.NoError(t, err)
+	// wait for the source to be ready
+	time.Sleep(100 * time.Millisecond)
+	pubsub.Produce(ctx, "test", &xsql.Tuple{Message: map[string]any{"ff": "value1", "gg": "value2"}})
+	produceUpdatable(ctx, "test", &xsql.Tuple{Message: map[string]any{"ff": "value1", "gg": "value2"}}, "delete", "value1")
+	produceUpdatable(ctx, "test", &xsql.Tuple{Message: map[string]any{"ff": "value2", "gg": "value2"}}, "insert", "value2")
+	pubsub.Produce(ctx, "test", &xsql.Tuple{Message: map[string]any{"ff": "value1", "gg": "value4"}})
+	produceUpdatable(ctx, "test", &xsql.Tuple{Message: map[string]any{"ff": "value2", "gg": "value2"}}, "delete", "value2")
+	pubsub.Produce(ctx, "test", &xsql.Tuple{Message: map[string]any{"ff": "value1", "gg": "value2"}})
+	pubsub.Produce(ctx, "test", &xsql.Tuple{Message: map[string]any{"ff": "value2", "gg": "value2"}})
+	// wait for table accumulation
+	time.Sleep(100 * time.Millisecond)
+	canctx, cancel := gocontext.WithCancel(gocontext.Background())
+	defer cancel()
+	go func() {
+		for {
+			select {
+			case <-canctx.Done():
+				return
+			case <-time.After(10 * time.Millisecond):
+				pubsub.Produce(ctx, "test", &xsql.Tuple{Message: map[string]any{"ff": "value4", "gg": "value2"}})
+			}
+		}
+	}()
+	expected := []map[string]any{
+		{"ff": "value1", "gg": "value2"},
+	}
+	result, err := ls.Lookup(ctx, []string{}, []string{"ff"}, []interface{}{"value1"})
+	assert.NoError(t, err)
+	assert.Equal(t, expected, result)
+	err = ls.Close(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+}
+
+func TestLookup(t *testing.T) {
+	contextLogger := conf.Log.WithField("rule", "test2")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	ls := GetLookupSource()
+	err := ls.Provision(ctx, map[string]any{"datasource": "test2", "key": "gg"})
+	assert.NoError(t, err)
+	err = ls.Connect(ctx)
+	assert.NoError(t, err)
+	// wait for the source to be ready
+	time.Sleep(100 * time.Millisecond)
+	pubsub.Produce(ctx, "test2", &xsql.Tuple{Message: map[string]any{"ff": "value1", "gg": "value2"}})
+	pubsub.Produce(ctx, "test2", &xsql.Tuple{Message: map[string]any{"ff": "value2", "gg": "value3"}})
+	pubsub.Produce(ctx, "test2", &xsql.Tuple{Message: map[string]any{"ff": "value1", "gg": "value4"}})
+	// wait for table accumulation
+	time.Sleep(100 * time.Millisecond)
+	canctx, cancel := gocontext.WithCancel(gocontext.Background())
+	defer cancel()
+	go func() {
+		for {
+			select {
+			case <-canctx.Done():
+				return
+			case <-time.After(10 * time.Millisecond):
+				pubsub.Produce(ctx, "test", &xsql.Tuple{Message: map[string]any{"ff": "value4", "gg": "value5"}})
+			}
+		}
+	}()
+	result, _ := ls.Lookup(ctx, []string{}, []string{"ff"}, []any{"value1"})
+	expected := []map[string]any{
+		{"ff": "value1", "gg": "value2"},
+		{"ff": "value1", "gg": "value4"},
+	}
+	if len(result) != 2 {
+		t.Errorf("expect %v but got %v", expected, result)
+	} else {
+		if result[0]["gg"] != "value2" {
+			result[0], result[1] = result[1], result[0]
+		}
+	}
+	assert.Equal(t, expected, result)
+	err = ls.Close(ctx)
+	assert.NoError(t, err)
+}
diff --git a/internal/io/memory/memory_test.go b/internal/io/memory/memory_test.go
index 6afcbd93..db095e20 100644
--- a/internal/io/memory/memory_test.go
+++ b/internal/io/memory/memory_test.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
+// Copyright 2022-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.
@@ -74,7 +74,7 @@ func TestSharedInmemoryNode(t *testing.T) {
 		}
 	}()
 	err = src.Subscribe(ctx, func(ctx api.StreamContext, res any, meta map[string]any, ts time.Time) {
-		expected := []*xsql.Tuple{{
+		expected := []pubsub.MemTuple{&xsql.Tuple{
 			Emitter:   "",
 			Timestamp: timex.GetNow(),
 			Metadata:  map[string]any{"topic": id},
@@ -87,6 +87,130 @@ func TestSharedInmemoryNode(t *testing.T) {
 	<-ctx.Done()
 }
 
+func TestUpdateListInmemoryNode(t *testing.T) {
+	mockclock.ResetClock(100)
+	pubsub.Reset()
+	id := "test_id"
+	sinkProps := map[string]any{
+		"rowkindField": "update",
+		"keyField":     "id",
+	}
+	sinkProps[pubsub.IdProperty] = id
+	src := GetSource()
+	snk := GetSink()
+	contextLogger := conf.Log.WithField("rule", "test")
+	ctx1 := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	ctx, cancel := ctx1.WithCancel()
+	srcProps := make(map[string]interface{})
+	srcProps["option"] = "value"
+	err := snk.Provision(ctx, sinkProps)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = snk.Connect(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	srcProps[pubsub.IdProperty] = id
+	srcProps["datasource"] = id
+	err = src.Provision(ctx, srcProps)
+	if err != nil {
+		t.Error(err)
+	}
+
+	rawTuple := &xsql.Tuple{
+		Message:  map[string]any{"temp": 20, "id": 1, "update": "update"},
+		Metadata: nil,
+	}
+	mockclock.GetMockClock().Add(100)
+	go func() {
+		err = snk.CollectList(ctx, &xsql.TransformedTupleList{Content: []api.MessageTuple{rawTuple}})
+		if err != nil {
+			t.Error(err)
+		}
+	}()
+	err = src.Subscribe(ctx, func(ctx api.StreamContext, res any, meta map[string]any, ts time.Time) {
+		expected := []pubsub.MemTuple{&pubsub.UpdatableTuple{
+			MemTuple: &xsql.Tuple{
+				Emitter:   "",
+				Timestamp: timex.GetNow(),
+				Metadata:  map[string]any{"topic": id},
+				Message:   rawTuple.Message,
+			},
+			Rowkind: "update",
+			Keyval:  1,
+		}}
+		assert.Equal(t, expected, res)
+		cancel()
+	}, nil)
+	assert.NoError(t, err)
+	<-ctx.Done()
+}
+
+func TestUpdateInmemoryNode(t *testing.T) {
+	mockclock.ResetClock(100)
+	pubsub.Reset()
+	id := "test_id"
+	sinkProps := map[string]any{
+		"rowkindField": "update",
+		"keyField":     "id",
+	}
+	sinkProps[pubsub.IdProperty] = id
+	src := GetSource()
+	snk := GetSink()
+	contextLogger := conf.Log.WithField("rule", "test")
+	ctx1 := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	ctx, cancel := ctx1.WithCancel()
+	srcProps := make(map[string]interface{})
+	srcProps["option"] = "value"
+	err := snk.Provision(ctx, sinkProps)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = snk.Connect(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	srcProps[pubsub.IdProperty] = id
+	srcProps["datasource"] = id
+	err = src.Provision(ctx, srcProps)
+	if err != nil {
+		t.Error(err)
+	}
+
+	rawTuple := &xsql.Tuple{
+		Message:  map[string]any{"temp": 20, "id": 1, "update": "delete"},
+		Metadata: nil,
+	}
+	mockclock.GetMockClock().Add(100)
+	go func() {
+		err = snk.Collect(ctx, rawTuple)
+		if err != nil {
+			t.Error(err)
+		}
+	}()
+	err = src.Subscribe(ctx, func(ctx api.StreamContext, res any, meta map[string]any, ts time.Time) {
+		expected := &pubsub.UpdatableTuple{
+			MemTuple: &xsql.Tuple{
+				Emitter:   "",
+				Timestamp: timex.GetNow(),
+				Metadata:  map[string]any{"topic": id},
+				Message:   rawTuple.Message,
+			},
+			Rowkind: "delete",
+			Keyval:  1,
+		}
+		assert.Equal(t, expected, res)
+		cancel()
+	}, nil)
+	assert.NoError(t, err)
+	<-ctx.Done()
+}
+
 func TestMultipleTopics(t *testing.T) {
 	pubsub.Reset()
 	var (
diff --git a/internal/io/memory/sink_test.go b/internal/io/memory/sink_test.go
new file mode 100644
index 00000000..7e3dc9f0
--- /dev/null
+++ b/internal/io/memory/sink_test.go
@@ -0,0 +1,70 @@
+// 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 memory
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestWrapUpdatable(t *testing.T) {
+	tests := []struct {
+		name  string
+		value map[string]any
+		err   string
+	}{
+		{
+			name: "empty rowkind with wrong key",
+			value: map[string]any{
+				"nokey": 100,
+			},
+			err: "key field id not found in data &{ map[nokey:100] 0001-01-01 00:00:00 +0000 UTC map[] map[] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}",
+		},
+		{
+			name: "wrong rowkind type",
+			value: map[string]any{
+				"rowkind": 100,
+			},
+			err: "rowkind field rowkind is not a string in data &{ map[rowkind:100] 0001-01-01 00:00:00 +0000 UTC map[] map[] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}",
+		},
+		{
+			name: "wrong rowkind value",
+			value: map[string]any{
+				"rowkind": "test",
+			},
+			err: "invalid rowkind test",
+		},
+	}
+	s := &sink{}
+	ctx := mockContext.NewMockContext("rule1", "test")
+	err := s.Provision(ctx, map[string]any{
+		"rowkindField": "rowkind",
+		"keyField":     "id",
+	})
+	assert.NoError(t, err)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := s.wrapUpdatable(&xsql.Tuple{
+				Message: tt.value,
+			})
+			assert.Error(t, err)
+			assert.EqualError(t, err, tt.err)
+		})
+	}
+}
diff --git a/internal/io/memory/store/db_test.go b/internal/io/memory/store/db_test.go
new file mode 100644
index 00000000..75d12039
--- /dev/null
+++ b/internal/io/memory/store/db_test.go
@@ -0,0 +1,134 @@
+// Copyright 2022-2023 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 store
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+)
+
+func TestTable(t *testing.T) {
+	tb := createTable("topicT", "a")
+	tb.add(&xsql.Tuple{Message: map[string]interface{}{"a": 1, "b": "0"}})
+	tb.add(&xsql.Tuple{Message: map[string]interface{}{"a": 2, "b": "0"}})
+	tb.add(&xsql.Tuple{Message: map[string]interface{}{"a": 3, "b": "4"}})
+	tb.add(&xsql.Tuple{Message: map[string]interface{}{"a": 1, "b": "1"}})
+	v, _ := tb.Read([]string{"a"}, []interface{}{1})
+	exp := []pubsub.MemTuple{
+		&xsql.Tuple{Message: map[string]interface{}{"a": 1, "b": "1"}},
+	}
+	if !reflect.DeepEqual(v, exp) {
+		t.Errorf("read a 1 expect %v, but got %v", exp, v)
+		return
+	}
+	v, _ = tb.Read([]string{"b"}, []interface{}{"0"})
+	exp = []pubsub.MemTuple{
+		&xsql.Tuple{Message: map[string]interface{}{"a": 2, "b": "0"}},
+	}
+	if !reflect.DeepEqual(v, exp) {
+		t.Errorf("read b 0 expect %v, but got %v", exp, v)
+		return
+	}
+	tb.add(&xsql.Tuple{Message: map[string]interface{}{"a": 5, "b": "0"}})
+	tb.delete(3)
+	tb.add(&xsql.Tuple{Message: map[string]interface{}{"a": 1, "b": "1"}})
+	v, _ = tb.Read([]string{"b"}, []interface{}{"0"})
+	exp = []pubsub.MemTuple{
+		&xsql.Tuple{Message: map[string]interface{}{"a": 2, "b": "0"}},
+		&xsql.Tuple{Message: map[string]interface{}{"a": 5, "b": "0"}},
+	}
+	if len(v) != 2 {
+		t.Errorf("read 1 again expect %v, but got %v", exp, v)
+		return
+	} else {
+		vv, _ := v[0].Value("a", "")
+		if vv != 2 {
+			v[0], v[1] = v[1], v[0]
+		}
+		if !reflect.DeepEqual(v, exp) {
+			t.Errorf("read 1 again expect %v, but got %v", exp, v)
+			return
+		}
+	}
+
+	v, _ = tb.Read([]string{"a", "b"}, []interface{}{1, "1"})
+	exp = []pubsub.MemTuple{
+		&xsql.Tuple{Message: map[string]interface{}{"a": 1, "b": "1"}},
+	}
+	if !reflect.DeepEqual(v, exp) {
+		t.Errorf("read a,b expect %v, but got %v", exp, v)
+		return
+	}
+	v, _ = tb.Read([]string{"a"}, []interface{}{3})
+	if v != nil {
+		t.Errorf("read a 3 expect nil, but got %v", v)
+		return
+	}
+	tb.delete(1)
+	v, _ = tb.Read([]string{"a"}, []interface{}{1})
+	if v != nil {
+		t.Errorf("read a 1 expect nil, but got %v", v)
+	}
+}
+
+func TestDb(t *testing.T) {
+	db = &database{
+		tables: make(map[string]*tableCount),
+	}
+	db.addTable("t1", "a")
+	db.addTable("t1", "b")
+	db.addTable("t2", "a")
+	db.addTable("t1", "a")
+	_, ok := db.getTable("t1", "a")
+	if !ok {
+		t.Errorf("table t1 a should exist")
+		return
+	}
+	_, ok = db.getTable("t1", "b")
+	if !ok {
+		t.Errorf("table t1 b should exist")
+		return
+	}
+	_, ok = db.getTable("t3", "a")
+	if ok {
+		t.Errorf("table t1 c should not exist")
+		return
+	}
+	tc := db.tables["t1_a"]
+	if tc.count != 2 {
+		t.Errorf("table t1 a should have 2 instances but got %d", tc.count)
+		return
+	}
+	tc = db.tables["t2_a"]
+	if tc.count != 1 {
+		t.Errorf("table t2 a should have 1 instances")
+		return
+	}
+	db.dropTable("t1", "a")
+	db.dropTable("t2", "a")
+	_, ok = db.getTable("t2", "a")
+	if ok {
+		t.Errorf("table t2 a should not exist")
+		return
+	}
+	tc = db.tables["t1_a"]
+	if tc.count != 1 {
+		t.Errorf("table t1 a should have 1 instances but got %d", tc.count)
+		return
+	}
+}
diff --git a/internal/io/memory/store/store_test.go b/internal/io/memory/store/store_test.go
new file mode 100644
index 00000000..fd53313e
--- /dev/null
+++ b/internal/io/memory/store/store_test.go
@@ -0,0 +1,57 @@
+// Copyright 2022-2023 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 store
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestReg(t *testing.T) {
+	db = &database{
+		tables: make(map[string]*tableCount),
+	}
+	reg1, err := Reg("test", nil, "a")
+	if err != nil {
+		t.Errorf("register test error: %v", err)
+		return
+	}
+	_, err2 := Reg("test", nil, "a")
+	if err2 != nil {
+		t.Errorf("register test error: %v", err2)
+		return
+	}
+	exp := map[string]*tableCount{
+		"test_a": {
+			count: 2,
+			t:     reg1,
+		},
+	}
+	if !reflect.DeepEqual(exp, db.tables) {
+		t.Errorf("register expect %v, but got %v", exp, db.tables)
+		return
+	}
+	Unreg("test", "a")
+	exp = map[string]*tableCount{
+		"test_a": {
+			count: 1,
+			t:     reg1,
+		},
+	}
+	if !reflect.DeepEqual(exp, db.tables) {
+		t.Errorf("unregister expect %v, but got %v", exp, db.tables)
+		return
+	}
+}
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager_test.go b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
index 4796055c..7fce1e96 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager_test.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
@@ -15,10 +15,12 @@
 package runtime
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"testing"
 
+	"github.com/stretchr/testify/require"
 	"go.nanomsg.org/mangos/v3"
 	"go.nanomsg.org/mangos/v3/protocol/req"
 
@@ -184,3 +186,14 @@ func createMockClient(pluginName string) (mangos.Socket, error) {
 	}
 	return sock, nil
 }
+
+func TestPluginStatus(t *testing.T) {
+	p := NewPluginIns("mock", nil, nil)
+	require.Equal(t, PluginStatusInit, p.GetStatus().Status)
+	p.Status.StartRunning()
+	require.Equal(t, PluginStatusRunning, p.GetStatus().Status)
+	p.Status.StatusErr(errors.New("mock"))
+	require.Equal(t, PluginStatusErr, p.GetStatus().Status)
+	p.Status.Stop()
+	require.Equal(t, PluginStatusStop, p.GetStatus().Status)
+}
diff --git a/internal/plugin/portable/runtime/plugin_ins_test.go b/internal/plugin/portable/runtime/plugin_ins_test.go
new file mode 100644
index 00000000..97f11ef3
--- /dev/null
+++ b/internal/plugin/portable/runtime/plugin_ins_test.go
@@ -0,0 +1,75 @@
+package runtime_test
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime"
+)
+
+func TestProcess(t *testing.T) {
+	pm := runtime.GetPluginInsManager4Test()
+	conf.InitConf()
+	dir, _ := os.Getwd()
+	p := filepath.Join(dir, "../../../../", "sdk/python/example/pysam/pysam.py")
+	meta := &runtime.PluginMeta{
+		Name:       "pysam",
+		Version:    "1.0.0",
+		Language:   "python",
+		Executable: p,
+	}
+	_, err := pm.GetOrStartProcess(meta, runtime.PortbleConf)
+	require.NoError(t, err)
+	err = pm.Kill("pysam")
+	require.NoError(t, err)
+}
+
+func TestProcessErr(t *testing.T) {
+	pm := runtime.GetPluginInsManager4Test()
+	conf.InitConf()
+	dir, _ := os.Getwd()
+	p := filepath.Join(dir, "../../../../", "sdk/python/example/pysam/pysam.py")
+	meta := &runtime.PluginMeta{
+		Name:       "pysam",
+		Version:    "1.0.0",
+		Language:   "python",
+		Executable: p,
+	}
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime/CreateControlChannelErr", "return(true)")
+	_, err := pm.GetOrStartProcess(meta, runtime.PortbleConf)
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime/CreateControlChannelErr")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime/confErr", "return(true)")
+	_, err = pm.GetOrStartProcess(meta, runtime.PortbleConf)
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime/confErr")
+
+	x := "mock"
+	_, err = pm.GetOrStartProcess(&runtime.PluginMeta{
+		Name:        "pysam",
+		Version:     "1.0.0",
+		Language:    "python",
+		Executable:  p,
+		VirtualType: &x,
+	}, runtime.PortbleConf)
+	require.Error(t, err)
+
+	_, err = pm.GetOrStartProcess(&runtime.PluginMeta{
+		Name:       "pysam",
+		Version:    "1.0.0",
+		Language:   "mock",
+		Executable: p,
+	}, runtime.PortbleConf)
+	require.Error(t, err)
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime/cmdStartErr", "return(true)")
+	_, err = pm.GetOrStartProcess(meta, runtime.PortbleConf)
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/plugin/portable/runtime/cmdStartErr")
+}
diff --git a/internal/processor/stream_test.go b/internal/processor/stream_test.go
index 726a7939..8101a366 100644
--- a/internal/processor/stream_test.go
+++ b/internal/processor/stream_test.go
@@ -216,6 +216,8 @@ func TestTableProcessor(t *testing.T) {
 func TestTableList(t *testing.T) {
 	p := NewStreamProcessor()
 	p.ExecStmt(`CREATE TABLE tt1 () WITH (DATASOURCE="users", FORMAT="JSON", KIND="scan")`)
+	p.ExecStmt(`CREATE TABLE tt2 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KEY="id", KIND="lookup")`)
+	p.ExecStmt(`CREATE TABLE tt3 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KEY="id", KIND="lookup")`)
 	p.ExecStmt(`CREATE TABLE tt4 () WITH (DATASOURCE="users", FORMAT="JSON")`)
 	defer func() {
 		p.ExecStmt(`DROP TABLE tt1`)
@@ -228,7 +230,7 @@ func TestTableList(t *testing.T) {
 		t.Errorf("Show lookup table fails: %s", err)
 		return
 	}
-	le := []string{}
+	le := []string{"tt2", "tt3"}
 	if !reflect.DeepEqual(le, la) {
 		t.Errorf("Show lookup table mismatch:\nexp=%s\ngot=%s", le, la)
 		return
@@ -255,6 +257,7 @@ func TestAll(t *testing.T) {
 		},
 		"tables": {
 			"tt1": `CREATE TABLE tt1 () WITH (DATASOURCE="users", FORMAT="JSON", KIND="scan")`,
+			"tt3": `CREATE TABLE tt3 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KEY="id", KIND="lookup")`,
 		},
 	}
 	p := NewStreamProcessor()
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 7664c19e..29162df0 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -125,13 +125,23 @@ func (suite *RestTestSuite) Test_sourcesManageHandler() {
 	suite.r.ServeHTTP(w, req)
 	assert.Equal(suite.T(), http.StatusOK, w.Code)
 
+	// create table
+	buf := bytes.NewBuffer([]byte(` {"sql":"CREATE TABLE alertTable() WITH (DATASOURCE=\"0\", TYPE=\"memory\", KEY=\"id\", KIND=\"lookup\")"}`))
+	req, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/streams?kind=lookup", buf)
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusCreated, w.Code)
+	var returnVal []byte
+	returnVal, _ = io.ReadAll(w.Result().Body)
+	fmt.Printf("returnVal %s\n", string(returnVal))
+
 	// create stream
-	buf := bytes.NewBuffer([]byte(`{"sql":"CREATE stream alert() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
+	buf = bytes.NewBuffer([]byte(`{"sql":"CREATE stream alert() WITH (DATASOURCE=\"0\", TYPE=\"mqtt\")"}`))
 	req, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/streams", buf)
 	w = httptest.NewRecorder()
 	suite.r.ServeHTTP(w, req)
 	assert.Equal(suite.T(), http.StatusCreated, w.Code)
-	returnVal, _ := io.ReadAll(w.Result().Body)
+	returnVal, _ = io.ReadAll(w.Result().Body)
 	fmt.Printf("returnVal %s\n", string(returnVal))
 
 	// get stream
@@ -151,6 +161,30 @@ func (suite *RestTestSuite) Test_sourcesManageHandler() {
 	suite.r.ServeHTTP(w, req)
 	assert.Equal(suite.T(), http.StatusOK, w.Code)
 
+	// get table
+	req, _ = http.NewRequest(http.MethodGet, "http://localhost:8080/tables/alertTable", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+	expect = []byte(`{"Name":"alertTable","Options":{"datasource":"0","type":"memory", "key":"id","kind":"lookup"},"Statement":null,"StreamFields":null,"StreamType":1}`)
+	exp = map[string]interface{}{}
+	_ = json.NewDecoder(bytes.NewBuffer(expect)).Decode(&exp)
+	res = map[string]interface{}{}
+	_ = json.NewDecoder(w.Result().Body).Decode(&res)
+	assert.Equal(suite.T(), exp, res)
+
+	req, _ = http.NewRequest(http.MethodGet, "http://localhost:8080/tables/alertTable/schema", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+
+	// put table
+	buf = bytes.NewBuffer([]byte(` {"sql":"CREATE TABLE alertTable() WITH (DATASOURCE=\"0\", TYPE=\"memory\", KEY=\"id\", KIND=\"lookup\")"}`))
+	req, _ = http.NewRequest(http.MethodPut, "http://localhost:8080/tables/alertTable", buf)
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+
 	// put stream
 	buf = bytes.NewBuffer([]byte(`{"sql":"CREATE stream alert() WITH (DATASOURCE=\"0\", TYPE=\"httppull\")"}`))
 	req, _ = http.NewRequest(http.MethodPut, "http://localhost:8080/streams/alert", buf)
@@ -158,6 +192,12 @@ func (suite *RestTestSuite) Test_sourcesManageHandler() {
 	suite.r.ServeHTTP(w, req)
 	assert.Equal(suite.T(), http.StatusOK, w.Code)
 
+	// drop table
+	req, _ = http.NewRequest(http.MethodDelete, "http://localhost:8080/tables/alertTable", bytes.NewBufferString("any"))
+	w = httptest.NewRecorder()
+	suite.r.ServeHTTP(w, req)
+	assert.Equal(suite.T(), http.StatusOK, w.Code)
+
 	// drop stream
 	req, _ = http.NewRequest(http.MethodDelete, "http://localhost:8080/streams/alert", bytes.NewBufferString("any"))
 	w = httptest.NewRecorder()
diff --git a/internal/topo/lookup/cache/cache_test.go b/internal/topo/lookup/cache/cache_test.go
index 335034ad..cccd16d9 100644
--- a/internal/topo/lookup/cache/cache_test.go
+++ b/internal/topo/lookup/cache/cache_test.go
@@ -19,10 +19,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
-	"github.com/lf-edge/ekuiper/v2/internal/xsql"
-	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 func TestExpiration(t *testing.T) {
@@ -30,10 +27,10 @@ func TestExpiration(t *testing.T) {
 	clock := mockclock.GetMockClock()
 	c := NewCache(20*time.Second, false)
 	defer c.Close()
-	expects := []api.MessageTupleList{
-		&xsql.TransformedTupleList{Content: []api.MessageTuple{&xsql.Tuple{Message: map[string]interface{}{"a": 1}, Timestamp: timex.GetNow()}}},
-		&xsql.TransformedTupleList{Content: []api.MessageTuple{&xsql.Tuple{Message: map[string]interface{}{"a": 2}, Timestamp: timex.GetNow()}, &xsql.Tuple{Message: map[string]interface{}{"a": 3}, Timestamp: timex.GetNow()}}},
-		&xsql.TransformedTupleList{Content: []api.MessageTuple{}},
+	expects := [][]map[string]any{
+		{{"a": 1}},
+		{{"a": 2}, {"a": 3}},
+		{},
 	}
 	// wait for cache to run
 	time.Sleep(10 * time.Millisecond)
@@ -94,10 +91,10 @@ func TestNoExpiration(t *testing.T) {
 	c := NewCache(0, true)
 	defer c.Close()
 
-	expects := []api.MessageTupleList{
-		&xsql.TransformedTupleList{Content: []api.MessageTuple{&xsql.Tuple{Message: map[string]interface{}{"a": 1}, Timestamp: timex.GetNow()}}},
-		&xsql.TransformedTupleList{Content: []api.MessageTuple{&xsql.Tuple{Message: map[string]interface{}{"a": 2}, Timestamp: timex.GetNow()}, &xsql.Tuple{Message: map[string]interface{}{"a": 3}, Timestamp: timex.GetNow()}}},
-		&xsql.TransformedTupleList{Content: []api.MessageTuple{}},
+	expects := [][]map[string]any{
+		{{"a": 1}},
+		{{"a": 2}, {"a": 3}},
+		{},
 	}
 	c.Set("a", expects[0])
 	clock.Add(10 * time.Second)
diff --git a/internal/topo/lookup/table_test.go b/internal/topo/lookup/table_test.go
new file mode 100644
index 00000000..b07bfd17
--- /dev/null
+++ b/internal/topo/lookup/table_test.go
@@ -0,0 +1,101 @@
+// Copyright 2022 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 lookup
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+func TestTable(t *testing.T) {
+	var wg sync.WaitGroup
+	wg.Add(2)
+	err := CreateInstance("test1", "memory", &ast.Options{
+		DATASOURCE: "test",
+		TYPE:       "memory",
+		KIND:       "lookup",
+		KEY:        "id",
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = CreateInstance("test2", "memory", &ast.Options{
+		DATASOURCE: "test2",
+		TYPE:       "memory",
+		KIND:       "lookup",
+		KEY:        "id",
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	go func() {
+		for i := 0; i < 3; i++ {
+			_, err := Attach("test1")
+			if err != nil {
+				t.Error(err)
+			}
+			_, err = Attach("test2")
+			if err != nil {
+				t.Error(err)
+			}
+		}
+		wg.Done()
+	}()
+	go func() {
+		for i := 0; i < 3; i++ {
+			_, err := Attach("test2")
+			if err != nil {
+				t.Error(err)
+			}
+			_, err = Attach("test1")
+			if err != nil {
+				t.Error(err)
+			}
+		}
+		wg.Done()
+	}()
+	wg.Wait()
+	if len(instances) != 2 {
+		t.Errorf("expect 2 instances, but got %d", len(instances))
+		return
+	}
+	for _, ins := range instances {
+		if ins.count != 6 {
+			t.Errorf("expect 6 count, but got %d", ins.count)
+			return
+		}
+	}
+	err = DropInstance("test1")
+	if err == nil {
+		t.Error("should have error to drop instance")
+		return
+	}
+	for i := 0; i < 6; i++ {
+		Detach("test1")
+	}
+	err = DropInstance("test1")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if len(instances) != 1 {
+		t.Errorf("expect 2 instances, but got %d", len(instances))
+		return
+	}
+}
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,
diff --git a/internal/topo/topotest/mock_topo.go b/internal/topo/topotest/mock_topo.go
index bc082edd..565e01e1 100644
--- a/internal/topo/topotest/mock_topo.go
+++ b/internal/topo/topotest/mock_topo.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.
@@ -55,14 +55,13 @@ func CommonResultFunc(result []any) [][]map[string]any {
 	maps := make([][]map[string]any, 0, len(result))
 	for _, v := range result {
 		switch rt := v.(type) {
-		case *xsql.Tuple:
-			m, _ := rt.All("")
+		case pubsub.MemTuple:
+			m := rt.ToMap()
 			maps = append(maps, []map[string]any{m})
-		case []*xsql.Tuple:
+		case []pubsub.MemTuple:
 			nm := make([]map[string]any, 0, len(rt))
 			for _, mm := range rt {
-				m, _ := mm.All("")
-				nm = append(nm, m)
+				nm = append(nm, mm.ToMap())
 			}
 			maps = append(maps, nm)
 		default:
diff --git a/internal/topo/topotest/rule_test.go b/internal/topo/topotest/rule_test.go
index 03bda7a5..2d438d6a 100644
--- a/internal/topo/topotest/rule_test.go
+++ b/internal/topo/topotest/rule_test.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.
@@ -769,7 +769,7 @@ func TestSingleSQL(t *testing.T) {
 				}},
 			},
 			M: map[string]interface{}{
-				"sink_memory_0_0_exceptions_total":  int64(2),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(4),
 				"sink_memory_0_0_records_out_total": int64(4),
 
@@ -1354,12 +1354,12 @@ func TestSingleSQLError(t *testing.T) {
 				}},
 			},
 			M: map[string]interface{}{
-				"op_3_project_0_exceptions_total":   int64(1),
+				"op_3_project_0_exceptions_total":   int64(0),
 				"op_3_project_0_process_latency_us": int64(0),
 				"op_3_project_0_records_in_total":   int64(2),
 				"op_3_project_0_records_out_total":  int64(2),
 
-				"sink_memory_0_0_exceptions_total":  int64(1),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(3),
 				"sink_memory_0_0_records_out_total": int64(3),
 
@@ -1397,7 +1397,7 @@ func TestSingleSQLError(t *testing.T) {
 				"op_2_project_0_records_in_total":   int64(5),
 				"op_2_project_0_records_out_total":  int64(4),
 
-				"sink_memory_0_0_exceptions_total":  int64(1),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(5),
 				"sink_memory_0_0_records_out_total": int64(5),
 
@@ -1430,7 +1430,7 @@ func TestSingleSQLError(t *testing.T) {
 				"op_2_project_0_records_in_total":   int64(5),
 				"op_2_project_0_records_out_total":  int64(4),
 
-				"sink_memory_0_0_exceptions_total":  int64(1),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(5),
 				"sink_memory_0_0_records_out_total": int64(5),
 
diff --git a/internal/topo/topotest/window_rule_test.go b/internal/topo/topotest/window_rule_test.go
index 1fe54db9..344723e5 100644
--- a/internal/topo/topotest/window_rule_test.go
+++ b/internal/topo/topotest/window_rule_test.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.
@@ -430,7 +430,7 @@ func TestWindow(t *testing.T) {
 				}},
 			},
 			M: map[string]interface{}{
-				"sink_memory_0_0_exceptions_total":  int64(3),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(6),
 				"sink_memory_0_0_records_out_total": int64(6),
 
@@ -1041,7 +1041,7 @@ func TestEventWindow(t *testing.T) {
 				}},
 			},
 			M: map[string]interface{}{
-				"sink_memory_0_0_exceptions_total":  int64(1),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(6),
 				"sink_memory_0_0_records_out_total": int64(6),
 
@@ -1049,7 +1049,7 @@ func TestEventWindow(t *testing.T) {
 				"source_demoErr_0_records_in_total":  int64(6),
 				"source_demoErr_0_records_out_total": int64(6),
 
-				"op_4_window_0_exceptions_total":  int64(1),
+				"op_4_window_0_exceptions_total":  int64(0),
 				"op_4_window_0_records_in_total":  int64(3),
 				"op_4_window_0_records_out_total": int64(5),
 			},
@@ -1319,7 +1319,7 @@ func TestWindowError(t *testing.T) {
 				}},
 			},
 			M: map[string]interface{}{
-				"sink_memory_0_0_exceptions_total":  int64(1),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(1),
 				"sink_memory_0_0_records_out_total": int64(1),
 
@@ -1344,7 +1344,7 @@ func TestWindowError(t *testing.T) {
 				}}, {},
 			},
 			M: map[string]interface{}{
-				"sink_memory_0_0_exceptions_total":  int64(1),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(4),
 				"sink_memory_0_0_records_out_total": int64(4),
 
@@ -1383,12 +1383,12 @@ func TestWindowError(t *testing.T) {
 				}},
 			},
 			M: map[string]interface{}{
-				"op_5_project_0_exceptions_total":   int64(3),
+				"op_5_project_0_exceptions_total":   int64(0),
 				"op_5_project_0_process_latency_us": int64(0),
 				"op_5_project_0_records_in_total":   int64(5),
 				"op_5_project_0_records_out_total":  int64(5),
 
-				"sink_memory_0_0_exceptions_total":  int64(3),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(8),
 				"sink_memory_0_0_records_out_total": int64(8),
 
@@ -1427,12 +1427,12 @@ func TestWindowError(t *testing.T) {
 				}, {}},
 			},
 			M: map[string]interface{}{
-				"op_6_project_0_exceptions_total":   int64(3),
+				"op_6_project_0_exceptions_total":   int64(0),
 				"op_6_project_0_process_latency_us": int64(0),
 				"op_6_project_0_records_in_total":   int64(2),
 				"op_6_project_0_records_out_total":  int64(2),
 
-				"sink_memory_0_0_exceptions_total":  int64(3),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(5),
 				"sink_memory_0_0_records_out_total": int64(5),
 
@@ -1469,7 +1469,7 @@ func TestWindowError(t *testing.T) {
 				}},
 			},
 			M: map[string]interface{}{
-				"sink_memory_0_0_exceptions_total":  int64(1),
+				"sink_memory_0_0_exceptions_total":  int64(0),
 				"sink_memory_0_0_records_in_total":  int64(3),
 				"sink_memory_0_0_records_out_total": int64(3),
 
diff --git a/test/run_jmeter.sh b/test/run_jmeter.sh
index 3f69c41b..b6c75feb 100755
--- a/test/run_jmeter.sh
+++ b/test/run_jmeter.sh
@@ -150,8 +150,8 @@ echo -e "---------------------------------------------\n"
 #/opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/graph_window_rule.jmx -Dfvt="$fvt_dir" -l jmeter_logs/graph_window_rule.jtl -j jmeter_logs/graph_window_rule.log
 #echo -e "---------------------------------------------\n"
 
-#/opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/lookup_table_memory.jmx -Dfvt="$fvt_dir" -l jmeter_logs/lookup_table_memory.jtl -j jmeter_logs/lookup_table_memory.log
-#echo -e "---------------------------------------------\n"
+/opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/lookup_table_memory.jmx -Dfvt="$fvt_dir" -l jmeter_logs/lookup_table_memory.jtl -j jmeter_logs/lookup_table_memory.log
+echo -e "---------------------------------------------\n"
 
 #/opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/lookup_table_redis.jmx -Dfvt="$fvt_dir" -l jmeter_logs/lookup_table_redis.jtl -j jmeter_logs/lookup_table_redis.log
 #echo -e "---------------------------------------------\n"