Repository: lf-edge/ekuiper
Pull request #2905 (2024-06-05 04:20:04)

fix(topo): fix error handling for nodes

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

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/topo/node/batch\_op.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Fbatch_op.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL2JhdGNoX29wLmdv) | 52.08% | [20 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/file/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&filepath=internal%2Fio%2Ffile%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZmlsZS9zb3VyY2UuZ28=) | 72.97% | [6 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/node/cache\_op.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Fcache_op.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL2NhY2hlX29wLmdv) | 79.07% | [6 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/node/sink\_node.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Fsink_node.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL3Npbmtfbm9kZS5nbw==) | 80.85% | [6 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/node/source\_node.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Fsource_node.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL3NvdXJjZV9ub2RlLmdv) | 57.14% | [4 Missing and 2 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/file/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&filepath=internal%2Fio%2Ffile%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZmlsZS9zaW5rLmdv) | 66.67% | [2 Missing and 2 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/memory/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?src=pr&el=tree&filepath=internal%2Fio%2Fmemory%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vbWVtb3J5L3NvdXJjZS5nbw==) | 71.43% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2905?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    #2905   +/-   ##
============================================
  Coverage             ?   67.81%           
============================================
  Files                ?      294           
  Lines                ?    28219           
  Branches             ?        0           
============================================
  Hits                 ?    19136           
  Misses               ?     7371           
  Partials             ?     1712           
```



</details>

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


Patch:
diff --git a/internal/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/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/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/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/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/rule_test.go b/internal/topo/topotest/rule_test.go
index 03bda7a5..d4738b6d 100644
--- a/internal/topo/topotest/rule_test.go
+++ b/internal/topo/topotest/rule_test.go
@@ -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..0063f422 100644
--- a/internal/topo/topotest/window_rule_test.go
+++ b/internal/topo/topotest/window_rule_test.go
@@ -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),