Repository: lf-edge/ekuiper
Pull request #2852 (2024-05-16 08:11:58)

refactor(sink): split cache op

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2852?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.73684%` with `69 lines` in your changes are missing coverage. Please review.
> :exclamation: No coverage uploaded for pull request base (`master-v2@5039883`). [Click here to learn what that means](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).

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2852?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/cache/sync\_cache.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2852?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Fcache%2Fsync_cache.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL2NhY2hlL3N5bmNfY2FjaGUuZ28=) | 74.07% | [19 Missing and 16 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2852?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/2852?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) | 62.79% | [24 Missing and 8 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2852?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/xsql/row.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2852?src=pr&el=tree&filepath=internal%2Fxsql%2Frow.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwveHNxbC9yb3cuZ28=) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2852?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    #2852   +/-   ##
============================================
  Coverage             ?   68.59%           
============================================
  Files                ?      272           
  Lines                ?    25716           
  Branches             ?        0           
============================================
  Hits                 ?    17638           
  Misses               ?     6491           
  Partials             ?     1587           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2852?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/topo/node/batch_op.go b/internal/topo/node/batch_op.go
index 051921c2..25d0d46c 100644
--- a/internal/topo/node/batch_op.go
+++ b/internal/topo/node/batch_op.go
@@ -86,19 +86,10 @@ func (b *BatchOp) ingest(ctx api.StreamContext, item any, checkSize bool) {
 	if processed {
 		return
 	}
-	switch d := data.(type) {
-	case error:
-		b.Broadcast(d)
-		b.statManager.IncTotalExceptions(d.Error())
-		return
-	case *xsql.WatermarkTuple:
-		b.Broadcast(d)
-		return
-	}
 
 	b.statManager.IncTotalRecordsIn()
 	b.statManager.ProcessTimeStart()
-	switch input := item.(type) {
+	switch input := data.(type) {
 	case xsql.Row:
 		b.buffer.AddTuple(input)
 	case xsql.Collection:
diff --git a/internal/topo/node/cache/sync_cache.go b/internal/topo/node/cache/sync_cache.go
index 6e7d7c02..ee9afa01 100644
--- a/internal/topo/node/cache/sync_cache.go
+++ b/internal/topo/node/cache/sync_cache.go
@@ -15,6 +15,7 @@
 package cache
 
 import (
+	"fmt"
 	"path"
 	"strconv"
 	"time"
@@ -22,14 +23,16 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
-	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/kv"
 )
 
+// SyncCache is the struct to handle cache saving and read
+// The data are sink tuples: MessageTuple, MessageTupleList or RawTuple
+
 // page Rotates storage for in memory cache
 // Not thread safe!
 type page struct {
-	Data [][]map[string]interface{}
+	Data []any
 	H    int
 	T    int
 	L    int
@@ -37,10 +40,9 @@ type page struct {
 }
 
 // newPage create a new cache page
-// TODO the page is created even not used, need dynamic?
 func newPage(size int) *page {
 	return &page{
-		Data: make([][]map[string]interface{}, size),
+		Data: make([]any, size),
 		H:    0, // When deleting, head++, if tail == head, it is empty
 		T:    0, // When append, tail++, if tail== head, it is full
 		Size: size,
@@ -48,7 +50,7 @@ func newPage(size int) *page {
 }
 
 // append item if list is not full and return true; otherwise return false
-func (p *page) append(item []map[string]interface{}) bool {
+func (p *page) append(item any) bool {
 	if p.L == p.Size { // full
 		return false
 	}
@@ -62,7 +64,7 @@ func (p *page) append(item []map[string]interface{}) bool {
 }
 
 // peak get the first item in the cache
-func (p *page) peak() ([]map[string]interface{}, bool) {
+func (p *page) peak() (any, bool) {
 	if p.L == 0 {
 		return nil, false
 	}
@@ -92,255 +94,189 @@ func (p *page) reset() {
 }
 
 type SyncCache struct {
-	// The input data to the cache
-	in      <-chan []map[string]interface{}
-	Out     chan []map[string]interface{}
-	Ack     chan bool
-	errorCh chan<- error
 	// cache config
 	cacheConf   *conf.SinkConf
 	maxDiskPage int
-	maxMemPage  int
 	// cache storage
-	memCache       []*page
-	diskBufferPage *page
+	writeBufferPage *page
+	readBufferPage  *page
 	// status
 	diskSize     int // the count of pages has been saved
 	CacheLength  int // readonly, for metrics only to save calculation
 	diskPageTail int // init from the database
 	diskPageHead int
-	sendStatus   int // 0: idle, 1: sending and waiting for ack, 2: stopped for error
 	// serialize
 	store kv.KeyValue
-
-	exitCh chan<- struct{}
-}
-
-func NewSyncCacheWithExitChanel(ctx api.StreamContext, in <-chan []map[string]interface{}, errCh chan<- error, cacheConf *conf.SinkConf, bufferLength int, exitCh chan<- struct{}) *SyncCache {
-	c := NewSyncCache(ctx, in, errCh, cacheConf, bufferLength)
-	c.exitCh = exitCh
-	return c
 }
 
-func NewSyncCache(ctx api.StreamContext, in <-chan []map[string]interface{}, errCh chan<- error, cacheConf *conf.SinkConf, bufferLength int) *SyncCache {
+func NewSyncCache(ctx api.StreamContext, cacheConf *conf.SinkConf) (*SyncCache, error) {
+	ctx.GetLogger().Infof("create sync cache with conf %+v", cacheConf)
+	// The maximum pages in disk. This includes readBuffer, all disk page and write buffer. When flush, all save into disk
+	diskPage := cacheConf.MaxDiskCache / cacheConf.BufferPageSize
+	if diskPage < 2 {
+		diskPage = 2
+		ctx.GetLogger().Warnf("disk page is less than 2, so set it to 2")
+	}
 	c := &SyncCache{
-		cacheConf:  cacheConf,
-		in:         in,
-		Out:        make(chan []map[string]interface{}, bufferLength),
-		Ack:        make(chan bool, 10),
-		errorCh:    errCh,
-		maxMemPage: cacheConf.MemoryCacheThreshold / cacheConf.BufferPageSize,
-		memCache:   make([]*page, 0),
+		cacheConf: cacheConf,
 		// add one more slot so that there will be at least one slot between head and tail to find out the head/tail id
-		maxDiskPage: (cacheConf.MaxDiskCache / cacheConf.BufferPageSize) + 1,
+		maxDiskPage:     diskPage,
+		writeBufferPage: newPage(cacheConf.BufferPageSize),
+		readBufferPage:  newPage(cacheConf.BufferPageSize),
 	}
-	go func() {
-		err := infra.SafeRun(func() error {
-			c.run(ctx)
-			return nil
-		})
-		if err != nil {
-			infra.DrainError(ctx, err, errCh)
-		}
-	}()
-	return c
+	err := c.initStore(ctx)
+	return c, err
 }
 
-func (c *SyncCache) run(ctx api.StreamContext) {
-	c.initStore(ctx)
-	defer c.onClose(ctx)
-	if c.CacheLength > 0 { // start to send the cache
-		c.send(ctx)
-	}
-	for {
-		select {
-		case item := <-c.in:
-			ctx.GetLogger().Debugf("adding cache %v", item)
-			// hack here: nil is a signal to continue sending, so not adding nil to cache
-			if item != nil {
-				c.addCache(ctx, item)
-			} else {
-				ctx.GetLogger().Debug("nil cache, continue sending")
-			}
-			if c.sendStatus == 2 {
-				c.sendStatus = 0
-				ctx.GetLogger().Debug("send status to 0 after adding cache in error state")
-			}
-			ctx.GetLogger().Debugf("cache status %d", c.sendStatus)
-			if c.sendStatus == 0 {
-				c.send(ctx)
-			}
-		case isSuccess := <-c.Ack:
-			// only send the next sink after receiving an ack
-			ctx.GetLogger().Debugf("cache ack")
-			if isSuccess {
-				ctx.GetLogger().Debugf("deleting cache")
-				c.deleteCache(ctx)
-				c.sendStatus = 0
-				ctx.GetLogger().Debug("send status to 0 after true ack")
-			} else {
-				c.sendStatus = 2
-				ctx.GetLogger().Debug("send status to 2 after false ack")
-			}
-			ctx.GetLogger().Debugf("cache status %d", c.sendStatus)
-			if c.sendStatus == 0 {
-				c.send(ctx)
-			}
-		case <-ctx.Done():
-			ctx.GetLogger().Infof("sink node %s instance cache %d done", ctx.GetOpId(), ctx.GetInstanceId())
-			return
+// AddCache not thread safe!
+func (c *SyncCache) AddCache(ctx api.StreamContext, item any) error {
+	isBufferNotFull := c.writeBufferPage.append(item)
+	if !isBufferNotFull { // cool page full, save to disk
+		err := c.appendWriteCache(ctx)
+		if err != nil {
+			return err
 		}
+		c.writeBufferPage.reset()
+		c.writeBufferPage.append(item)
+	} else {
+		ctx.GetLogger().Debugf("added cache to disk buffer page %v", c.writeBufferPage)
 	}
+	c.CacheLength++
+	ctx.GetLogger().Debugf("added cache %d", c.CacheLength)
+	return nil
 }
 
-func (c *SyncCache) send(ctx api.StreamContext) {
-	if c.CacheLength > 1 && c.cacheConf.ResendInterval > 0 {
-		time.Sleep(time.Duration(c.cacheConf.ResendInterval) * time.Millisecond)
-	}
-	d, ok := c.peakMemCache(ctx)
-	if ok {
-		ctx.GetLogger().Debugf("sending cache item %v", d)
-		c.sendStatus = 1
-		ctx.GetLogger().Debug("send status to 0 after sending tuple")
-		select {
-		case c.Out <- d:
-			ctx.GetLogger().Debugf("sink cache send out %v", d)
-		case <-ctx.Done():
-			ctx.GetLogger().Debugf("stop sink cache send")
+func (c *SyncCache) appendWriteCache(ctx api.StreamContext) error {
+	if c.diskSize == c.maxDiskPage {
+		// disk full, replace read buffer page
+		err := c.deleteDiskPage(ctx, false)
+		if err != nil {
+			return err
 		}
+		// also delete read buffer which is even older
+		c.CacheLength -= c.readBufferPage.L
+		ctx.GetLogger().Debug("disk full, remove the last page %v", c.readBufferPage)
+		c.readBufferPage.reset()
+	}
+	err := c.store.Set(strconv.Itoa(c.diskPageTail), c.writeBufferPage)
+	if err != nil {
+		return fmt.Errorf("fail to store disk cache %v", err)
 	} else {
-		ctx.GetLogger().Debug("no cache to send")
+		ctx.GetLogger().Debug("add cache to disk. the new disk buffer page is %v", c.writeBufferPage)
+		c.diskPageTail++
+		c.diskSize++
+		err := c.store.Set("size", c.diskSize)
+		if err != nil {
+			ctx.GetLogger().Warnf("fail to store disk cache size %v", err)
+		}
+		// rotate
+		if c.diskPageTail == c.maxDiskPage {
+			c.diskPageTail = 0
+		}
 	}
+	return nil
 }
 
-// addCache not thread safe!
-func (c *SyncCache) addCache(ctx api.StreamContext, item []map[string]interface{}) {
-	isNotFull := c.appendMemCache(item)
-	if !isNotFull {
-		if c.diskBufferPage == nil {
-			c.diskBufferPage = newPage(c.cacheConf.BufferPageSize)
+func (c *SyncCache) insertReadCache(ctx api.StreamContext) error {
+	// insert before current head
+	head := c.diskPageHead - 1
+	if head < 0 {
+		head = c.maxDiskPage - 1
+	}
+	err := c.store.Set(strconv.Itoa(head), c.readBufferPage)
+	if err != nil {
+		return fmt.Errorf("fail to insert read cache to disk %v", err)
+	} else {
+		c.diskPageHead = head
+		err = c.store.Set("head", c.diskPageHead)
+		if err != nil {
+			ctx.GetLogger().Warnf("fail to store disk cache size %v", err)
+			return err
 		}
-		isBufferNotFull := c.diskBufferPage.append(item)
-		if !isBufferNotFull { // cool page full, save to disk
-			if c.diskSize == c.maxDiskPage {
-				// disk full, read the oldest page to the hot page
-				c.loadFromDisk(ctx)
-				ctx.GetLogger().Debug("disk full, remove the last page")
-			}
-			err := c.store.Set(strconv.Itoa(c.diskPageTail), c.diskBufferPage)
+		// If not full
+		if c.diskSize < c.maxDiskPage {
+			c.diskSize++
+			err = c.store.Set("size", c.diskSize)
 			if err != nil {
-				ctx.GetLogger().Errorf("fail to store disk cache %v", err)
-				return
-			} else {
-				ctx.GetLogger().Debug("add cache to disk. the new disk buffer page is %v", c.diskBufferPage)
-				c.diskPageTail++
-				c.diskSize++
-				err := c.store.Set("size", c.diskSize)
-				if err != nil {
-					ctx.GetLogger().Warnf("fail to store disk cache size %v", err)
-				}
-				// rotate
-				if c.diskPageTail == c.maxDiskPage {
-					c.diskPageTail = 0
-				}
+				ctx.GetLogger().Warnf("fail to store disk cache size %v", err)
 			}
-			c.diskBufferPage.reset()
-			c.diskBufferPage.append(item)
-		} else {
-			ctx.GetLogger().Debugf("added cache to disk buffer page %v", c.diskBufferPage)
 		}
-	} else {
-		ctx.GetLogger().Debugf("added cache to mem cache %v", item)
 	}
-	c.CacheLength++
-	ctx.GetLogger().Debugf("added cache %d", c.CacheLength)
+	return nil
 }
 
-// deleteCache not thread safe!
-func (c *SyncCache) deleteCache(ctx api.StreamContext) {
-	ctx.GetLogger().Debugf("deleting cache. CacheLength: %d, diskSize: %d", c.CacheLength, c.diskSize)
-	if len(c.memCache) == 0 {
-		ctx.GetLogger().Debug("mem cache is empty")
-		return
+// PopCache not thread safe!
+func (c *SyncCache) PopCache(ctx api.StreamContext) (any, bool) {
+	ctx.GetLogger().Debugf("poping cache. CacheLength: %d, diskSize: %d", c.CacheLength, c.diskSize)
+	if c.readBufferPage.isEmpty() {
+		// read from disk or cool list
+		if c.diskSize > 0 {
+			err := c.loadFromDisk(ctx)
+			if err != nil {
+				ctx.GetLogger().Error(err)
+			}
+		} else if !c.writeBufferPage.isEmpty() { // use cool page as the new page
+			ctx.GetLogger().Debugf("reading from writeBufferPage: %d", c.CacheLength)
+			c.readBufferPage = c.writeBufferPage
+			c.writeBufferPage = newPage(c.cacheConf.BufferPageSize)
+		}
 	}
-	isNotEmpty := c.memCache[0].delete()
+	result, _ := c.readBufferPage.peak()
+	isNotEmpty := c.readBufferPage.delete()
 	if isNotEmpty {
 		c.CacheLength--
 		ctx.GetLogger().Debugf("deleted cache: %d", c.CacheLength)
 	}
-	if c.memCache[0].isEmpty() { // read from disk or cool list
-		c.memCache = c.memCache[1:]
-		if c.diskSize > 0 {
-			c.loadFromDisk(ctx)
-		} else if c.diskBufferPage != nil { // use cool page as the new page
-			ctx.GetLogger().Debugf("reading from diskBufferPage: %d", c.CacheLength)
-			c.memCache = append(c.memCache, c.diskBufferPage)
-			c.diskBufferPage = nil
-		}
-	}
-	ctx.GetLogger().Debugf("deleted cache. CacheLength: %d, diskSize: %d, memCache: %v", c.CacheLength, c.diskSize, c.memCache)
+	ctx.GetLogger().Debugf("deleted cache. CacheLength: %d, diskSize: %d, readPage: %v", c.CacheLength, c.diskSize, c.readBufferPage)
+	return result, true
 }
 
-func (c *SyncCache) loadFromDisk(ctx api.StreamContext) {
-	// load page from the disk
-	ctx.GetLogger().Debugf("loading from disk %d. CacheLength: %d, diskSize: %d", c.diskPageTail, c.CacheLength, c.diskSize)
-	hotPage := newPage(c.cacheConf.BufferPageSize)
-	ok, err := c.store.Get(strconv.Itoa(c.diskPageHead), hotPage)
-	if err != nil {
-		ctx.GetLogger().Errorf("fail to load disk cache %v", err)
-	} else if !ok {
-		ctx.GetLogger().Errorf("nothing in the disk, should not happen")
-	} else {
-		_ = c.store.Delete(strconv.Itoa(c.diskPageHead))
-		if len(c.memCache) >= c.maxMemPage {
-			ctx.GetLogger().Warnf("drop a page of %d items in memory", c.memCache[0].L)
-			c.CacheLength -= c.memCache[0].L
-			c.memCache = c.memCache[1:]
-		}
-		c.memCache = append(c.memCache, hotPage)
-		c.diskPageHead++
-		c.diskSize--
-		err := c.store.Set("size", c.diskSize)
-		if err != nil {
-			ctx.GetLogger().Warnf("fail to store disk cache size %v", err)
-		}
-		if c.diskPageHead == c.maxDiskPage {
-			c.diskPageHead = 0
-		}
-		err = c.store.Set("head", c.diskPageHead)
-		if err != nil {
-			ctx.GetLogger().Warnf("fail to store disk cache size %v", err)
-		}
+// loaded means whether load the page to memory or just drop
+func (c *SyncCache) deleteDiskPage(ctx api.StreamContext, loaded bool) error {
+	_ = c.store.Delete(strconv.Itoa(c.diskPageHead))
+	ctx.GetLogger().Warnf("drop a read page of %d items in memory", c.readBufferPage.L)
+	c.diskPageHead++
+	c.diskSize--
+	if !loaded {
+		c.CacheLength -= c.cacheConf.BufferPageSize
 	}
-	ctx.GetLogger().Debugf("loaded from disk %d. CacheLength: %d, diskSize: %d", c.diskPageTail, c.CacheLength, c.diskSize)
-}
-
-func (c *SyncCache) appendMemCache(item []map[string]interface{}) bool {
-	if len(c.memCache) > c.maxMemPage {
-		return false
+	err := c.store.Set("size", c.diskSize)
+	if err != nil {
+		ctx.GetLogger().Warnf("fail to store disk cache size %v", err)
+		return err
 	}
-	if len(c.memCache) == 0 {
-		c.memCache = append(c.memCache, newPage(c.cacheConf.BufferPageSize))
+	if c.diskPageHead == c.maxDiskPage {
+		c.diskPageHead = 0
 	}
-	isNotFull := c.memCache[len(c.memCache)-1].append(item)
-	if !isNotFull {
-		if len(c.memCache) == c.maxMemPage {
-			return false
-		}
-		c.memCache = append(c.memCache, newPage(c.cacheConf.BufferPageSize))
-		return c.memCache[len(c.memCache)-1].append(item)
+	err = c.store.Set("head", c.diskPageHead)
+	if err != nil {
+		ctx.GetLogger().Warnf("fail to store disk cache size %v", err)
+		return err
 	}
-	return true
+	return nil
 }
 
-func (c *SyncCache) peakMemCache(_ api.StreamContext) ([]map[string]interface{}, bool) {
-	if len(c.memCache) == 0 {
-		return nil, false
+func (c *SyncCache) loadFromDisk(ctx api.StreamContext) error {
+	// load page from the disk
+	ctx.GetLogger().Debugf("loading from disk %d. CacheLength: %d, diskSize: %d", c.diskPageTail, c.CacheLength, c.diskSize)
+	// caution, must create a new page instance
+	p := &page{}
+	ok, err := c.store.Get(strconv.Itoa(c.diskPageHead), p)
+	if err != nil {
+		return fmt.Errorf("fail to load disk cache %v", err)
+	} else if !ok {
+		return fmt.Errorf("nothing in the disk, should not happen")
 	}
-	return c.memCache[0].peak()
+	c.readBufferPage = p
+	err = c.deleteDiskPage(ctx, true)
+	if err != nil {
+		return err
+	}
+	ctx.GetLogger().Debugf("loaded from disk %d. CacheLength: %d, diskSize: %d", c.diskPageTail, c.CacheLength, c.diskSize)
+	return nil
 }
 
-func (c *SyncCache) initStore(ctx api.StreamContext) {
+func (c *SyncCache) initStore(ctx api.StreamContext) error {
 	kvTable := path.Join("sink", ctx.GetRuleId()+ctx.GetOpId()+strconv.Itoa(ctx.GetInstanceId()))
 	if c.cacheConf.CleanCacheAtStop {
 		ctx.GetLogger().Infof("creating cache store %s", kvTable)
@@ -349,7 +285,7 @@ func (c *SyncCache) initStore(ctx api.StreamContext) {
 	var err error
 	c.store, err = store.GetCacheKV(kvTable)
 	if err != nil {
-		infra.DrainError(ctx, err, c.errorCh)
+		return err
 	}
 	// restore the sink cache from disk
 	if !c.cacheConf.CleanCacheAtStop {
@@ -372,24 +308,11 @@ func (c *SyncCache) initStore(ctx api.StreamContext) {
 		}
 		_ = c.store.Set("storeSig", 0)
 		ctx.GetLogger().Infof("start to restore cache from disk")
-		// restore the memCache
-		_, err = c.store.Get("memcache", &c.memCache)
-		if err != nil {
-			ctx.GetLogger().Errorf("fail to restore mem cache %v", err)
-		}
-		for _, p := range c.memCache {
-			c.CacheLength += p.L
-		}
-		err = c.store.Delete("memcache")
-		if err != nil {
-			ctx.GetLogger().Errorf("fail to delete mem cache %v", err)
-		}
-		ctx.GetLogger().Infof("restored mem cache %d", c.CacheLength)
 		// restore the disk cache
 		var size int
 		ok, _ = c.store.Get("size", &size)
 		if !ok || size == 0 { // no disk cache
-			return
+			return nil
 		}
 		c.diskSize = size
 		var head int
@@ -397,55 +320,40 @@ func (c *SyncCache) initStore(ctx api.StreamContext) {
 		if ok {
 			c.diskPageHead = head
 		}
-		c.CacheLength += (c.diskSize - 1) * c.cacheConf.BufferPageSize
-		c.diskPageTail = (c.diskPageHead + c.diskSize - 1) % c.maxDiskPage
-		// load buffer page
-		hotPage := newPage(c.cacheConf.BufferPageSize)
-		ok, err = c.store.Get(strconv.Itoa(c.diskPageTail), hotPage)
-		if err != nil {
-			ctx.GetLogger().Errorf("fail to load disk cache to buffer %v", err)
-		} else if !ok {
-			ctx.GetLogger().Errorf("nothing in the disk, should not happen")
-		} else {
-			c.diskBufferPage = hotPage
-			c.CacheLength += c.diskBufferPage.L
-			c.diskSize--
+		var cacheLength int
+		ok, _ = c.store.Get("cacheLength", &cacheLength)
+		if ok {
+			c.CacheLength = cacheLength
 		}
+		c.diskPageTail = (c.diskPageHead + c.diskSize) % c.maxDiskPage
 		ctx.GetLogger().Infof("restored all cache %d. diskSize %d", c.CacheLength, c.diskSize)
 	}
+	return nil
 }
 
-// save memory states to disk
-func (c *SyncCache) onClose(ctx api.StreamContext) {
-	defer func() {
-		if c.exitCh != nil {
-			c.exitCh <- struct{}{}
-		}
-	}()
+// Flush save memory states to disk.
+func (c *SyncCache) Flush(ctx api.StreamContext) {
 	ctx.GetLogger().Infof("sink node %s instance cache %d closing", ctx.GetOpId(), ctx.GetInstanceId())
 	if c.cacheConf.CleanCacheAtStop {
 		kvTable := path.Join("sink", ctx.GetRuleId()+ctx.GetOpId()+strconv.Itoa(ctx.GetInstanceId()))
 		ctx.GetLogger().Infof("cleaning cache store %s", kvTable)
 		_ = store.DropCacheKV(kvTable)
 	} else {
-		if c.diskBufferPage != nil {
-			err := c.store.Set(strconv.Itoa(c.diskPageTail), c.diskBufferPage)
-			if err != nil {
-				ctx.GetLogger().Errorf("fail to store disk cache %v", err)
-			}
-			err = c.store.Set("size", c.diskSize+1)
-			if err != nil {
-				ctx.GetLogger().Errorf("fail to store disk size %v", err)
-			}
-			ctx.GetLogger().Debug("store disk cache")
+		var err error
+		if !c.readBufferPage.isEmpty() {
+			err = c.insertReadCache(ctx)
 		}
-		// store the memory states
-		if len(c.memCache) > 0 {
-			err := c.store.Set("memcache", c.memCache)
-			if err != nil {
-				ctx.GetLogger().Errorf("fail to store memory cache to disk %v", err)
-			}
-			ctx.GetLogger().Debugf("store memory cache %d", len(c.memCache))
+		if !c.writeBufferPage.isEmpty() {
+			err = c.appendWriteCache(ctx)
+		}
+		if err != nil {
+			ctx.GetLogger().Error(err)
+		} else {
+			ctx.GetLogger().Infof("append write cache to disk")
+		}
+		err = c.store.Set("cacheLength", c.CacheLength)
+		if err != nil {
+			ctx.GetLogger().Warnf("fail to store disk cache size %v", err)
 		}
 		_ = c.store.Set("storeSig", 1)
 	}
diff --git a/internal/topo/node/cache_op.go b/internal/topo/node/cache_op.go
new file mode 100644
index 00000000..faa1c1d0
--- /dev/null
+++ b/internal/topo/node/cache_op.go
@@ -0,0 +1,174 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package node
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/benbjohnson/clock"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/checkpoint"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/node/cache"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+// CacheOp receives tuples and decide to send through or save to disk. Run right before sink
+// Immutable: true
+// Input: any (mostly MessageTuple/MessageTupleList, may receive RawTuple after transformOp)
+// Special validation: one output only
+type CacheOp struct {
+	*defaultSinkNode
+	// configs
+	cacheConf *conf.SinkConf
+	// state
+	cache    *cache.SyncCache
+	currItem any
+	hasCache bool
+	// send timer, only enabled when there is cache. disable when all cache are sent
+	resendTicker  *clock.Ticker
+	resendTimerCh <-chan time.Time
+}
+
+func NewCacheOp(ctx api.StreamContext, name string, rOpt *def.RuleOption, sc *conf.SinkConf) (*CacheOp, error) {
+	// use channel buffer as memory cache
+	c, err := cache.NewSyncCache(ctx, sc)
+	if err != nil {
+		return nil, err
+	}
+	return &CacheOp{
+		defaultSinkNode: newDefaultSinkNode(name, rOpt),
+		cache:           c,
+		cacheConf:       sc,
+	}, nil
+}
+
+// Exec ingest data and send through.
+// If channel full, save data to disk cache and start send timer
+// Once all cache sent, stop send timer
+func (s *CacheOp) Exec(ctx api.StreamContext, errCh chan<- error) {
+	if len(s.outputs) > 1 {
+		infra.DrainError(ctx, fmt.Errorf("cache op should have only 1 output but got %+v", s.outputs), errCh)
+	}
+	s.prepareExec(ctx, errCh, "op")
+	go func() {
+		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)
+						break
+					}
+				} else {
+					s.currItem = data
+				}
+				s.send()
+
+				s.statManager.ProcessTimeEnd()
+				s.statManager.IncTotalMessagesProcessed(1)
+			case <-s.resendTimerCh:
+				ctx.GetLogger().Debugf("ticker is triggered")
+				s.statManager.ProcessTimeStart()
+				s.send()
+				s.statManager.ProcessTimeEnd()
+			}
+		}
+	}()
+}
+
+func (s *CacheOp) send() {
+	if s.currItem == nil { // current item sent out finally
+		if s.cache.CacheLength > 0 {
+			// read
+			var readOk bool
+			s.currItem, readOk = s.cache.PopCache(s.ctx)
+			if !readOk { // should never happen
+				s.ctx.GetLogger().Errorf("fail to read from cache")
+			} else {
+				s.ctx.GetLogger().Debugf("read from cache %v", s.currItem)
+			}
+		} else {
+			// cancel the timer since all cache are sent
+			s.resendTicker.Stop()
+			s.hasCache = false
+			s.ctx.GetLogger().Debugf("cache all sent, stop ticker")
+			return
+		}
+	}
+	// Send by custom broadcast, if successful, reset currItem to nil
+	s.Broadcast(s.currItem)
+}
+
+func (s *CacheOp) Broadcast(val interface{}) {
+	if _, ok := val.(error); ok && !s.sendError {
+		return
+	}
+	if s.qos >= def.AtLeastOnce {
+		boe := &checkpoint.BufferOrEvent{
+			Data:    val,
+			Channel: s.name,
+		}
+		s.doBroadcast(boe)
+		return
+	}
+	s.doBroadcast(val)
+	return
+}
+
+func (s *CacheOp) doBroadcast(val interface{}) {
+	var out chan<- any
+	for _, output := range s.outputs {
+		out = output
+	}
+	select {
+	case out <- val:
+		s.ctx.GetLogger().Debugf("send out data %v", val)
+		// send through. The sink must retry until successful
+		s.currItem = nil
+		s.statManager.IncTotalRecordsOut()
+	case <-s.ctx.Done():
+		// rule stop so stop waiting
+	default:
+		if !s.hasCache {
+			s.ctx.GetLogger().Debugf("memory buffer full, start to save cache")
+			// Start the send interval
+			s.resendTicker = timex.GetTicker(int64(s.cacheConf.ResendInterval))
+			s.resendTimerCh = s.resendTicker.C
+			s.hasCache = true
+		}
+	}
+}
diff --git a/internal/topo/planner/planner_sink.go b/internal/topo/planner/planner_sink.go
index 71165de1..ec7c277e 100644
--- a/internal/topo/planner/planner_sink.go
+++ b/internal/topo/planner/planner_sink.go
@@ -143,5 +143,15 @@ func splitSink(tp *topo.Topo, inputs []node.Emitter, s api.Sink, sinkName string
 			newInputs = []node.Emitter{encryptOp}
 		}
 	}
+	// Caching
+	if sc.EnableCache {
+		cacheOp, err := node.NewCacheOp(tp.GetContext(), fmt.Sprintf("%s_%d_cache", sinkName, index), options, &sc.SinkConf)
+		if err != nil {
+			return nil, err
+		}
+		index++
+		tp.AddOperator(newInputs, cacheOp)
+		newInputs = []node.Emitter{cacheOp}
+	}
 	return newInputs, nil
 }
diff --git a/internal/xsql/gob_register.go b/internal/xsql/gob_register.go
index b79408cb..47fef246 100644
--- a/internal/xsql/gob_register.go
+++ b/internal/xsql/gob_register.go
@@ -21,8 +21,9 @@ import (
 
 func init() {
 	gob.Register(time.Time{})
-	gob.Register(make(map[string]interface{}))
+	gob.Register(make(map[string]any))
 	gob.Register(make(map[string][]*Tuple))
-	gob.Register(Tuple{})
-	gob.Register([]interface{}{})
+	gob.Register(&Tuple{})
+	gob.Register([]any{})
+	gob.Register(&RawTuple{})
 }
diff --git a/internal/xsql/row.go b/internal/xsql/row.go
index 887570f5..5f916f85 100644
--- a/internal/xsql/row.go
+++ b/internal/xsql/row.go
@@ -15,6 +15,7 @@
 package xsql
 
 import (
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -315,6 +316,10 @@ func (t *Tuple) AllMeta() map[string]any {
 	return t.Metadata
 }
 
+func (t *Tuple) ToString() string {
+	return strconv.FormatInt(t.Timestamp, 10)
+}
+
 var (
 	_ Row          = &Tuple{}
 	_ MetaData     = &Tuple{}


Test patch:
diff --git a/internal/topo/node/cache/sync_cache_test.go b/internal/topo/node/cache/sync_cache_test.go
index 50729ca9..1c8f10b0 100644
--- a/internal/topo/node/cache/sync_cache_test.go
+++ b/internal/topo/node/cache/sync_cache_test.go
@@ -20,13 +20,15 @@ import (
 	"path/filepath"
 	"reflect"
 	"testing"
-	"time"
+
+	"github.com/stretchr/testify/assert"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/state"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 )
 
 func TestPage(t *testing.T) {
@@ -111,140 +113,292 @@ func TestPage(t *testing.T) {
 	}
 }
 
-// TestRun test for
-// 1. cache in memory only
-// 2. cache in memory and disk buffer only
-// 3. cache in memory and disk
-// 4. cache in memory and disk buffer and overflow
-// Each flow test rule restart
-// Each flow use slightly different config like bufferPageSize
-func TestRun(t *testing.T) {
+func TestCache(t *testing.T) {
+	testx.InitEnv("cache")
+	tempStore, err := state.CreateStore("mock", def.AtMostOnce)
+	assert.NoError(t, err)
+	deleteCachedb()
+	contextLogger := conf.Log.WithField("rule", "TestCache")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger).WithMeta("TestCache", "op1", tempStore)
+	s, err := NewSyncCache(ctx, &conf.SinkConf{
+		MaxDiskCache:     6,
+		BufferPageSize:   2,
+		EnableCache:      true,
+		ResendInterval:   0,
+		CleanCacheAtStop: false,
+	})
+	assert.NoError(t, err)
+	// prepare data
+	tuples := make([]any, 15)
+	for i := 0; i < 15; i++ {
+		tuples[i] = &xsql.RawTuple{
+			Emitter:   "test",
+			Timestamp: int64(i),
+			Rawdata:   []byte("hello"),
+			Metadata:  map[string]any{"topic": "demo"},
+		}
+	}
+
 	tests := []struct {
-		sconf   *conf.SinkConf
-		dataIn  [][]map[string]interface{}
-		dataOut [][]map[string]interface{}
-		stopPt  int // restart the rule in this point
+		name   string
+		inputs []any
+		output any
+		length int
 	}{
-		{ // 0
-			sconf: &conf.SinkConf{
-				MemoryCacheThreshold: 4,
-				MaxDiskCache:         12,
-				BufferPageSize:       2,
-				EnableCache:          true,
-				ResendInterval:       0,
-				CleanCacheAtStop:     false,
+		{
+			name:   "read empty",
+			length: 0,
+		},
+		{
+			name:   "read in write buffer",
+			inputs: tuples[:2],
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 0,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
 			},
-			dataIn: [][]map[string]interface{}{
-				{{"a": 1}}, {{"a": 2}}, {{"a": 3}}, {{"a": 4}}, {{"a": 5}},
+			length: 1,
+		},
+		{
+			name:   "read in read and write buffer",
+			inputs: tuples[2:4],
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 1,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
 			},
-			stopPt: 4,
+			length: 2,
 		},
-		{ // 1
-			sconf: &conf.SinkConf{
-				MemoryCacheThreshold: 4,
-				MaxDiskCache:         8,
-				BufferPageSize:       2,
-				EnableCache:          true,
-				ResendInterval:       0,
-				CleanCacheAtStop:     false,
+		{
+			name:   "read in disk",
+			inputs: tuples[4:7],
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 2,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
 			},
-			dataIn: [][]map[string]interface{}{
-				{{"a": 1}}, {{"a": 2}}, {{"a": 3}}, {{"a": 4}}, {{"a": 5}}, {{"a": 6}},
+			length: 4,
+		},
+		{
+			name:   "read in mem and disk overflow",
+			inputs: tuples[7:],
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 8,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
 			},
-			stopPt: 5,
+			length: 6,
 		},
-		{ // 2
-			sconf: &conf.SinkConf{
-				MemoryCacheThreshold: 1,
-				MaxDiskCache:         8,
-				BufferPageSize:       1,
-				EnableCache:          true,
-				ResendInterval:       0,
-				CleanCacheAtStop:     false,
+		{
+			name: "read in left read buffer",
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 9,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
 			},
-			dataIn: [][]map[string]interface{}{
-				{{"a": 1}}, {{"a": 2}}, {{"a": 3}}, {{"a": 4}}, {{"a": 5}}, {{"a": 6}},
+			length: 5,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			for _, tuple := range tt.inputs {
+				err = s.AddCache(ctx, tuple)
+				assert.NoError(t, err)
+			}
+			r, _ := s.PopCache(ctx)
+			assert.Equal(t, tt.output, r)
+			assert.Equal(t, tt.length, s.CacheLength, "cache length")
+		})
+	}
+}
+
+func TestCacheCase2(t *testing.T) {
+	testx.InitEnv("cache2")
+	tempStore, err := state.CreateStore("mock", def.AtMostOnce)
+	assert.NoError(t, err)
+	deleteCachedb()
+	contextLogger := conf.Log.WithField("rule", "TestCache")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger).WithMeta("TestCache", "op1", tempStore)
+	s, err := NewSyncCache(ctx, &conf.SinkConf{
+		MemoryCacheThreshold: 2,
+		MaxDiskCache:         4,
+		BufferPageSize:       2,
+		EnableCache:          true,
+		ResendInterval:       10,
+	})
+	assert.NoError(t, err)
+	// prepare data
+	tuples := make([]any, 15)
+	for i := 0; i < 15; i++ {
+		tuples[i] = &xsql.RawTuple{
+			Emitter:   "test",
+			Timestamp: int64(i),
+			Rawdata:   []byte("hello"),
+			Metadata:  map[string]any{"topic": "demo"},
+		}
+	}
+
+	tests := []struct {
+		name   string
+		inputs []any
+		output any
+		length int
+	}{
+		{
+			name:   "read in write buffer",
+			inputs: tuples[4:5],
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 4,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
 			},
-			stopPt: 4,
+			length: 0,
 		},
-		{ // 3
-			sconf: &conf.SinkConf{
-				MemoryCacheThreshold: 2,
-				MaxDiskCache:         4,
-				BufferPageSize:       2,
-				EnableCache:          true,
-				ResendInterval:       0,
-				CleanCacheAtStop:     false,
+		{
+			name:   "read in write buffer cont",
+			inputs: tuples[5:6],
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 5,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
 			},
-			dataIn: [][]map[string]interface{}{
-				{{"a": 1}}, {{"a": 2}}, {{"a": 3}}, {{"a": 4}}, {{"a": 5}}, {{"a": 6}}, {{"a": 7}}, {{"a": 8}}, {{"a": 9}}, {{"a": 10}}, {{"a": 11}}, {{"a": 12}}, {{"a": 13}},
+			length: 0,
+		},
+		{
+			name:   "read in write buffer cont 2",
+			inputs: tuples[6:7],
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 6,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
 			},
-			dataOut: [][]map[string]interface{}{
-				{{"a": 1}}, {{"a": 6}}, {{"a": 7}}, {{"a": 8}}, {{"a": 9}}, {{"a": 10}}, {{"a": 11}}, {{"a": 12}}, {{"a": 13}},
+			length: 0,
+		},
+		{
+			name:   "write more",
+			inputs: tuples[7:10],
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 7,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
 			},
-			stopPt: 4,
+			length: 2,
+		},
+		{
+			name:   "write over",
+			inputs: tuples[10:11],
+			output: &xsql.RawTuple{
+				Emitter:   "test",
+				Timestamp: 8,
+				Rawdata:   []byte("hello"),
+				Metadata:  map[string]any{"topic": "demo"},
+			},
+			length: 2,
 		},
 	}
-	testx.InitEnv("cache")
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	tempStore, _ := state.CreateStore("mock", def.AtMostOnce)
-	deleteCachedb()
-	for i, tt := range tests {
-		contextLogger := conf.Log.WithField("rule", fmt.Sprintf("TestRun-%d", i))
-		ctx, cancel := context.WithValue(context.Background(), context.LoggerKey, contextLogger).WithMeta(fmt.Sprintf("rule%d", i), fmt.Sprintf("op%d", i), tempStore).WithCancel()
-		in := make(chan []map[string]interface{})
-		errCh := make(chan error)
-		var result []interface{}
-		go func() {
-			err := <-errCh
-			t.Log(err)
-			return
-		}()
-		exitCh := make(chan struct{})
-		// send data
-		_ = NewSyncCacheWithExitChanel(ctx, in, errCh, tt.sconf, 100, exitCh)
-		for i := 0; i < tt.stopPt; i++ {
-			in <- tt.dataIn[i]
-			time.Sleep(1 * time.Millisecond)
-		}
-		cancel()
-		// wait a cleanup job done
-		<-exitCh
-
-		// send the second half data
-		ctx, cancel = context.WithValue(context.Background(), context.LoggerKey, contextLogger).WithMeta(fmt.Sprintf("rule%d", i), fmt.Sprintf("op%d", i), tempStore).WithCancel()
-		sc := NewSyncCache(ctx, in, errCh, tt.sconf, 100)
-		for i := tt.stopPt; i < len(tt.dataIn); i++ {
-			in <- tt.dataIn[i]
-			time.Sleep(1 * time.Millisecond)
-		}
-	loop:
-		for range tt.dataIn {
-			sc.Ack <- true
-			select {
-			case r := <-sc.Out:
-				result = append(result, r)
-			case <-time.After(1 * time.Second):
-				t.Log(fmt.Sprintf("test %d no data", i))
-				break loop
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			for _, tuple := range tt.inputs {
+				err = s.AddCache(ctx, tuple)
+				assert.NoError(t, err)
 			}
-		}
+			r, _ := s.PopCache(ctx)
+			assert.Equal(t, tt.output, r)
+			assert.Equal(t, tt.length, s.CacheLength, "cache length")
+		})
+	}
+}
 
-		cancel()
-		if tt.dataOut == nil {
-			tt.dataOut = tt.dataIn
-		}
-		if len(tt.dataOut) != len(result) {
-			t.Errorf("test %d data mismatch\nexpect\t%v\nbut got\t%v", i, tt.dataOut, result)
-			continue
-		}
-		for i, v := range result {
-			if !reflect.DeepEqual(tt.dataOut[i], v) {
-				t.Errorf("test %d data mismatch\nexpect\t%v\nbut got\t%v", i, tt.dataOut, result)
-				break
-			}
+func TestCacheInit(t *testing.T) {
+	// Test flush and reload
+	testx.InitEnv("cache3")
+	tempStore, err := state.CreateStore("mock", def.AtMostOnce)
+	assert.NoError(t, err)
+	deleteCachedb()
+	contextLogger := conf.Log.WithField("rule", "TestCache")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger).WithMeta("TestCache", "op1", tempStore)
+	s, err := NewSyncCache(ctx, &conf.SinkConf{
+		MemoryCacheThreshold: 0,
+		MaxDiskCache:         4,
+		BufferPageSize:       2,
+		EnableCache:          true,
+		ResendInterval:       0,
+		CleanCacheAtStop:     false,
+	})
+	assert.NoError(t, err)
+	// prepare data
+	tuples := make([]any, 10)
+	for i := 0; i < 10; i++ {
+		tuples[i] = &xsql.RawTuple{
+			Emitter:   "test",
+			Timestamp: int64(i),
+			Rawdata:   []byte("hello"),
+			Metadata:  map[string]any{"topic": "demo"},
 		}
+		err = s.AddCache(ctx, tuples[i])
+		assert.NoError(t, err)
 	}
+	assert.Equal(t, 6, s.CacheLength, "cache length before flush")
+	s.Flush(ctx)
+	s = nil
+	s, err = NewSyncCache(ctx, &conf.SinkConf{
+		MemoryCacheThreshold: 0,
+		MaxDiskCache:         4,
+		BufferPageSize:       2,
+		EnableCache:          true,
+		ResendInterval:       0,
+		CleanCacheAtStop:     false,
+	})
+	assert.NoError(t, err)
+	r, _ := s.PopCache(ctx)
+	assert.Equal(t, 3, s.CacheLength, "cache length after pop")
+	assert.Equal(t, &xsql.RawTuple{
+		Emitter:   "test",
+		Timestamp: int64(6),
+		Rawdata:   []byte("hello"),
+		Metadata:  map[string]any{"topic": "demo"},
+	}, r)
+	s.Flush(ctx)
+	s = nil
+	s, err = NewSyncCache(ctx, &conf.SinkConf{
+		MemoryCacheThreshold: 0,
+		MaxDiskCache:         4,
+		BufferPageSize:       2,
+		EnableCache:          true,
+		ResendInterval:       0,
+		CleanCacheAtStop:     false,
+	})
+	assert.NoError(t, err)
+	r, _ = s.PopCache(ctx)
+	assert.Equal(t, 2, s.CacheLength, "cache length after pop")
+	assert.Equal(t, &xsql.RawTuple{
+		Emitter:   "test",
+		Timestamp: int64(7),
+		Rawdata:   []byte("hello"),
+		Metadata:  map[string]any{"topic": "demo"},
+	}, r)
+	s.cacheConf.CleanCacheAtStop = true
+	s.Flush(ctx)
+	s = nil
+	s, err = NewSyncCache(ctx, &conf.SinkConf{
+		MemoryCacheThreshold: 0,
+		MaxDiskCache:         2,
+		BufferPageSize:       2,
+		EnableCache:          true,
+		ResendInterval:       0,
+		CleanCacheAtStop:     true,
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, 0, s.CacheLength, "cache length after clean")
 }
 
 func deleteCachedb() {
diff --git a/internal/topo/node/cache_op_test.go b/internal/topo/node/cache_op_test.go
new file mode 100644
index 00000000..823b445c
--- /dev/null
+++ b/internal/topo/node/cache_op_test.go
@@ -0,0 +1,192 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package node
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/exp/maps"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+func TestCacheRun(t *testing.T) {
+	testx.InitEnv("cacheOp")
+	deleteCachedb()
+	timex.Set(0)
+	// prepare data
+	tuples := make([]any, 20)
+	for i := 0; i < 20; i++ {
+		tuples[i] = &xsql.Tuple{
+			Emitter:   "test",
+			Timestamp: int64(i),
+			Message:   map[string]any{"key": "value"},
+			Metadata:  map[string]any{"topic": "demo"},
+		}
+	}
+
+	tests := []struct {
+		name         string
+		sendUntil    int
+		receiveCount int
+		lastReceive  any
+	}{
+		{ // 0
+			name:         "in channel",
+			sendUntil:    2,
+			receiveCount: 1,
+			lastReceive:  tuples[0],
+		},
+		{ // 1
+			name:         "in disk buffer",
+			sendUntil:    5,
+			receiveCount: 3,
+			lastReceive:  tuples[3],
+		},
+		{
+			name:         "disk overflow",
+			sendUntil:    19,
+			receiveCount: 2,
+			lastReceive:  tuples[5],
+		},
+		{
+			name: "pull by time",
+			// no send
+			sendUntil:    19,
+			receiveCount: 1,
+			lastReceive:  tuples[6],
+		},
+		{
+			name: "pull by time, receive after dropped tuple",
+			// no send
+			sendUntil:    19,
+			receiveCount: 1,
+			lastReceive:  tuples[13],
+		},
+		{
+			name: "receive all",
+			// no send
+			sendUntil:    19,
+			receiveCount: 5,
+			lastReceive:  tuples[18],
+		},
+		{
+			name: "send in no buffer",
+			// no send
+			sendUntil:    20,
+			receiveCount: 1,
+			lastReceive:  tuples[19],
+		},
+	}
+
+	ctx := mockContext.NewMockContext("testCache", "op1")
+	cacheOp, err := NewCacheOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &conf.SinkConf{
+		MemoryCacheThreshold: 2,
+		MaxDiskCache:         4,
+		BufferPageSize:       2,
+		EnableCache:          true,
+		ResendInterval:       10,
+	})
+	assert.NoError(t, err)
+	// In sink_node planner, set this buffer length
+	out := make(chan any, 2)
+	err = cacheOp.AddOutput(out, "test")
+	assert.NoError(t, err)
+	index := 0
+	errCh := make(chan error)
+	cacheOp.Exec(ctx, errCh)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var r any
+			for index < tt.sendUntil {
+				cacheOp.input <- tuples[index]
+				index++
+			}
+			timex.Add(100 * time.Millisecond)
+			for { // wait until all processed
+				processed := cacheOp.statManager.GetMetrics()[2]
+				if processed == int64(tt.sendUntil) {
+					break
+				}
+				time.Sleep(10 * time.Millisecond)
+			}
+			if tt.receiveCount > 0 {
+				for j := 0; j < tt.receiveCount-1; j++ {
+					a := <-out
+					timex.Add(20 * time.Millisecond)
+					ctx.GetLogger().Infof("receive %d", a.(*xsql.Tuple).Timestamp)
+					if j%2 == 0 { // because channel length is 2, so need to wait for each 2
+						time.Sleep(10 * time.Millisecond)
+					}
+				}
+				r = <-out
+				ctx.GetLogger().Infof("receive %d", r.(*xsql.Tuple).Timestamp)
+			}
+			assert.Equal(t, tt.lastReceive, r)
+		})
+	}
+}
+
+func TestRunError(t *testing.T) {
+	ctx, cancel := mockContext.NewMockContext("testError", "op1").WithCancel()
+	// Test multiple output error
+	testx.InitEnv("cacheErr")
+	op, err := NewCacheOp(ctx, "test", &def.RuleOption{BufferLength: 10, SendError: true}, &conf.SinkConf{
+		MemoryCacheThreshold: 2,
+		MaxDiskCache:         4,
+		BufferPageSize:       2,
+		EnableCache:          true,
+		ResendInterval:       10,
+	})
+	assert.NoError(t, err)
+	err = op.AddOutput(make(chan any, 2), "output1")
+	assert.NoError(t, err)
+	err = op.AddOutput(make(chan any, 2), "output2")
+	assert.NoError(t, err)
+	errCh := make(chan error, 1)
+	op.Exec(ctx, errCh)
+	err = <-errCh
+	assert.Error(t, err)
+	assert.True(t, strings.Contains(err.Error(), "cache op should have only 1 output but got"), err.Error())
+	// Test done
+	maps.Clear(op.outputs)
+	err = op.AddOutput(make(chan any, 2), "output1")
+	assert.NoError(t, err)
+	op.Exec(ctx, errCh)
+	cancel()
+	assert.Equal(t, 0, len(errCh))
+}
+
+func deleteCachedb() {
+	loc, err := conf.GetDataLoc()
+	if err != nil {
+		fmt.Println(err)
+	}
+	err = os.RemoveAll(filepath.Join(loc, "cache.db"))
+	if err != nil {
+		fmt.Println(err)
+	}
+}
diff --git a/internal/topo/node/encrypt_op_test.go b/internal/topo/node/encrypt_op_test.go
index b3996fea..569c4667 100644
--- a/internal/topo/node/encrypt_op_test.go
+++ b/internal/topo/node/encrypt_op_test.go
@@ -30,6 +30,9 @@ func TestNewEncryptOp(t *testing.T) {
 	_, err := NewEncryptOp("test", &def.RuleOption{}, "non")
 	assert.Error(t, err)
 	assert.Equal(t, "get encryptor non fail with error: unsupported encryptor: non", err.Error())
+	if conf.Config != nil {
+		conf.Config.AesKey = nil
+	}
 	_, err = NewEncryptOp("test", &def.RuleOption{}, "aes")
 	assert.Error(t, err)
 	assert.Equal(t, errors.New("get encryptor aes fail with error: AES key is not defined"), err)
diff --git a/internal/topo/planner/planner_sink_test.go b/internal/topo/planner/planner_sink_test.go
index 95fa307a..b5b96817 100644
--- a/internal/topo/planner/planner_sink_test.go
+++ b/internal/topo/planner/planner_sink_test.go
@@ -122,13 +122,14 @@ func TestSinkPlan(t *testing.T) {
 			},
 		},
 		{
-			name: "encrypt and compress sink plan",
+			name: "encrypt and compress and cache sink plan",
 			rule: &def.Rule{
 				Actions: []map[string]any{
 					{
 						"log": map[string]any{
 							"compression": "gzip",
 							"encryption":  "aes",
+							"enableCache": true,
 						},
 					},
 				},
@@ -150,6 +151,9 @@ func TestSinkPlan(t *testing.T) {
 						"op_log_0_3_encrypt",
 					},
 					"op_log_0_3_encrypt": {
+						"op_log_0_4_cache",
+					},
+					"op_log_0_4_cache": {
 						"sink_log_0",
 					},
 				},