Repository: lf-edge/ekuiper
Pull request #2811 (2024-04-26 01:50:01)

feat(sink): add back file sink


Patch:
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index aab9427a..a25634e7 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -16,6 +16,7 @@ package io
 
 import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/file"
 	"github.com/lf-edge/ekuiper/v2/internal/io/memory"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt"
 	"github.com/lf-edge/ekuiper/v2/internal/io/sink"
@@ -39,7 +40,7 @@ func init() {
 	modules.RegisterSink("nop", func() api.Sink { return &sink.NopSink{} })
 	modules.RegisterSink("memory", func() api.Sink { return memory.GetSink() })
 	// modules.RegisterSink("neuron", func() api.Sink { return neuron.GetSink() })
-	// modules.RegisterSink("file", func() api.Sink { return file.File() })
+	modules.RegisterSink("file", file.GetSink)
 	// modules.RegisterSink("websocket", func() api.Sink { return &websocket.WebSocketSink{} })
 
 	// modules.RegisterLookupSource("memory", func() api.LookupSource { return memory.GetLookupSource() })
diff --git a/internal/converter/binary/converter.go b/internal/converter/binary/converter.go
index 6ff4e8f8..670b1789 100644
--- a/internal/converter/binary/converter.go
+++ b/internal/converter/binary/converter.go
@@ -17,6 +17,7 @@ package binary
 import (
 	"fmt"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 )
@@ -29,7 +30,7 @@ func GetConverter() (message.Converter, error) {
 	return converter, nil
 }
 
-func (c *Converter) Encode(d interface{}) (b []byte, err error) {
+func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -39,7 +40,7 @@ func (c *Converter) Encode(d interface{}) (b []byte, err error) {
 	return nil, fmt.Errorf("not supported")
 }
 
-func (c *Converter) Decode(b []byte) (m interface{}, err error) {
+func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 	result := make(map[string]interface{})
 	result[message.DefaultField] = b
 	return result, nil
diff --git a/internal/converter/custom/converter.go b/internal/converter/custom/converter.go
deleted file mode 100644
index 0cac3f1e..00000000
--- a/internal/converter/custom/converter.go
+++ /dev/null
@@ -1,41 +0,0 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package custom
-
-import (
-	"fmt"
-
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/converter/static"
-	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
-	"github.com/lf-edge/ekuiper/v2/internal/schema"
-	"github.com/lf-edge/ekuiper/v2/pkg/message"
-)
-
-type Converter struct{}
-
-var converter = &Converter{}
-
-func LoadConverter(schemaFile string, messageName string, _ string) (message.Converter, error) {
-	conf.Log.Infof("Load custom converter from file %s, for symbol Get%s", schemaFile, messageName)
-	ffs, err := schema.GetSchemaFile(def.CUSTOM, schemaFile)
-	if err != nil {
-		return nil, err
-	}
-	if ffs.SoFile == "" {
-		return nil, fmt.Errorf("no so file found for custom schema %s", messageName)
-	}
-	return static.LoadStaticConverter(ffs.SoFile, messageName)
-}
diff --git a/internal/converter/delimited/converter.go b/internal/converter/delimited/converter.go
index 88e434b0..e9160b42 100644
--- a/internal/converter/delimited/converter.go
+++ b/internal/converter/delimited/converter.go
@@ -21,6 +21,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 )
@@ -42,7 +43,7 @@ func (c *Converter) SetColumns(cols []string) {
 }
 
 // Encode If no columns defined, the default order is sort by key
-func (c *Converter) Encode(d interface{}) (b []byte, err error) {
+func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -74,7 +75,7 @@ func (c *Converter) Encode(d interface{}) (b []byte, err error) {
 
 // Decode If the cols is not set, the default key name is col1, col2, col3...
 // The return value is always a map
-func (c *Converter) Decode(b []byte) (ma interface{}, err error) {
+func (c *Converter) Decode(ctx api.StreamContext, b []byte) (ma any, err error) {
 	tokens := strings.Split(string(b), c.delimiter)
 	m := make(map[string]interface{})
 	if len(c.cols) == 0 {
diff --git a/internal/converter/ext_converter.go b/internal/converter/ext_converter.go
index 4166d326..28564c27 100644
--- a/internal/converter/ext_converter.go
+++ b/internal/converter/ext_converter.go
@@ -17,7 +17,6 @@
 package converter
 
 import (
-	"github.com/lf-edge/ekuiper/v2/internal/converter/custom"
 	"github.com/lf-edge/ekuiper/v2/internal/converter/protobuf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/schema"
@@ -33,5 +32,4 @@ func init() {
 		}
 		return protobuf.NewConverter(ffs.SchemaFile, ffs.SoFile, schemaMessageName)
 	})
-	modules.RegisterConverter(message.FormatCustom, custom.LoadConverter)
 }
diff --git a/internal/converter/ext_mock.go b/internal/converter/ext_mock.go
index b3f34b8c..dd6fcdc5 100644
--- a/internal/converter/ext_mock.go
+++ b/internal/converter/ext_mock.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
@@ -34,13 +35,13 @@ func init() {
 // MockConverter mocks a slow converter for benchmark test
 type MockConverter struct{}
 
-func (m MockConverter) Encode(d interface{}) ([]byte, error) {
+func (m MockConverter) Encode(ctx api.StreamContext, d any) ([]byte, error) {
 	time.Sleep(10 * time.Millisecond)
 	now := timex.GetNowInMilli()
 	return []byte(fmt.Sprintf(`{"temperature":23.4,"humidity":76,"ts": %d}`, now)), nil
 }
 
-func (m MockConverter) Decode(b []byte) (interface{}, error) {
+func (m MockConverter) Decode(ctx api.StreamContext, b []byte) (any, error) {
 	time.Sleep(10 * time.Millisecond)
 	return map[string]any{
 		"temperature": 23.4,
diff --git a/internal/converter/json/converter.go b/internal/converter/json/converter.go
index bce44941..e1f9ef94 100644
--- a/internal/converter/json/converter.go
+++ b/internal/converter/json/converter.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/valyala/fastjson"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/converter/merge"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
@@ -36,11 +37,11 @@ func GetConverter() (message.Converter, error) {
 	return converter, nil
 }
 
-func (c *Converter) Encode(d interface{}) (b []byte, err error) {
+func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	return json.Marshal(d)
 }
 
-func (c *Converter) Decode(b []byte) (m interface{}, err error) {
+func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -144,11 +145,11 @@ func mergeSchema(originSchema, newSchema map[string]*ast.JsonStreamField) (map[s
 	return merge.MergeSchema(originSchema, newSchema)
 }
 
-func (c *FastJsonConverter) Encode(d interface{}) (b []byte, err error) {
+func (c *FastJsonConverter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	return json.Marshal(d)
 }
 
-func (c *FastJsonConverter) Decode(b []byte) (m interface{}, err error) {
+func (c *FastJsonConverter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -157,7 +158,7 @@ func (c *FastJsonConverter) Decode(b []byte) (m interface{}, err error) {
 	c.RLock()
 	defer c.RUnlock()
 	if len(c.wildcardMap) > 0 {
-		return converter.Decode(b)
+		return converter.Decode(ctx, b)
 	}
 	return c.decodeWithSchema(b, c.schema)
 }
diff --git a/internal/converter/protobuf/converter.go b/internal/converter/protobuf/converter.go
index dc788e19..c2113f36 100644
--- a/internal/converter/protobuf/converter.go
+++ b/internal/converter/protobuf/converter.go
@@ -20,6 +20,7 @@ import (
 	"github.com/jhump/protoreflect/desc"
 	"github.com/jhump/protoreflect/desc/protoparse"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	kconf "github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/converter/static"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
@@ -58,7 +59,7 @@ func NewConverter(schemaFile string, soFile string, messageName string) (message
 	}
 }
 
-func (c *Converter) Encode(d interface{}) (b []byte, err error) {
+func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
@@ -76,7 +77,7 @@ func (c *Converter) Encode(d interface{}) (b []byte, err error) {
 	}
 }
 
-func (c *Converter) Decode(b []byte) (m interface{}, err error) {
+func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
diff --git a/internal/encryptor/encryptor.go b/internal/encryptor/encryptor.go
index d150cfde..40196e6d 100644
--- a/internal/encryptor/encryptor.go
+++ b/internal/encryptor/encryptor.go
@@ -17,6 +17,7 @@ package encryptor
 import (
 	"crypto/rand"
 	"fmt"
+	"io"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/encryptor/aes"
@@ -25,19 +26,33 @@ import (
 
 func GetEncryptor(name string) (message.Encryptor, error) {
 	if name == "aes" {
-		if conf.Config == nil || conf.Config.AesKey == nil {
-			return nil, fmt.Errorf("AES key is not defined")
+		key, iv, err := GetKeyIv()
+		if err != nil {
+			return nil, err
 		}
-		key, iv := getKeyIv()
 		return aes.NewStreamEncrypter(key, iv)
 	}
 	return nil, fmt.Errorf("unsupported encryptor: %s", name)
 }
 
-func getKeyIv() ([]byte, []byte) {
+func GetEncryptWriter(name string, output io.Writer) (io.Writer, error) {
+	if name == "aes" {
+		key, iv, err := GetKeyIv()
+		if err != nil {
+			return nil, err
+		}
+		return aes.NewStreamWriter(key, iv, output)
+	}
+	return nil, fmt.Errorf("unsupported encryptor: %s", name)
+}
+
+func GetKeyIv() ([]byte, []byte, error) {
+	if conf.Config == nil || conf.Config.AesKey == nil {
+		return nil, nil, fmt.Errorf("AES key is not defined")
+	}
 	key := conf.Config.AesKey
 	iv := make([]byte, 16)
 	// Use the crypto/rand package to generate random bytes
 	_, _ = rand.Read(iv)
-	return key, iv
+	return key, iv, nil
 }
diff --git a/internal/topo/context/transform.go b/internal/io/file/constant.go
similarity index 56%
rename from internal/topo/context/transform.go
rename to internal/io/file/constant.go
index 64075369..b1166c21 100644
--- a/internal/topo/context/transform.go
+++ b/internal/io/file/constant.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2023 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,22 +12,28 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package context
+package file
 
-import (
-	"fmt"
+type FileType string
 
-	"github.com/lf-edge/ekuiper/v2/internal/topo/transform"
+const (
+	JSON_TYPE  FileType = "json"
+	CSV_TYPE   FileType = "csv"
+	LINES_TYPE FileType = "lines"
 )
 
-const TransKey = "$$trans"
+const (
+	GZIP = "gzip"
+	ZSTD = "zstd"
+)
+
+var fileTypes = map[FileType]struct{}{
+	JSON_TYPE:  {},
+	CSV_TYPE:   {},
+	LINES_TYPE: {},
+}
 
-// TransformOutput Lazy transform output to bytes
-func (c *DefaultContext) TransformOutput(data interface{}) ([]byte, bool, error) {
-	v := c.Value(TransKey)
-	f, ok := v.(transform.TransFunc)
-	if ok {
-		return f(data)
-	}
-	return nil, false, fmt.Errorf("no transform configured")
+var compressionTypes = map[string]struct{}{
+	GZIP: {},
+	ZSTD: {},
 }
diff --git a/internal/io/file/file_writer.go b/internal/io/file/file_writer.go
new file mode 100644
index 00000000..bf09d429
--- /dev/null
+++ b/internal/io/file/file_writer.go
@@ -0,0 +1,141 @@
+// Copyright 2023 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package file
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/compressor"
+	"github.com/lf-edge/ekuiper/v2/internal/encryptor"
+	"github.com/lf-edge/ekuiper/v2/internal/io/file/writer"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type fileWriter struct {
+	File       *os.File
+	Writer     io.Writer
+	Hook       writerHooks
+	Start      time.Time
+	Count      int
+	Compress   string
+	fileBuffer *writer.BufioWrapWriter
+	// Whether the file has written any data. It is only used to determine if new line is needed when writing data.
+	Written bool
+}
+
+func (m *fileSink) createFileWriter(ctx api.StreamContext, fn string, ft FileType, headers string, compressAlgorithm string, encryption string) (_ *fileWriter, ge error) {
+	ctx.GetLogger().Infof("Create new file writer for %s", fn)
+	fws := &fileWriter{Start: timex.GetNow()}
+	var (
+		f   *os.File
+		err error
+	)
+	Dir := filepath.Dir(fn)
+	if _, err = os.Stat(Dir); os.IsNotExist(err) {
+		if err := os.Mkdir(Dir, 0o777); err != nil {
+			return nil, fmt.Errorf("fail to create file %s: %v", fn, err)
+		}
+	}
+
+	if _, err = os.Stat(fn); os.IsNotExist(err) {
+		if _, err := os.Create(fn); err != nil {
+			return nil, fmt.Errorf("fail to create file %s: %v", fn, err)
+		}
+	}
+	f, err = os.OpenFile(fn, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, os.ModeAppend)
+	if err != nil {
+		return nil, fmt.Errorf("fail to open file sink for %s: %v", fn, err)
+	}
+	defer func() {
+		if ge != nil {
+			_ = f.Close()
+		}
+	}()
+	fws.File = f
+	switch ft {
+	case JSON_TYPE:
+		fws.Hook = jsonHooks
+	case CSV_TYPE:
+		fws.Hook = &csvWriterHooks{header: []byte(headers)}
+	case LINES_TYPE:
+		fws.Hook = linesHooks
+	}
+
+	fws.fileBuffer = writer.NewBufioWrapWriter(bufio.NewWriter(f))
+	var currWriter io.Writer = fws.fileBuffer
+	fws.Writer, err = m.CreateWriter(ctx, currWriter, compressAlgorithm, encryption)
+	if err != nil {
+		return nil, err
+	}
+	_, err = fws.Writer.Write(fws.Hook.Header())
+	if err != nil {
+		return nil, err
+	}
+	return fws, nil
+}
+
+func (m *fileSink) CreateWriter(_ api.StreamContext, currWriter io.Writer, compression string, encryption string) (io.Writer, error) {
+	var err error
+	if encryption != "" {
+		currWriter, err = encryptor.GetEncryptWriter(encryption, currWriter)
+		if err != nil {
+			return nil, fmt.Errorf("fail to get encrypt writer for %s: %v", encryption, err)
+		}
+	}
+	if compression != "" {
+		currWriter, err = compressor.GetCompressWriter(compression, currWriter)
+		if err != nil {
+			return nil, fmt.Errorf("fail to get compress writer for %s: %v", compression, err)
+		}
+	}
+	return currWriter, nil
+}
+
+func (fw *fileWriter) Close(ctx api.StreamContext) error {
+	var err error
+	if fw.File != nil {
+		ctx.GetLogger().Debugf("File sync before close")
+		_, e := fw.Writer.Write(fw.Hook.Footer())
+		if e != nil {
+			ctx.GetLogger().Errorf("file sink fails to write footer with error %s.", e)
+		}
+
+		// Close the compressor and encryptor firstly
+		if w, ok := fw.Writer.(io.Closer); ok {
+			e := w.Close()
+			if e != nil {
+				ctx.GetLogger().Errorf("file sink fails to close compress/encrypt writer with error %s.", err)
+			}
+		}
+		err = fw.fileBuffer.Flush()
+		if err != nil {
+			ctx.GetLogger().Errorf("file sink fails to flush with error %s.", err)
+		}
+
+		err = fw.File.Sync()
+		if err != nil {
+			ctx.GetLogger().Errorf("file sink fails to sync with error %s.", err)
+		}
+		ctx.GetLogger().Infof("Close file %s", fw.File.Name())
+		return fw.File.Close()
+	}
+	return nil
+}
diff --git a/internal/io/file/sink.go b/internal/io/file/sink.go
new file mode 100644
index 00000000..61cefe9f
--- /dev/null
+++ b/internal/io/file/sink.go
@@ -0,0 +1,264 @@
+// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package file
+
+import (
+	"errors"
+	"fmt"
+	"path/filepath"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/message"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type sinkConf struct {
+	RollingInterval    int64    `json:"rollingInterval"`
+	RollingCount       int      `json:"rollingCount"`
+	RollingNamePattern string   `json:"rollingNamePattern"` // where to add the timestamp to the file name
+	CheckInterval      int64    `json:"checkInterval"`
+	Path               string   `json:"path"` // support dynamic property, when rolling, make sure the path is updated
+	FileType           FileType `json:"fileType"`
+	HasHeader          bool     `json:"hasHeader"`
+	Delimiter          string   `json:"delimiter"`
+	Format             string   `json:"format"` // only use for validation; transformation is done in sink_node
+	Compression        string   `json:"compression"`
+	Encryption         string   `json:"encryption"`
+	Fields             []string `json:"fields"` // only use for extracting header for csv; transformation is done in sink_node
+}
+
+type fileSink struct {
+	c *sinkConf
+
+	mux sync.Mutex
+	fws map[string]*fileWriter
+}
+
+func (m *fileSink) Provision(ctx api.StreamContext, props map[string]interface{}) error {
+	c := &sinkConf{
+		RollingCount:  1000000,
+		Path:          "cache",
+		FileType:      LINES_TYPE,
+		CheckInterval: (5 * time.Minute).Milliseconds(),
+	}
+	if err := cast.MapToStruct(props, c); err != nil {
+		return err
+	}
+	if c.RollingInterval < 0 {
+		return fmt.Errorf("rollingInterval must be positive")
+	}
+	if c.RollingCount < 0 {
+		return fmt.Errorf("rollingCount must be positive")
+	}
+
+	if c.CheckInterval < 0 {
+		return fmt.Errorf("checkInterval must be positive")
+	}
+	if c.RollingInterval == 0 && c.RollingCount == 0 {
+		return fmt.Errorf("one of rollingInterval and rollingCount must be set")
+	}
+	if c.RollingNamePattern != "" && c.RollingNamePattern != "prefix" && c.RollingNamePattern != "suffix" && c.RollingNamePattern != "none" {
+		return fmt.Errorf("rollingNamePattern must be one of prefix, suffix or none")
+	}
+	if c.Path == "" {
+		return fmt.Errorf("path must be set")
+	}
+	if c.FileType != JSON_TYPE && c.FileType != CSV_TYPE && c.FileType != LINES_TYPE {
+		return fmt.Errorf("fileType must be one of json, csv or lines")
+	}
+	if c.FileType == CSV_TYPE {
+		if c.Format != message.FormatDelimited {
+			return fmt.Errorf("format must be delimited when fileType is csv")
+		}
+		if c.Delimiter == "" {
+			ctx.GetLogger().Warnf("delimiter is not set, use default ','")
+			c.Delimiter = ","
+		}
+	}
+
+	if _, ok := compressionTypes[c.Compression]; !ok && c.Compression != "" {
+		return fmt.Errorf("compression must be one of gzip, zstd")
+	}
+
+	m.c = c
+	m.fws = make(map[string]*fileWriter)
+	return nil
+}
+
+func (m *fileSink) Connect(ctx api.StreamContext) error {
+	ctx.GetLogger().Debug("Opening file sink")
+	// 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() {
+			defer t.Stop()
+			for {
+				select {
+				case now := <-t.C:
+					m.mux.Lock()
+					for k, v := range m.fws {
+						if now.Sub(v.Start) > time.Duration(m.c.RollingInterval)*time.Millisecond {
+							ctx.GetLogger().Debugf("rolling file %s", k)
+							err := v.Close(ctx)
+							// TODO how to inform this error to the rule
+							if err != nil {
+								ctx.GetLogger().Errorf("file sink fails to close file %s with error %s.", k, err)
+							}
+							delete(m.fws, k)
+							// The file will be created when the next item comes
+							v.Written = false
+						}
+					}
+					m.mux.Unlock()
+				case <-ctx.Done():
+					ctx.GetLogger().Info("file sink done")
+					return
+				}
+			}
+		}()
+	}
+	return nil
+}
+
+func (m *fileSink) Collect(ctx api.StreamContext, item []byte) error {
+	ctx.GetLogger().Debugf("file sink receive %s", item)
+	fn, err := ctx.ParseTemplate(m.c.Path, item)
+	if err != nil {
+		return err
+	}
+	fw, err := m.GetFws(ctx, fn, item)
+	if err != nil {
+		return err
+	}
+
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	if fw.Written {
+		_, e := fw.Writer.Write(fw.Hook.Line())
+		if e != nil {
+			return e
+		}
+	} else {
+		fw.Written = true
+	}
+	_, e := fw.Writer.Write(item)
+	if e != nil {
+		return e
+	}
+	if m.c.RollingCount > 0 {
+		fw.Count++
+		if fw.Count >= m.c.RollingCount {
+			e = fw.Close(ctx)
+			if e != nil {
+				return e
+			}
+			delete(m.fws, fn)
+			fw.Count = 0
+			fw.Written = false
+		}
+	}
+
+	return nil
+}
+
+func (m *fileSink) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Closing file sink")
+	var errs []error
+	for k, v := range m.fws {
+		if e := v.Close(ctx); e != nil {
+			ctx.GetLogger().Errorf("failed to close file %s: %v", k, e)
+			errs = append(errs, e)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// GetFws returns the file writer for the given file name, if the file writer does not exist, it will create one
+// The item is used to get the csv header if needed
+func (m *fileSink) GetFws(ctx api.StreamContext, fn string, item interface{}) (*fileWriter, error) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	fws, ok := m.fws[fn]
+	if !ok {
+		var e error
+		// extract header for csv
+		var headers string
+		if m.c.FileType == CSV_TYPE && m.c.HasHeader {
+			var header []string
+			if len(m.c.Fields) > 0 {
+				header = m.c.Fields
+			} else {
+				switch v := item.(type) {
+				case map[string]interface{}:
+					header = make([]string, len(v))
+					i := 0
+					for k := range item.(map[string]interface{}) {
+						header[i] = k
+						i++
+					}
+				case []map[string]interface{}:
+					if len(v) > 0 {
+						header = make([]string, len(v[0]))
+						i := 0
+						for k := range v[0] {
+							header[i] = k
+							i++
+						}
+					}
+				}
+				sort.Strings(header)
+			}
+			headers = strings.Join(header, m.c.Delimiter)
+		}
+		nfn := fn
+		if m.c.RollingNamePattern != "" {
+			newFile := ""
+			fileDir := filepath.Dir(fn)
+			fileName := filepath.Base(fn)
+			switch m.c.RollingNamePattern {
+			case "prefix":
+				newFile = fmt.Sprintf("%d-%s", timex.GetNowInMilli(), fileName)
+			case "suffix":
+				ext := filepath.Ext(fn)
+				newFile = fmt.Sprintf("%s-%d%s", strings.TrimSuffix(fileName, ext), timex.GetNowInMilli(), ext)
+			default:
+				newFile = fileName
+			}
+			nfn = filepath.Join(fileDir, newFile)
+		}
+
+		fws, e = m.createFileWriter(ctx, nfn, m.c.FileType, headers, m.c.Compression, m.c.Encryption)
+		if e != nil {
+			return nil, e
+		}
+		m.fws[fn] = fws
+	}
+	return fws, nil
+}
+
+func GetSink() api.Sink {
+	return &fileSink{}
+}
+
+var (
+	_ api.BytesCollector = &fileSink{}
+	_ model.StreamWriter = &fileSink{}
+)
diff --git a/internal/io/file/writer/bufioall.go b/internal/io/file/writer/bufioall.go
new file mode 100644
index 00000000..99361890
--- /dev/null
+++ b/internal/io/file/writer/bufioall.go
@@ -0,0 +1,39 @@
+// Copyright 2023 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package writer
+
+import "bufio"
+
+// BufioWrapWriter wraps bufio writer to support flush all once buffer is full
+type BufioWrapWriter struct {
+	w *bufio.Writer
+}
+
+func NewBufioWrapWriter(w *bufio.Writer) *BufioWrapWriter {
+	return &BufioWrapWriter{
+		w: w,
+	}
+}
+
+func (b *BufioWrapWriter) Write(p []byte) (int, error) {
+	if len(p) > b.w.Available() {
+		_ = b.w.Flush()
+	}
+	return b.w.Write(p)
+}
+
+func (b *BufioWrapWriter) Flush() error {
+	return b.w.Flush()
+}
diff --git a/internal/io/file/writer_hooks.go b/internal/io/file/writer_hooks.go
new file mode 100644
index 00000000..b49b93e8
--- /dev/null
+++ b/internal/io/file/writer_hooks.go
@@ -0,0 +1,78 @@
+// Copyright 2023 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package file
+
+import "bytes"
+
+type writerHooks interface {
+	Header() []byte
+	Line() []byte
+	Footer() []byte
+}
+
+type jsonWriterHooks struct{}
+
+func (j *jsonWriterHooks) Header() []byte {
+	return []byte("[")
+}
+
+func (j *jsonWriterHooks) Line() []byte {
+	return []byte(",")
+}
+
+func (j *jsonWriterHooks) Footer() []byte {
+	return []byte("]")
+}
+
+var jsonHooks = &jsonWriterHooks{}
+
+type linesWriterHooks struct{}
+
+func (l *linesWriterHooks) Header() []byte {
+	return nil
+}
+
+func (l *linesWriterHooks) Line() []byte {
+	return []byte("\n")
+}
+
+func (l *linesWriterHooks) Footer() []byte {
+	return nil
+}
+
+var linesHooks = &linesWriterHooks{}
+
+type csvWriterHooks struct {
+	header []byte
+}
+
+func (c *csvWriterHooks) Header() []byte {
+	if c.header != nil {
+		return bytes.Join([][]byte{c.header, c.Line()}, nil)
+	}
+	return nil
+}
+
+func (c *csvWriterHooks) Line() []byte {
+	return []byte("\n")
+}
+
+func (c *csvWriterHooks) Footer() []byte {
+	return nil
+}
+
+func (c *csvWriterHooks) SetHeader(header string) {
+	c.header = []byte(header)
+}
diff --git a/internal/schema/ext_inferer_custom.go b/internal/schema/ext_inferer_custom.go
deleted file mode 100644
index 6c2a8cd9..00000000
--- a/internal/schema/ext_inferer_custom.go
+++ /dev/null
@@ -1,71 +0,0 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//go:build schema || !core
-
-package schema
-
-import (
-	"encoding/json"
-	"fmt"
-	"plugin"
-
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
-	"github.com/lf-edge/ekuiper/v2/pkg/ast"
-	"github.com/lf-edge/ekuiper/v2/pkg/cast"
-	"github.com/lf-edge/ekuiper/v2/pkg/message"
-)
-
-func init() {
-	inferes[message.FormatCustom] = InferCustom
-}
-
-func InferCustom(schemaFile string, messageName string) (ast.StreamFields, error) {
-	conf.Log.Infof("Load custom schema from file %s, for symbol Get%s", schemaFile, messageName)
-	ffs, err := GetSchemaFile(def.CUSTOM, schemaFile)
-	if err != nil {
-		return nil, err
-	}
-	if ffs.SoFile == "" {
-		return nil, fmt.Errorf("no so file found for custom schema %s", messageName)
-	}
-	sp, err := plugin.Open(ffs.SoFile)
-	if err != nil {
-		conf.Log.Errorf(fmt.Sprintf("custom schema file %s open error: %v", ffs.SoFile, err))
-		return nil, fmt.Errorf("cannot open %s: %v", ffs.SoFile, err)
-	}
-	nf, err := sp.Lookup("Get" + messageName)
-	if err != nil {
-		conf.Log.Warnf(fmt.Sprintf("cannot find schemaId %s, please check if it is exported: Get%v", messageName, err))
-		return nil, nil
-	}
-	nff, ok := nf.(func() interface{})
-	if !ok {
-		conf.Log.Errorf("exported symbol Get%s is not func to return interface{}", messageName)
-		return nil, fmt.Errorf("load custom schema %s, message %s error", ffs.SoFile, messageName)
-	}
-	mc, ok := nff().(message.SchemaProvider)
-	if ok {
-		sj := mc.GetSchemaJson()
-		var result ast.StreamFields
-		err := json.Unmarshal(cast.StringToBytes(sj), &result)
-		if err != nil {
-			return nil, fmt.Errorf("invalid schema json %s: %v", sj, err)
-		}
-		return result, nil
-	} else {
-		return nil, fmt.Errorf("get schema converter failed, exported symbol %s is not type of message.Converter", messageName)
-	}
-}
diff --git a/internal/server/meta_init.go b/internal/server/meta_init.go
index ea6b4d61..83ff0ae2 100644
--- a/internal/server/meta_init.go
+++ b/internal/server/meta_init.go
@@ -13,7 +13,6 @@
 // limitations under the License.
 
 //go:build ui || !core
-// +build ui !core
 
 package server
 
diff --git a/internal/topo/context/decoder.go b/internal/topo/context/decoder.go
deleted file mode 100644
index 36eb4121..00000000
--- a/internal/topo/context/decoder.go
+++ /dev/null
@@ -1,71 +0,0 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package context
-
-import (
-	"fmt"
-
-	"github.com/lf-edge/ekuiper/v2/pkg/message"
-)
-
-const (
-	DecodeKey = "$$decode"
-)
-
-func (c *DefaultContext) Decode(data []byte) (map[string]interface{}, error) {
-	v := c.Value(DecodeKey)
-	f, ok := v.(message.Converter)
-	if ok {
-		t, err := f.Decode(data)
-		if err != nil {
-			return nil, fmt.Errorf("decode failed: %v", err)
-		}
-		if result, ok := t.(map[string]interface{}); ok {
-			return result, nil
-		} else {
-			return nil, fmt.Errorf("only map[string]interface{} is supported but got: %v", t)
-		}
-	}
-	return nil, fmt.Errorf("no decoder configured")
-}
-
-func (c *DefaultContext) DecodeIntoList(data []byte) ([]map[string]interface{}, error) {
-	v := c.Value(DecodeKey)
-	f, ok := v.(message.Converter)
-	if ok {
-		t, err := f.Decode(data)
-		if err != nil {
-			return nil, fmt.Errorf("decode failed: %v", err)
-		}
-		switch r := t.(type) {
-		case map[string]interface{}:
-			return []map[string]interface{}{r}, nil
-		case []map[string]interface{}:
-			return r, nil
-		case []interface{}:
-			rs := make([]map[string]interface{}, len(r))
-			for i, v := range r {
-				if vc, ok := v.(map[string]interface{}); ok {
-					rs[i] = vc
-				} else {
-					return nil, fmt.Errorf("only map[string]interface{} and []map[string]interface{} is supported but got: %v", t)
-				}
-			}
-			return rs, nil
-		}
-		return nil, fmt.Errorf("only map[string]interface{} and []map[string]interface{} is supported but got: %v", t)
-	}
-	return nil, fmt.Errorf("no decoder configured")
-}
diff --git a/internal/topo/node/compress_op.go b/internal/topo/node/compress_op.go
index 5ee1850b..3bf48aac 100644
--- a/internal/topo/node/compress_op.go
+++ b/internal/topo/node/compress_op.go
@@ -53,7 +53,7 @@ func (o *CompressOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *CompressOp) Worker(_ api.Logger, item any) []any {
+func (o *CompressOp) Worker(_ api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
diff --git a/internal/topo/node/concurrent.go b/internal/topo/node/concurrent.go
index 3e600504..b9cdfec3 100644
--- a/internal/topo/node/concurrent.go
+++ b/internal/topo/node/concurrent.go
@@ -22,7 +22,7 @@ import (
 // WorkerFunc is the function to process the data
 // The function do not need to process error and control messages
 // The function must return a slice of data for each input. To omit the data, return nil
-type workerFunc func(logger api.Logger, item any) []any
+type workerFunc func(ctx api.StreamContext, item any) []any
 
 func runWithOrder(ctx api.StreamContext, node *defaultSinkNode, numWorkers int, wf workerFunc) {
 	workerChans := make([]chan any, numWorkers)
@@ -107,7 +107,7 @@ func worker(ctx api.StreamContext, node *defaultSinkNode, i int, wf workerFunc,
 				result = []any{item}
 			default:
 				node.statManager.IncTotalRecordsIn()
-				result = wf(ctx.GetLogger(), item)
+				result = wf(ctx, item)
 			}
 			select {
 			case output <- result:
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index de582abc..0a9b05c0 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -84,14 +84,14 @@ func (o *DecodeOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *DecodeOp) Worker(_ api.Logger, item any) []any {
+func (o *DecodeOp) Worker(ctx api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
 	case error:
 		return []any{d}
 	case *xsql.Tuple:
-		result, err := o.converter.Decode(d.Raw)
+		result, err := o.converter.Decode(ctx, d.Raw)
 		if err != nil {
 			return []any{err}
 		}
diff --git a/internal/topo/node/decompress_op.go b/internal/topo/node/decompress_op.go
index cc56e605..e663d9ce 100644
--- a/internal/topo/node/decompress_op.go
+++ b/internal/topo/node/decompress_op.go
@@ -54,7 +54,7 @@ func (o *DecompressOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *DecompressOp) Worker(_ api.Logger, item any) []any {
+func (o *DecompressOp) Worker(_ api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
diff --git a/internal/topo/node/dedup_trigger_op.go b/internal/topo/node/dedup_trigger_op.go
new file mode 100644
index 00000000..16dc91e8
--- /dev/null
+++ b/internal/topo/node/dedup_trigger_op.go
@@ -0,0 +1,329 @@
+// 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"
+	"strconv"
+	"time"
+
+	"github.com/benbjohnson/clock"
+
+	"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/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type DedupTriggerNode struct {
+	*defaultSinkNode
+	// config
+	aliasName  string
+	startField string
+	endField   string
+	nowField   string
+	expire     int64
+	// state
+	requests      PriorityQueue // All the cached events in order
+	timeoutTicker *clock.Timer
+	timeout       <-chan time.Time
+}
+
+func NewDedupTriggerNode(name string, options *def.RuleOption, aliasName string, startField string, endField string, nowField string, expire int64) *DedupTriggerNode {
+	aname := "dedup_trigger"
+	if aliasName != "" {
+		aname = aliasName
+	}
+	return &DedupTriggerNode{
+		defaultSinkNode: &defaultSinkNode{
+			input: make(chan interface{}, options.BufferLength),
+			defaultNode: &defaultNode{
+				outputs:   make(map[string]chan<- interface{}),
+				name:      name,
+				sendError: options.SendError,
+			},
+		},
+		aliasName:  aname,
+		startField: startField,
+		endField:   endField,
+		expire:     expire,
+		nowField:   nowField,
+		requests:   make(PriorityQueue, 0),
+	}
+}
+
+func (w *DedupTriggerNode) Exec(ctx api.StreamContext, errCh chan<- error) {
+	w.prepareExec(ctx, errCh, "op")
+
+	go func() {
+		err := infra.SafeRun(func() error {
+			for {
+				select {
+				case <-ctx.Done():
+					ctx.GetLogger().Infof("dedup trigger node %s is finished", w.name)
+					return nil
+				case item := <-w.input:
+					data, processed := w.commonIngest(ctx, item)
+					if processed {
+						break
+					}
+					w.statManager.IncTotalRecordsIn()
+					w.statManager.ProcessTimeStart()
+					switch d := data.(type) {
+					case xsql.Row:
+						w.statManager.IncTotalRecordsIn()
+						r, err := w.rowToReq(d)
+						if err != nil {
+							w.Broadcast(err)
+							w.statManager.IncTotalExceptions(err.Error())
+						} else {
+							w.requests.Push(r)
+							w.trigger(ctx, r.now)
+						}
+					default:
+						e := fmt.Errorf("run dedup trigger op error: expect *xsql.Tuple type but got %[1]T(%[1]v)", d)
+						w.Broadcast(e)
+						w.statManager.IncTotalExceptions(e.Error())
+					}
+				// future trigger event
+				case <-w.timeout:
+					w.trigger(ctx, 0)
+				}
+			}
+		})
+		if err != nil {
+			infra.DrainError(ctx, err, errCh)
+		}
+	}()
+}
+
+func (w *DedupTriggerNode) trigger(ctx api.StreamContext, now int64) {
+	for len(w.requests) > 0 {
+		r := w.requests.Peek()
+		ctx.GetLogger().Debugf("dedup trigger node %s trigger event %v", w.name, r)
+		if now == 0 {
+			now = r.end
+		}
+		// trigger by event with timestamp, keep triggering until all history events are triggered
+		if r.end > now {
+			if w.timeoutTicker != nil {
+				w.timeoutTicker.Stop()
+				w.timeoutTicker.Reset(time.Duration(r.end-now) * time.Millisecond)
+			} else {
+				w.timeoutTicker = timex.GetTimer(r.end - now)
+				w.timeout = w.timeoutTicker.C
+				ctx.GetLogger().Debugf("Dedup trigger next trigger time %d", r.end)
+			}
+			break
+		}
+		r = w.requests.Pop()
+		result, err := doTrigger(ctx, r.start, r.end, r.now, r.exp)
+		if err != nil {
+			w.Broadcast(err)
+			w.statManager.IncTotalExceptions(err.Error())
+		} else {
+			w.statManager.ProcessTimeStart()
+			r.tuple.Set(w.aliasName, result)
+			w.Broadcast(r.tuple)
+			ctx.GetLogger().Debug("send out event", r.tuple)
+			w.statManager.IncTotalRecordsOut()
+			w.statManager.ProcessTimeEnd()
+		}
+	}
+}
+
+func (w *DedupTriggerNode) rowToReq(d xsql.Row) (*TriggerRequest, error) {
+	var (
+		begin int64
+		end   int64
+		now   int64
+		err   error
+	)
+	if s, ok := d.Value(w.startField, ""); ok {
+		begin, err = cast.ToInt64(s, cast.CONVERT_SAMEKIND)
+		if err != nil {
+			return nil, fmt.Errorf("dedup_trigger start time %s is not int64", s)
+		}
+	} else {
+		return nil, fmt.Errorf("dedup_trigger %s is missing", w.startField)
+	}
+	if e, ok := d.Value(w.endField, ""); ok {
+		end, err = cast.ToInt64(e, cast.CONVERT_SAMEKIND)
+		if err != nil {
+			return nil, fmt.Errorf("dedup_trigger end time %s is not int64", e)
+		}
+	} else {
+		return nil, fmt.Errorf("dedup_trigger %s is missing", w.endField)
+	}
+	if begin >= end {
+		return nil, fmt.Errorf("dedup_trigger start time %d is greater than end time %d", begin, end)
+	}
+	if n, ok := d.Value(w.nowField, ""); ok {
+		now, err = cast.ToInt64(n, cast.CONVERT_SAMEKIND)
+		if err != nil {
+			return nil, fmt.Errorf("dedup_trigger now time %s is not int64", n)
+		}
+	} else {
+		return nil, fmt.Errorf("dedup_trigger %s is missing", w.nowField)
+	}
+	return &TriggerRequest{begin, end, now, w.expire, d}, nil
+}
+
+func doTrigger(ctx api.StreamContext, start int64, end int64, now int64, exp int64) ([]map[string]any, error) {
+	var result []map[string]any
+	leftmost := now - exp
+	if end < leftmost {
+		return result, nil
+	}
+	if start < leftmost {
+		start = leftmost
+	}
+
+	// histogram state, the timeslots which have been taken [{start, end}, {start, end}]
+	st, err := ctx.GetState("histogram")
+	if err != nil {
+		ctx.GetLogger().Errorf("dedup_trigger get histogram state error: %s", err)
+		return nil, fmt.Errorf("dedup_trigger get histogram state error: %s", err)
+	}
+	if st == nil {
+		st = [][]int64{}
+	}
+	hg := st.([][]int64)
+	if len(hg) > 0 {
+		// clean up the expired timeslots
+		i := 0
+		for ; i < len(hg); i++ {
+			if hg[i][1] >= leftmost {
+				break
+			}
+		}
+		hg = hg[i:]
+		// Find the timeslots which have been taken
+		// Default to the rightest slot
+		leftFound := 2 * len(hg)
+		rightFound := 2 * len(hg)
+		for i, v := range hg {
+			if leftFound == 2*len(hg) {
+				if start < v[0] {
+					leftFound = 2 * i
+				} else if start < v[1] {
+					leftFound = 2*i + 1
+				}
+			}
+			if leftFound < 2*len(hg) {
+				if end <= v[0] {
+					rightFound = 2 * i
+					break
+				} else if end <= v[1] {
+					rightFound = 2*i + 1
+					break
+				}
+			}
+		}
+		// calculate timeslots and update histogram for each cases
+		if leftFound == rightFound {
+			// In a continuous empty slot
+			if leftFound%2 == 0 {
+				index := leftFound / 2
+				result = append(result, map[string]any{"start_key": strconv.FormatInt(start, 10), "end_key": strconv.FormatInt(end, 10)})
+				hg = append(hg[:index], append([][]int64{{start, end}}, hg[index:]...)...)
+			} else { // do nothing
+				ctx.GetLogger().Infof("dedup_trigger start time %d and end time %d are already sent before", start, end)
+			}
+		} else {
+			if leftFound%2 == 0 {
+				if rightFound > 0 && rightFound%2 == 0 { // left empty slot, right empty slot
+					// left slot + multiple middle empty slots + right slot
+					lhg := hg[leftFound/2]
+					rhg := hg[rightFound/2-1]
+					result = append(result, map[string]any{"start_key": strconv.FormatInt(start, 10), "end_key": strconv.FormatInt(lhg[0], 10)})
+					for i := leftFound / 2; i < rightFound/2-1; i++ {
+						result = append(result, map[string]any{"start_key": strconv.FormatInt(hg[i][1], 10), "end_key": strconv.FormatInt(hg[i+1][0], 10)})
+					}
+					result = append(result, map[string]any{"start_key": strconv.FormatInt(rhg[1], 10), "end_key": strconv.FormatInt(end, 10)})
+					hg = append(hg[:leftFound/2], append([][]int64{{start, end}}, hg[rightFound/2:]...)...)
+				} else { // left empty slot, right not empty slot
+					// left slot + multiple middle empty slots
+					lhg := hg[leftFound/2]
+					rhg := hg[(rightFound-1)/2]
+					result = append(result, map[string]any{"start_key": strconv.FormatInt(start, 10), "end_key": strconv.FormatInt(lhg[0], 10)})
+					for i := leftFound / 2; i < (rightFound-1)/2; i++ {
+						result = append(result, map[string]any{"start_key": strconv.FormatInt(hg[i][1], 10), "end_key": strconv.FormatInt(hg[i+1][0], 10)})
+					}
+					hg = append(hg[:leftFound/2], append([][]int64{{start, rhg[1]}}, hg[(rightFound+1)/2:]...)...)
+				}
+			} else {
+				if rightFound > 0 && rightFound%2 == 0 { // left not empty slot, right empty slot
+					// multiple middle empty slots + right slot
+					lhg := hg[leftFound/2]
+					rhg := hg[rightFound/2-1]
+					for i := leftFound / 2; i < rightFound/2-1; i++ {
+						result = append(result, map[string]any{"start_key": strconv.FormatInt(hg[i][1], 10), "end_key": strconv.FormatInt(hg[i+1][0], 10)})
+					}
+					result = append(result, map[string]any{"start_key": strconv.FormatInt(rhg[1], 10), "end_key": strconv.FormatInt(end, 10)})
+					hg = append(hg[:leftFound/2], append([][]int64{{lhg[0], end}}, hg[rightFound/2:]...)...)
+				} else { // left not empty slot, right not empty slot
+					lhg := hg[leftFound/2]
+					rhg := hg[(rightFound-1)/2]
+					// multiple middle empty slots
+					for i := leftFound / 2; i < (rightFound-1)/2; i++ {
+						result = append(result, map[string]any{"start_key": strconv.FormatInt(hg[i][1], 10), "end_key": strconv.FormatInt(hg[i+1][0], 10)})
+					}
+					hg = append(hg[:leftFound/2], append([][]int64{{lhg[0], rhg[1]}}, hg[(rightFound+1)/2:]...)...)
+				}
+			}
+		}
+	} else {
+		result = append(result, map[string]any{"start_key": strconv.FormatInt(start, 10), "end_key": strconv.FormatInt(end, 10)})
+		hg = append(hg, []int64{start, end})
+	}
+	_ = ctx.PutState("histogram", hg)
+	return result, nil
+}
+
+type TriggerRequest struct {
+	start int64
+	end   int64
+	now   int64
+	exp   int64
+	tuple xsql.Row
+}
+
+type PriorityQueue []*TriggerRequest
+
+// Push adds an item to the priority queue
+func (pq *PriorityQueue) Push(x *TriggerRequest) {
+	for i, r := range *pq {
+		if r.end > x.end {
+			*pq = append((*pq)[:i], append(PriorityQueue{x}, (*pq)[i:]...)...)
+			return
+		}
+	}
+	*pq = append(*pq, x)
+}
+
+// Pop removes and returns the item with the highest priority from the priority queue
+func (pq *PriorityQueue) Pop() *TriggerRequest {
+	old := *pq
+	item := old[0]
+	*pq = old[1:]
+	return item
+}
+
+func (pq *PriorityQueue) Peek() *TriggerRequest {
+	return (*pq)[0]
+}
diff --git a/internal/topo/node/encode_op.go b/internal/topo/node/encode_op.go
index 440b01e8..ef4bb2a7 100644
--- a/internal/topo/node/encode_op.go
+++ b/internal/topo/node/encode_op.go
@@ -54,14 +54,14 @@ func (o *EncodeOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *EncodeOp) Worker(_ api.Logger, item any) []any {
+func (o *EncodeOp) Worker(ctx api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
 	case []byte:
 		return []any{d}
 	default:
-		r, err := o.converter.Encode(item)
+		r, err := o.converter.Encode(ctx, item)
 		if err != nil {
 			return []any{err}
 		} else {
diff --git a/internal/topo/node/encrypt_op.go b/internal/topo/node/encrypt_op.go
index 87306f0e..2eedb3dc 100644
--- a/internal/topo/node/encrypt_op.go
+++ b/internal/topo/node/encrypt_op.go
@@ -53,7 +53,7 @@ func (o *EncryptNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func (o *EncryptNode) Worker(_ api.Logger, item any) []any {
+func (o *EncryptNode) Worker(_ api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
diff --git a/internal/topo/node/transform_op.go b/internal/topo/node/transform_op.go
index 3d6387ef..76b9d46a 100644
--- a/internal/topo/node/transform_op.go
+++ b/internal/topo/node/transform_op.go
@@ -74,16 +74,16 @@ func (t *TransformOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 }
 
 // Worker do not need to process error and control messages
-func (t *TransformOp) Worker(logger api.Logger, item any) []any {
+func (t *TransformOp) Worker(ctx api.StreamContext, item any) []any {
 	t.statManager.ProcessTimeStart()
 	defer t.statManager.ProcessTimeEnd()
 	if ic, ok := item.(xsql.Collection); ok && t.omitIfEmpty && ic.Len() == 0 {
-		logger.Debugf("receive empty collection, dropped")
+		ctx.GetLogger().Debugf("receive empty collection, dropped")
 		return nil
 	}
 	outs := itemToMap(item)
 	if t.omitIfEmpty && (item == nil || len(outs) == 0) {
-		logger.Debugf("receive empty result %v in sink, dropped", outs)
+		ctx.GetLogger().Debugf("receive empty result %v in sink, dropped", outs)
 		return nil
 	}
 	var result []any
diff --git a/internal/topo/planner/dedup_trigger_plan.go b/internal/topo/planner/dedup_trigger_plan.go
new file mode 100644
index 00000000..757a4c5e
--- /dev/null
+++ b/internal/topo/planner/dedup_trigger_plan.go
@@ -0,0 +1,37 @@
+// 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 planner
+
+import "github.com/lf-edge/ekuiper/v2/pkg/ast"
+
+type DedupTriggerPlan struct {
+	baseLogicalPlan
+	aliasName  string
+	startField *ast.FieldRef
+	endField   *ast.FieldRef
+	nowField   *ast.FieldRef
+	expire     int64
+}
+
+func (p DedupTriggerPlan) Init() *DedupTriggerPlan {
+	p.baseLogicalPlan.self = &p
+	return &p
+}
+
+func (p *DedupTriggerPlan) PruneColumns(fields []ast.Expr) error {
+	return p.baseLogicalPlan.PruneColumns(append(fields, p.startField, p.endField, p.nowField))
+}
+
+var _ LogicalPlan = &DedupTriggerPlan{}
diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
index 4c3495a3..af9d158c 100644
--- a/internal/topo/planner/planner.go
+++ b/internal/topo/planner/planner.go
@@ -247,6 +247,8 @@ func buildOps(lp LogicalPlan, tp *topo.Topo, options *def.RuleOption, sources ma
 		if err != nil {
 			return nil, 0, err
 		}
+	case *DedupTriggerPlan:
+		op = node.NewDedupTriggerNode(fmt.Sprintf("%d_dedup_trigger", newIndex), options, t.aliasName, t.startField.Name, t.endField.Name, t.nowField.Name, t.expire)
 	case *LookupPlan:
 		op, err = node.NewLookupNode(t.joinExpr.Name, t.fields, t.keys, t.joinExpr.JoinType, t.valvars, t.options, options)
 	case *JoinAlignPlan:
@@ -487,6 +489,47 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *def.RuleOption, store kv.
 	}
 	srfMapping := extractSRFMapping(stmt)
 	if stmt.Fields != nil {
+		// extract dedup trigger op
+		fields := make([]ast.Field, 0, len(stmt.Fields))
+		for _, field := range stmt.Fields {
+			if field.Expr != nil {
+				var (
+					exp  *ast.Expr
+					name string
+					fc   *ast.Call
+				)
+				if f, ok := field.Expr.(*ast.FieldRef); ok {
+					if f.AliasRef != nil && f.AliasRef.Expression != nil {
+						if wf, ok := f.AliasRef.Expression.(*ast.Call); ok && wf.FuncType == ast.FuncTypeTrigger {
+							exp = &f.AliasRef.Expression
+							name = field.AName
+							fc = wf
+						}
+					}
+				} else if f, ok := field.Expr.(*ast.Call); ok && f.FuncType == ast.FuncTypeTrigger {
+					name = field.Name
+					exp = &field.Expr
+					fc = f
+				}
+				if exp != nil {
+					p = DedupTriggerPlan{
+						aliasName:  name,
+						startField: fc.Args[0].(*ast.FieldRef),
+						endField:   fc.Args[1].(*ast.FieldRef),
+						nowField:   fc.Args[2].(*ast.FieldRef),
+						expire:     fc.Args[3].(*ast.IntegerLiteral).Val,
+					}.Init()
+					p.SetChildren(children)
+					children = []LogicalPlan{p}
+
+					*exp = &ast.FieldRef{
+						StreamName: ast.DefaultStream,
+						Name:       name,
+					}
+				}
+			}
+			fields = append(fields, field)
+		}
 		enableLimit := false
 		limitCount := 0
 		if stmt.Limit != nil && len(srfMapping) == 0 {
@@ -495,7 +538,7 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *def.RuleOption, store kv.
 		}
 		p = ProjectPlan{
 			windowFuncNames: windowFuncsNames,
-			fields:          stmt.Fields,
+			fields:          fields,
 			isAggregate:     xsql.WithAggFields(stmt),
 			sendMeta:        opt.SendMetaToSink,
 			enableLimit:     enableLimit,
diff --git a/internal/topo/planner/planner_sink.go b/internal/topo/planner/planner_sink.go
index 029d07b3..53ad29e0 100644
--- a/internal/topo/planner/planner_sink.go
+++ b/internal/topo/planner/planner_sink.go
@@ -23,6 +23,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/topo"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/node"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
 )
 
 // SinkPlanner is the planner for sink node. It transforms logical sink plan to multiple physical nodes.
@@ -103,8 +104,8 @@ func splitSink(tp *topo.Topo, inputs []node.Emitter, s api.Sink, sinkName string
 		index++
 		tp.AddOperator(newInputs, encodeOp)
 		newInputs = []node.Emitter{encodeOp}
-
-		if sc.Compression != "" {
+		_, isStreamWriter := s.(model.StreamWriter)
+		if !isStreamWriter && sc.Compression != "" {
 			compressOp, err := node.NewCompressOp(fmt.Sprintf("%s_%d_compress", sinkName, index), options, sc.Compression)
 			if err != nil {
 				return nil, err
@@ -114,7 +115,7 @@ func splitSink(tp *topo.Topo, inputs []node.Emitter, s api.Sink, sinkName string
 			newInputs = []node.Emitter{compressOp}
 		}
 
-		if sc.Encryption != "" {
+		if !isStreamWriter && sc.Encryption != "" {
 			encryptOp, err := node.NewEncryptOp(fmt.Sprintf("%s_%d_encrypt", sinkName, index), options, sc.Encryption)
 			if err != nil {
 				return nil, err
diff --git a/internal/topo/transform/template.go b/internal/topo/transform/template.go
index 2f941139..ffe37ff0 100644
--- a/internal/topo/transform/template.go
+++ b/internal/topo/transform/template.go
@@ -15,107 +15,13 @@
 package transform
 
 import (
-	"bytes"
 	"encoding/json"
 	"fmt"
 	"text/template"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/converter"
-	"github.com/lf-edge/ekuiper/v2/internal/converter/delimited"
-	"github.com/lf-edge/ekuiper/v2/pkg/ast"
-	"github.com/lf-edge/ekuiper/v2/pkg/message"
 )
 
-// TransFunc is the function to transform data
-type TransFunc func(interface{}) ([]byte, bool, error)
-
-func GenTransform(dt string, format string, schemaId string, delimiter string, dataField string, fields []string) (TransFunc, error) {
-	var (
-		tp  *template.Template = nil
-		c   message.Converter
-		err error
-	)
-	switch format {
-	case message.FormatProtobuf, message.FormatCustom:
-		c, err = converter.GetOrCreateConverter(&ast.Options{FORMAT: format, SCHEMAID: schemaId})
-		if err != nil {
-			return nil, err
-		}
-	case message.FormatDelimited:
-		c, err = converter.GetOrCreateConverter(&ast.Options{FORMAT: format, DELIMITER: delimiter})
-		if err != nil {
-			return nil, err
-		}
-		c.(*delimited.Converter).SetColumns(fields)
-	case message.FormatJson:
-		c, err = converter.GetOrCreateConverter(&ast.Options{FORMAT: format})
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	if dt != "" {
-		temp, err := template.New("sink").Funcs(conf.FuncMap).Parse(dt)
-		if err != nil {
-			return nil, err
-		}
-		tp = temp
-	}
-	return func(d interface{}) ([]byte, bool, error) {
-		var (
-			bs          []byte
-			transformed bool
-			selected    bool
-			m           interface{}
-			e           error
-		)
-		if tp != nil {
-			var output bytes.Buffer
-			err := tp.Execute(&output, d)
-			if err != nil {
-				return nil, false, fmt.Errorf("fail to encode data %v with dataTemplate for error %v", d, err)
-			}
-			bs = output.Bytes()
-			transformed = true
-		}
-
-		if transformed {
-			m, selected, e = TransItem(bs, dataField, fields)
-		} else {
-			m, selected, e = TransItem(d, dataField, fields)
-		}
-		if e != nil {
-			return nil, false, fmt.Errorf("fail to TransItem data %v for error %v", d, e)
-		}
-		if selected {
-			d = m
-		}
-
-		switch format {
-		case message.FormatJson:
-			if transformed && !selected {
-				return bs, true, nil
-			}
-			outBytes, err := c.Encode(d)
-			return outBytes, transformed || selected, err
-		case message.FormatProtobuf, message.FormatCustom, message.FormatDelimited:
-			if transformed && !selected {
-				m := make(map[string]interface{})
-				err := json.Unmarshal(bs, &m)
-				if err != nil {
-					return nil, false, fmt.Errorf("fail to decode data %s after applying dataTemplate for error %v", string(bs), err)
-				}
-				d = m
-			}
-			outBytes, err := c.Encode(d)
-			return outBytes, transformed || selected, err
-		default: // should not happen
-			return nil, false, fmt.Errorf("unsupported format %v", format)
-		}
-	}, nil
-}
-
 func GenTp(dt string) (*template.Template, error) {
 	return template.New("sink").Funcs(conf.FuncMap).Parse(dt)
 }
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index 41ede65c..e971a062 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -25,6 +25,7 @@ import (
 	"github.com/golang-collections/collections/stack"
 
 	"github.com/lf-edge/ekuiper/v2/internal/binder/function"
+	_ "github.com/lf-edge/ekuiper/v2/internal/converter"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
@@ -869,6 +870,7 @@ var WindowFuncs = map[string]struct{}{
 	"sessionwindow":  {},
 	"slidingwindow":  {},
 	"countwindow":    {},
+	"dedup_trigger":  {},
 }
 
 func convFuncName(n string) (string, bool) {
@@ -918,8 +920,27 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 		}
 	}
 	if wt, err := validateWindows(name, args); wt == ast.NOT_WINDOW {
-		if valErr := validateFuncs(name, args); valErr != nil {
-			return nil, valErr
+		switch name {
+		case "dedup_trigger":
+			ft = ast.FuncTypeTrigger
+			if len(args) != 4 {
+				return nil, fmt.Errorf("dedup_trigger function should have 4 arguments")
+			}
+			for i, arg := range args {
+				if i == 3 {
+					break
+				}
+				if _, ok := arg.(*ast.FieldRef); !ok {
+					return nil, fmt.Errorf("dedup_trigger function should have fieldRef as the %d argument", i+1)
+				}
+			}
+			if _, ok := args[3].(*ast.IntegerLiteral); !ok {
+				return nil, fmt.Errorf("dedup_trigger function should have integer as the fourth argument")
+			}
+		default:
+			if valErr := validateFuncs(name, args); valErr != nil {
+				return nil, valErr
+			}
 		}
 		// Add context for some aggregate func
 		if name == "deduplicate" {
diff --git a/pkg/ast/expr.go b/pkg/ast/expr.go
index 0dfcef3e..6f2c8b20 100644
--- a/pkg/ast/expr.go
+++ b/pkg/ast/expr.go
@@ -210,6 +210,7 @@ const (
 	FuncTypeCols
 	FuncTypeSrf
 	FuncTypeWindow
+	FuncTypeTrigger
 )
 
 type Call struct {
diff --git a/pkg/message/artifacts.go b/pkg/message/artifacts.go
index 3708a42c..a8b90e54 100644
--- a/pkg/message/artifacts.go
+++ b/pkg/message/artifacts.go
@@ -15,6 +15,7 @@
 package message
 
 import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 )
 
@@ -31,8 +32,8 @@ const (
 
 // Converter converts bytes & map or []map according to the schema
 type Converter interface {
-	Encode(d interface{}) ([]byte, error)
-	Decode(b []byte) (interface{}, error)
+	Encode(ctx api.StreamContext, d any) ([]byte, error)
+	Decode(ctx api.StreamContext, b []byte) (any, error)
 }
 
 // ConverterProvider The format, schema information are passed in by stream options
diff --git a/pkg/mock/context/mock.go b/pkg/mock/context/mock.go
index 3d6076e5..dcd63c10 100644
--- a/pkg/mock/context/mock.go
+++ b/pkg/mock/context/mock.go
@@ -20,14 +20,11 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"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/topo/transform"
 )
 
 func NewMockContext(ruleId string, opId string) api.StreamContext {
 	contextLogger := conf.Log.WithField("rule", ruleId)
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	tf, _ := transform.GenTransform("", "json", "", "", "", []string{})
-	ctx = context.WithValue(ctx, context.TransKey, tf)
 	tempStore, _ := state.CreateStore(ruleId, def.AtMostOnce)
 	return ctx.WithMeta(ruleId, opId, tempStore)
 }
diff --git a/pkg/timex/time.go b/pkg/timex/time.go
index 58fbf14b..0bddadd7 100644
--- a/pkg/timex/time.go
+++ b/pkg/timex/time.go
@@ -69,3 +69,17 @@ func GetNowInMilli() int64 {
 func GetNow() time.Time {
 	return Clock.Now()
 }
+
+// Mock time, only use in test
+
+func Set(t int64) {
+	if IsTesting {
+		Clock.(*clock.Mock).Set(time.UnixMilli(t))
+	}
+}
+
+func Add(d time.Duration) {
+	if IsTesting {
+		Clock.(*clock.Mock).Add(d)
+	}
+}


Test patch:
diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index 688719ac..495cb222 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -45,12 +45,6 @@ jobs:
         mkdir -p log
         mkdir -p plugins/sinks
         sed -i -r "s/debug: .*/debug: true/1" etc/kuiper.yaml
-        go build --buildmode=plugin -o data/test/helloworld.so internal/converter/protobuf/test/*.go
-        go build --buildmode=plugin -o data/myFormat.so internal/converter/custom/test/*.go
-        mkdir -p data/custom
-        cp data/myFormat.so data/custom/myFormat.so
-        mkdir -p data/schema
-        cp data/myFormat.so data/schema/myFormat.so
     - name: Run test case
       run: |
         make failpoint-enable
diff --git a/internal/converter/binary/converter_test.go b/internal/converter/binary/converter_test.go
index 2f4ffda4..ef544ea5 100644
--- a/internal/converter/binary/converter_test.go
+++ b/internal/converter/binary/converter_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestMessageDecode(t *testing.T) {
@@ -42,8 +43,9 @@ func TestMessageDecode(t *testing.T) {
 		},
 	}
 	conv, _ := GetConverter()
+	ctx := mockContext.NewMockContext("test", "op1")
 	for i, tt := range tests {
-		result, err := conv.Decode(tt.payload)
+		result, err := conv.Decode(ctx, tt.payload)
 		if err != nil {
 			t.Errorf("%d decode error: %v", i, err)
 		}
@@ -54,7 +56,8 @@ func TestMessageDecode(t *testing.T) {
 }
 
 func TestError(t *testing.T) {
-	_, err := converter.Encode(nil)
+	ctx := mockContext.NewMockContext("test", "op1")
+	_, err := converter.Encode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok := err.(errorx.ErrorWithCode)
 	require.True(t, ok)
diff --git a/internal/converter/custom/converter_test.go b/internal/converter/custom/converter_test.go
deleted file mode 100644
index 01cc3bbf..00000000
--- a/internal/converter/custom/converter_test.go
+++ /dev/null
@@ -1,142 +0,0 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package custom
-
-import (
-	"fmt"
-	"os"
-	"path/filepath"
-	"reflect"
-	"testing"
-
-	"github.com/gdexlab/go-render/render"
-
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/schema"
-	"github.com/lf-edge/ekuiper/v2/internal/testx"
-)
-
-func init() {
-	testx.InitEnv("custom")
-}
-
-func TestCustomConverter(t *testing.T) {
-	dataDir, err := conf.GetDataLoc()
-	if err != nil {
-		t.Fatal(err)
-	}
-	etcDir := filepath.Join(dataDir, "schemas", "custom")
-	err = os.MkdirAll(etcDir, os.ModePerm)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(etcDir)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}()
-	// build the so file into data/test prior to running the test
-	// Copy the helloworld.so
-	bytesRead, err := os.ReadFile(filepath.Join(dataDir, "myFormat.so"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = os.WriteFile(filepath.Join(etcDir, "myFormat.so"), bytesRead, 0o755)
-	if err != nil {
-		t.Fatal(err)
-	}
-	schema.InitRegistry()
-	testEncode(t)
-	testDecode(t)
-}
-
-func testEncode(t *testing.T) {
-	c, err := LoadConverter("myFormat", "Sample", "")
-	if err != nil {
-		t.Fatal(err)
-	}
-	tests := []struct {
-		m map[string]interface{}
-		r []byte
-		e string
-	}{
-		{
-			m: map[string]interface{}{
-				"id":   12,
-				"name": "test",
-			},
-			r: []byte(`{"id":12,"name":"test","age":0,"hobbies":{"indoor":null,"outdoor":null}}`),
-		}, {
-			m: map[string]interface{}{
-				"id":   7,
-				"name": "John Doe",
-				"age":  22,
-				"hobbies": map[string]interface{}{
-					"indoor": []string{
-						"Chess",
-					},
-					"outdoor": []string{
-						"Basketball",
-					},
-				},
-			},
-			r: []byte(`{"id":7,"name":"John Doe","age":22,"hobbies":{"indoor":["Chess"],"outdoor":["Basketball"]}}`),
-		},
-	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	for i, tt := range tests {
-		a, err := c.Encode(tt.m)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%s\n\ngot=%s\n\n", i, tt.r, a)
-		}
-	}
-}
-
-func testDecode(t *testing.T) {
-	c, err := LoadConverter("myFormat", "Sample", "")
-	if err != nil {
-		t.Fatal(err)
-	}
-	tests := []struct {
-		m map[string]interface{}
-		r []byte
-		e string
-	}{
-		{
-			m: map[string]interface{}{
-				"name": "test",
-				"age":  int64(0),
-				"id":   int64(0),
-				"hobbies": map[string]interface{}{
-					"indoor":  []string(nil),
-					"outdoor": []string(nil),
-				},
-			},
-			r: []byte(`{"name":"test"}`),
-		},
-	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	for i, tt := range tests {
-		a, err := c.Decode(tt.r)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.m, a) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%v\n\ngot=%v\n\n", i, render.AsCode(tt.m), render.AsCode(a))
-		}
-	}
-}
diff --git a/internal/converter/custom/test/Makefile b/internal/converter/custom/test/Makefile
deleted file mode 100644
index 22f23892..00000000
--- a/internal/converter/custom/test/Makefile
+++ /dev/null
@@ -1,5 +0,0 @@
-.PHONY: build
-build:
-	@echo "Start build"
-	CGO_ENABLED=1 go build -trimpath --buildmode=plugin
-	@echo "Build successfully"
\ No newline at end of file
diff --git a/internal/converter/custom/test/go.mod b/internal/converter/custom/test/go.mod
deleted file mode 100644
index cf3a2978..00000000
--- a/internal/converter/custom/test/go.mod
+++ /dev/null
@@ -1,5 +0,0 @@
-module test
-
-go 1.22
-
-require github.com/mitchellh/mapstructure v1.5.0
diff --git a/internal/converter/custom/test/go.sum b/internal/converter/custom/test/go.sum
deleted file mode 100644
index 59f4b8e6..00000000
--- a/internal/converter/custom/test/go.sum
+++ /dev/null
@@ -1,2 +0,0 @@
-github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
-github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
diff --git a/internal/converter/custom/test/myformat.go b/internal/converter/custom/test/myformat.go
deleted file mode 100644
index 6c33ef78..00000000
--- a/internal/converter/custom/test/myformat.go
+++ /dev/null
@@ -1,117 +0,0 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package main
-
-import (
-	"encoding/json"
-	"fmt"
-
-	"github.com/mitchellh/mapstructure"
-)
-
-type Hobbies struct {
-	Indoor  []string `json:"indoor"`
-	Outdoor []string `json:"outdoor"`
-}
-
-type Sample struct {
-	Id      int64   `json:"id"`
-	Name    string  `json:"name"`
-	Age     int64   `json:"age"`
-	Hobbies Hobbies `json:"hobbies"`
-}
-
-func (x *Sample) GetSchemaJson() string {
-	// return a static schema
-	return `{
-		"id": {
-			"type": "bigint"
-	},
-		"name": {
-			"type": "string"
-	},
-		"age": {
-			"type": "bigint"
-	},
-		"hobbies": {
-			"type": "struct",
-			"properties": {
-			"indoor": {
-				"type": "array",
-					"items": {
-						"type": "string"
-				}
-			},
-			"outdoor": {
-				"type": "array",
-					"items": {
-						"type": "string"
-				}
-			}
-		}
-	}
-	}`
-}
-
-func (x *Sample) Encode(d interface{}) ([]byte, error) {
-	switch r := d.(type) {
-	case map[string]interface{}:
-		result := &Sample{}
-		err := MapToStructStrict(r, result)
-		if err != nil {
-			return nil, err
-		}
-		return json.Marshal(result)
-	default:
-		return nil, fmt.Errorf("unsupported type %v, must be a map", d)
-	}
-}
-
-func (x *Sample) Decode(b []byte) (interface{}, error) {
-	result := &Sample{}
-	// check error
-	err := json.Unmarshal(b, &result)
-	if err != nil {
-		return nil, err
-	}
-	// convert struct to map
-	hobbyMap := make(map[string]interface{}, 2)
-	hobbyMap["indoor"] = result.Hobbies.Indoor
-	hobbyMap["outdoor"] = result.Hobbies.Outdoor
-	resultMap := make(map[string]interface{}, 4)
-	resultMap["id"] = result.Id
-	resultMap["name"] = result.Name
-	resultMap["age"] = result.Age
-	resultMap["hobbies"] = hobbyMap
-	return resultMap, err
-}
-
-func MapToStructStrict(input, output interface{}) error {
-	config := &mapstructure.DecoderConfig{
-		ErrorUnused: true,
-		TagName:     "json",
-		Result:      output,
-	}
-	decoder, err := mapstructure.NewDecoder(config)
-	if err != nil {
-		return err
-	}
-
-	return decoder.Decode(input)
-}
-
-func GetSample() interface{} {
-	return &Sample{}
-}
diff --git a/internal/converter/delimited/converter_test.go b/internal/converter/delimited/converter_test.go
index 5c93740f..bd3f37d6 100644
--- a/internal/converter/delimited/converter_test.go
+++ b/internal/converter/delimited/converter_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestEncode(t *testing.T) {
@@ -55,12 +56,13 @@ func TestEncode(t *testing.T) {
 		},
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	ctx := mockContext.NewMockContext("test", "op1")
 	for i, tt := range tests {
 		c, err := NewConverter(":")
 		if err != nil {
 			t.Fatal(err)
 		}
-		a, err := c.Encode(tt.m)
+		a, err := c.Encode(ctx, tt.m)
 		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
 			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
 		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
@@ -102,14 +104,15 @@ func TestDecode(t *testing.T) {
 		},
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	ctx := mockContext.NewMockContext("test", "op1")
 	for i, tt := range tests {
-		a, err := c.Decode(tt.r)
+		a, err := c.Decode(ctx, tt.r)
 		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
 			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
 		} else if tt.e == "" && !reflect.DeepEqual(tt.m, a) {
 			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%v\n\ngot=%v\n\n", i, tt.m, a)
 		}
-		b, err := ch.Decode(tt.r)
+		b, err := ch.Decode(ctx, tt.r)
 		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
 			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
 		} else if tt.e == "" && !reflect.DeepEqual(tt.nm, b) {
@@ -121,7 +124,8 @@ func TestDecode(t *testing.T) {
 func TestError(t *testing.T) {
 	converter, err := NewConverter(",")
 	require.NoError(t, err)
-	_, err = converter.Encode(nil)
+	ctx := mockContext.NewMockContext("test", "op1")
+	_, err = converter.Encode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok := err.(errorx.ErrorWithCode)
 	require.True(t, ok)
diff --git a/internal/converter/ext_mock_test.go b/internal/converter/ext_mock_test.go
index 1e733641..666554fb 100644
--- a/internal/converter/ext_mock_test.go
+++ b/internal/converter/ext_mock_test.go
@@ -20,15 +20,18 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestEncodeWithMockConverter(t *testing.T) {
 	mockConverter := MockConverter{}
+	ctx := mockContext.NewMockContext("test", "op1")
 	data := map[string]interface{}{
 		"temperature": 23.4,
 		"humidity":    76,
 	}
-	encodedData, err := mockConverter.Encode(data)
+	encodedData, err := mockConverter.Encode(ctx, data)
 
 	assert.Nil(t, err)
 	assert.Contains(t, string(encodedData), `"temperature":23.4`)
@@ -37,8 +40,9 @@ func TestEncodeWithMockConverter(t *testing.T) {
 
 func TestDecodeWithMockConverter(t *testing.T) {
 	mockConverter := MockConverter{}
+	ctx := mockContext.NewMockContext("test", "op1")
 	data := []byte(`{"temperature":23.4,"humidity":76,"ts":1633027200000}`)
-	decodedData, err := mockConverter.Decode(data)
+	decodedData, err := mockConverter.Decode(ctx, data)
 
 	assert.Nil(t, err)
 	assert.Equal(t, 23.4, decodedData.(map[string]interface{})["temperature"])
diff --git a/internal/converter/json/convert_bench_test.go b/internal/converter/json/convert_bench_test.go
index 3989b310..732a1171 100644
--- a/internal/converter/json/convert_bench_test.go
+++ b/internal/converter/json/convert_bench_test.go
@@ -19,6 +19,7 @@ import (
 	"testing"
 
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func BenchmarkSimpleTuples(b *testing.B) {
@@ -100,6 +101,7 @@ func BenchmarkComplexTuplesWithSchema(b *testing.B) {
 }
 
 func benchmarkByFiles(filePath string, b *testing.B, schema map[string]*ast.JsonStreamField) {
+	ctx := mockContext.NewMockContext("test", "test")
 	payload, err := os.ReadFile(filePath)
 	if err != nil {
 		b.Fatalf(err.Error())
@@ -108,12 +110,12 @@ func benchmarkByFiles(filePath string, b *testing.B, schema map[string]*ast.Json
 		f := NewFastJsonConverter("", "", schema, false, false)
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			f.Decode(payload)
+			f.Decode(ctx, payload)
 		}
 	} else {
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			converter.Decode(payload)
+			converter.Decode(ctx, payload)
 		}
 	}
 }
diff --git a/internal/converter/json/converter_test.go b/internal/converter/json/converter_test.go
index 5996ff97..62014b37 100644
--- a/internal/converter/json/converter_test.go
+++ b/internal/converter/json/converter_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/converter/merge"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestMessageDecode(t *testing.T) {
@@ -65,8 +66,9 @@ func TestMessageDecode(t *testing.T) {
 		},
 	}
 	conv, _ := GetConverter()
+	ctx := mockContext.NewMockContext("test", "op1")
 	for i, tt := range tests {
-		result, err := conv.Decode(tt.payload)
+		result, err := conv.Decode(ctx, tt.payload)
 		if err != nil {
 			t.Errorf("%d decode error: %v", i, err)
 		}
@@ -214,9 +216,10 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 			},
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		v, err := f.Decode(tc.payload)
+		v, err := f.Decode(ctx, tc.payload)
 		require.NoError(t, err)
 		require.Equal(t, v, tc.require)
 	}
@@ -227,7 +230,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 			tc.require,
 		}
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		v, err := f.Decode(arrayPayload)
+		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
 	}
@@ -384,10 +387,10 @@ func TestFastJsonConverterWithSchemaError(t *testing.T) {
 			err: fmt.Errorf("array has wrong type:true, expect:datetime"),
 		},
 	}
-
+	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		_, err := f.Decode(tc.payload)
+		_, err := f.Decode(ctx, tc.payload)
 		require.Error(t, err)
 		require.Equal(t, err.Error(), tc.err.Error())
 	}
@@ -396,8 +399,9 @@ func TestFastJsonConverterWithSchemaError(t *testing.T) {
 func TestFastJsonEncode(t *testing.T) {
 	a := make(map[string]int)
 	a["a"] = 1
+	ctx := mockContext.NewMockContext("test", "op1")
 	f := NewFastJsonConverter("", "", nil, false, false)
-	v, err := f.Encode(a)
+	v, err := f.Encode(ctx, a)
 	require.NoError(t, err)
 	require.Equal(t, v, []byte(`{"a":1}`))
 }
@@ -428,8 +432,9 @@ func TestArrayWithArray(t *testing.T) {
 			},
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	f := NewFastJsonConverter("", "", schema, false, false)
-	v, err := f.Decode(payload)
+	v, err := f.Decode(ctx, payload)
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
 		"a": []interface{}{
@@ -603,13 +608,14 @@ func TestTypeNull(t *testing.T) {
 			},
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
 		arrayPayload := []byte(fmt.Sprintf("[%s]", string(tc.payload)))
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		v, err := f.Decode(arrayPayload)
+		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
 	}
@@ -619,7 +625,7 @@ func TestTypeNull(t *testing.T) {
 			tc.require,
 		}
 		f := NewFastJsonConverter("", "", tc.schema, false, false)
-		v, err := f.Decode(arrayPayload)
+		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
 	}
@@ -634,8 +640,9 @@ func TestConvertBytea(t *testing.T) {
 			Type: "bytea",
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	f := NewFastJsonConverter("", "", schema, false, false)
-	v, err := f.Decode([]byte(payload))
+	v, err := f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
 		"a": []byte(origin),
@@ -651,7 +658,7 @@ func TestConvertBytea(t *testing.T) {
 		},
 	}
 	f = NewFastJsonConverter("", "", schema, false, false)
-	v, err = f.Decode([]byte(payload))
+	v, err = f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
 		"a": []interface{}{[]byte(origin)},
@@ -846,12 +853,13 @@ func TestMergeWildcardSchema(t *testing.T) {
 		"b": float64(2),
 		"c": float64(3),
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	bs, _ := json.Marshal(data)
-	d, err := f.Decode(bs)
+	d, err := f.Decode(ctx, bs)
 	require.NoError(t, err)
 	require.Equal(t, data, d)
 	require.NoError(t, f.DetachSchema("2"))
-	d, err = f.Decode(bs)
+	d, err = f.Decode(ctx, bs)
 	require.NoError(t, err)
 	require.Equal(t, map[string]interface{}{
 		"a": int64(1),
@@ -901,23 +909,25 @@ func TestSchemaless(t *testing.T) {
 			},
 		},
 	}
+	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
 		bs, _ := json.Marshal(tc.data)
-		v, err := f.Decode(bs)
+		v, err := f.Decode(ctx, bs)
 		require.NoError(t, err)
 		require.Equal(t, tc.expect, v)
 	}
 }
 
 func TestJsonError(t *testing.T) {
-	_, err := converter.Decode(nil)
+	ctx := mockContext.NewMockContext("test", "op1")
+	_, err := converter.Decode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok := err.(errorx.ErrorWithCode)
 	require.True(t, ok)
 	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
 	// fastjson
 	c := NewFastJsonConverter("", "", nil, false, true)
-	_, err = c.Decode(nil)
+	_, err = c.Decode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok = err.(errorx.ErrorWithCode)
 	require.True(t, ok)
diff --git a/internal/converter/protobuf/converter_test.go b/internal/converter/protobuf/converter_test.go
index 37bea85c..ba2b0247 100644
--- a/internal/converter/protobuf/converter_test.go
+++ b/internal/converter/protobuf/converter_test.go
@@ -16,30 +16,28 @@ package protobuf
 
 import (
 	"fmt"
-	"os"
-	"path/filepath"
 	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/schema"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestOneOfDecode(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test5.proto", "", "Book")
 	require.NoError(t, err)
-	v, err := c.Decode([]byte{0x0A, 0x03, 0x31, 0x32, 0x33, 0x1A, 0x04, 0x31, 0x32, 0x33, 0x34})
+	v, err := c.Decode(ctx, []byte{0x0A, 0x03, 0x31, 0x32, 0x33, 0x1A, 0x04, 0x31, 0x32, 0x33, 0x34})
 	require.NoError(t, err)
 	require.Equal(t, map[string]interface{}{
 		"a": "123", "c": "1234",
 	}, v)
 
-	v, err = c.Decode([]byte{0x0A, 0x03, 0x31, 0x32, 0x33, 0x22, 0x04, 0x31, 0x32, 0x33, 0x34})
+	v, err = c.Decode(ctx, []byte{0x0A, 0x03, 0x31, 0x32, 0x33, 0x22, 0x04, 0x31, 0x32, 0x33, 0x34})
 	require.NoError(t, err)
 	require.Equal(t, map[string]interface{}{
 		"a": "123", "d": "1234",
@@ -47,6 +45,7 @@ func TestOneOfDecode(t *testing.T) {
 }
 
 func TestEncode(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test1.proto", "", "Person")
 	if err != nil {
 		t.Fatal(err)
@@ -84,7 +83,7 @@ func TestEncode(t *testing.T) {
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	for i, tt := range tests {
-		a, err := c.Encode(tt.m)
+		a, err := c.Encode(ctx, tt.m)
 		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
 			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
 		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
@@ -94,6 +93,7 @@ func TestEncode(t *testing.T) {
 }
 
 func TestEmbedType(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test3.proto", "", "DrivingData")
 	if err != nil {
 		t.Fatal(err)
@@ -126,16 +126,17 @@ func TestEmbedType(t *testing.T) {
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	for _, tt := range tests {
-		a, err := c.Encode(tt.m)
+		a, err := c.Encode(ctx, tt.m)
 		assert.NoError(t, err)
 		assert.Equal(t, tt.r, a)
-		m, err := c.Decode(a)
+		m, err := c.Decode(ctx, a)
 		assert.NoError(t, err)
 		assert.Equal(t, tt.d, m)
 	}
 }
 
 func TestDecode(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test1.proto", "", "Person")
 	if err != nil {
 		t.Fatal(err)
@@ -169,79 +170,15 @@ func TestDecode(t *testing.T) {
 
 	for i, tt := range tests {
 		t.Run(fmt.Sprintf("test %d", i), func(t *testing.T) {
-			a, err := c.Decode(tt.r)
+			a, err := c.Decode(ctx, tt.r)
 			assert.NoError(t, err)
 			assert.Equal(t, tt.m, a)
 		})
 	}
 }
 
-func TestStatic(t *testing.T) {
-	dataDir, err := conf.GetDataLoc()
-	if err != nil {
-		t.Fatal(err)
-	}
-	etcDir := filepath.Join(dataDir, "schemas", "custom")
-	err = os.MkdirAll(etcDir, os.ModePerm)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(etcDir)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}()
-	// build the so file into data/test prior to running the test
-	// Copy the helloworld.so
-	bytesRead, err := os.ReadFile(filepath.Join(dataDir, "helloworld.so"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = os.WriteFile(filepath.Join(etcDir, "helloworld.so"), bytesRead, 0o755)
-	if err != nil {
-		t.Fatal(err)
-	}
-	schema.InitRegistry()
-	c, err := NewConverter("../../schema/test/test1.proto", "../../../data/test/schemas/custom/helloworld.so", "HelloReply")
-	if err != nil {
-		t.Fatal(err)
-	}
-	tests := []struct {
-		m map[string]interface{}
-		r []byte
-		e string
-	}{
-		{
-			m: map[string]interface{}{
-				"message": "test",
-			},
-			r: []byte{0x0a, 0x04, 0x74, 0x65, 0x73, 0x74},
-		}, {
-			m: map[string]interface{}{
-				"message": "another test 2",
-			},
-			r: []byte{0x0a, 0x0e, 0x61, 0x6e, 0x6f, 0x74, 0x68, 0x65, 0x72, 0x20, 0x74, 0x65, 0x73, 0x74, 0x20, 0x32},
-		},
-	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	for i, tt := range tests {
-		a, err := c.Encode(tt.m)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.r, a) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%x\n\ngot=%x\n\n", i, tt.r, a)
-		}
-		m, err := c.Decode(tt.r)
-		if !reflect.DeepEqual(tt.e, testx.Errstring(err)) {
-			t.Errorf("%d.error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.e, err)
-		} else if tt.e == "" && !reflect.DeepEqual(tt.m, m) {
-			t.Errorf("%d. \n\nresult mismatch:\n\nexp=%v\n\ngot=%v\n\n", i, tt.m, m)
-		}
-	}
-}
-
 func TestDecodeProto3(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test4.proto", "", "Classroom")
 	if err != nil {
 		t.Fatal(err)
@@ -276,7 +213,7 @@ func TestDecodeProto3(t *testing.T) {
 
 	for i, tt := range tests {
 		t.Run(fmt.Sprintf("test %d", i), func(t *testing.T) {
-			a, err := c.Decode(tt.r)
+			a, err := c.Decode(ctx, tt.r)
 			assert.NoError(t, err)
 			assert.Equal(t, tt.m, a)
 		})
@@ -284,6 +221,7 @@ func TestDecodeProto3(t *testing.T) {
 }
 
 func TestEncodeDecodeForAllTypes(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/alltypes.proto", "", "AllTypesTest")
 	if err != nil {
 		t.Fatal(err)
@@ -337,10 +275,10 @@ func TestEncodeDecodeForAllTypes(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			a, err := c.Encode(tt.m)
+			a, err := c.Encode(ctx, tt.m)
 			assert.NoError(t, err)
 			assert.Equal(t, tt.b, a)
-			m, err := c.Decode(a)
+			m, err := c.Decode(ctx, a)
 			assert.NoError(t, err)
 			assert.Equal(t, tt.r, m)
 		})
@@ -348,14 +286,15 @@ func TestEncodeDecodeForAllTypes(t *testing.T) {
 }
 
 func TestErr(t *testing.T) {
+	ctx := mockContext.NewMockContext("test", "op1")
 	c, err := NewConverter("../../schema/test/test1.proto", "", "Person")
 	require.NoError(t, err)
-	_, err = c.Encode("123")
+	_, err = c.Encode(ctx, "123")
 	require.Error(t, err)
 	errWithCode, ok := err.(errorx.ErrorWithCode)
 	require.True(t, ok)
 	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
-	_, err = c.Decode(nil)
+	_, err = c.Decode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok = err.(errorx.ErrorWithCode)
 	require.True(t, ok)
diff --git a/internal/converter/protobuf/test/Makefile b/internal/converter/protobuf/test/Makefile
deleted file mode 100644
index 22f23892..00000000
--- a/internal/converter/protobuf/test/Makefile
+++ /dev/null
@@ -1,5 +0,0 @@
-.PHONY: build
-build:
-	@echo "Start build"
-	CGO_ENABLED=1 go build -trimpath --buildmode=plugin
-	@echo "Build successfully"
\ No newline at end of file
diff --git a/internal/converter/protobuf/test/go.mod b/internal/converter/protobuf/test/go.mod
deleted file mode 100644
index 86185d65..00000000
--- a/internal/converter/protobuf/test/go.mod
+++ /dev/null
@@ -1,8 +0,0 @@
-module helloworld
-
-go 1.22
-
-require (
-	github.com/golang/protobuf v1.5.2
-	google.golang.org/protobuf v1.33.0
-)
diff --git a/internal/converter/protobuf/test/go.sum b/internal/converter/protobuf/test/go.sum
deleted file mode 100644
index 0052f576..00000000
--- a/internal/converter/protobuf/test/go.sum
+++ /dev/null
@@ -1,11 +0,0 @@
-github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
-github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
-github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
-github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
-github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
-google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
-google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
diff --git a/internal/converter/protobuf/test/helloworld.pb.go b/internal/converter/protobuf/test/helloworld.pb.go
deleted file mode 100644
index 789002cd..00000000
--- a/internal/converter/protobuf/test/helloworld.pb.go
+++ /dev/null
@@ -1,217 +0,0 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// Code generated by protoc-gen-go. DO NOT EDIT.
-// versions:
-// 	protoc-gen-go v1.28.1
-// 	protoc        v3.6.1
-// source: helloworld.proto
-
-package main
-
-import (
-	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
-	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
-	reflect "reflect"
-	sync "sync"
-)
-
-const (
-	// Verify that this generated code is sufficiently up-to-date.
-	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
-	// Verify that runtime/protoimpl is sufficiently up-to-date.
-	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
-)
-
-type HelloRequest struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
-}
-
-func (x *HelloRequest) Reset() {
-	*x = HelloRequest{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_helloworld_proto_msgTypes[0]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *HelloRequest) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*HelloRequest) ProtoMessage() {}
-
-func (x *HelloRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_helloworld_proto_msgTypes[0]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use HelloRequest.ProtoReflect.Descriptor instead.
-func (*HelloRequest) Descriptor() ([]byte, []int) {
-	return file_helloworld_proto_rawDescGZIP(), []int{0}
-}
-
-func (x *HelloRequest) GetName() string {
-	if x != nil {
-		return x.Name
-	}
-	return ""
-}
-
-type HelloReply struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
-}
-
-func (x *HelloReply) Reset() {
-	*x = HelloReply{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_helloworld_proto_msgTypes[1]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *HelloReply) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*HelloReply) ProtoMessage() {}
-
-func (x *HelloReply) ProtoReflect() protoreflect.Message {
-	mi := &file_helloworld_proto_msgTypes[1]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use HelloReply.ProtoReflect.Descriptor instead.
-func (*HelloReply) Descriptor() ([]byte, []int) {
-	return file_helloworld_proto_rawDescGZIP(), []int{1}
-}
-
-func (x *HelloReply) GetMessage() string {
-	if x != nil {
-		return x.Message
-	}
-	return ""
-}
-
-var File_helloworld_proto protoreflect.FileDescriptor
-
-var file_helloworld_proto_rawDesc = []byte{
-	0x0a, 0x10, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x77, 0x6f, 0x72, 0x6c, 0x64, 0x2e, 0x70, 0x72, 0x6f,
-	0x74, 0x6f, 0x22, 0x22, 0x0a, 0x0c, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65,
-	0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
-	0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x22, 0x26, 0x0a, 0x0a, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x52,
-	0x65, 0x70, 0x6c, 0x79, 0x12, 0x18, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18,
-	0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x62, 0x06,
-	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
-}
-
-var (
-	file_helloworld_proto_rawDescOnce sync.Once
-	file_helloworld_proto_rawDescData = file_helloworld_proto_rawDesc
-)
-
-func file_helloworld_proto_rawDescGZIP() []byte {
-	file_helloworld_proto_rawDescOnce.Do(func() {
-		file_helloworld_proto_rawDescData = protoimpl.X.CompressGZIP(file_helloworld_proto_rawDescData)
-	})
-	return file_helloworld_proto_rawDescData
-}
-
-var file_helloworld_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
-var file_helloworld_proto_goTypes = []interface{}{
-	(*HelloRequest)(nil), // 0: HelloRequest
-	(*HelloReply)(nil),   // 1: HelloReply
-}
-var file_helloworld_proto_depIdxs = []int32{
-	0, // [0:0] is the sub-list for method output_type
-	0, // [0:0] is the sub-list for method input_type
-	0, // [0:0] is the sub-list for extension type_name
-	0, // [0:0] is the sub-list for extension extendee
-	0, // [0:0] is the sub-list for field type_name
-}
-
-func init() { file_helloworld_proto_init() }
-func file_helloworld_proto_init() {
-	if File_helloworld_proto != nil {
-		return
-	}
-	if !protoimpl.UnsafeEnabled {
-		file_helloworld_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*HelloRequest); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_helloworld_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*HelloReply); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-	}
-	type x struct{}
-	out := protoimpl.TypeBuilder{
-		File: protoimpl.DescBuilder{
-			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
-			RawDescriptor: file_helloworld_proto_rawDesc,
-			NumEnums:      0,
-			NumMessages:   2,
-			NumExtensions: 0,
-			NumServices:   0,
-		},
-		GoTypes:           file_helloworld_proto_goTypes,
-		DependencyIndexes: file_helloworld_proto_depIdxs,
-		MessageInfos:      file_helloworld_proto_msgTypes,
-	}.Build()
-	File_helloworld_proto = out.File
-	file_helloworld_proto_rawDesc = nil
-	file_helloworld_proto_goTypes = nil
-	file_helloworld_proto_depIdxs = nil
-}
diff --git a/internal/converter/protobuf/test/helloworld.proto b/internal/converter/protobuf/test/helloworld.proto
deleted file mode 100644
index c0d57e1e..00000000
--- a/internal/converter/protobuf/test/helloworld.proto
+++ /dev/null
@@ -1,9 +0,0 @@
-syntax = "proto3";
-
-message HelloRequest {
-  string name = 1;
-}
-
-message HelloReply {
-  string message = 1;
-}
\ No newline at end of file
diff --git a/internal/converter/protobuf/test/helloworld_wrapper.go b/internal/converter/protobuf/test/helloworld_wrapper.go
deleted file mode 100644
index e68c37b3..00000000
--- a/internal/converter/protobuf/test/helloworld_wrapper.go
+++ /dev/null
@@ -1,89 +0,0 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package main
-
-import (
-	"fmt"
-
-	"github.com/golang/protobuf/proto"
-)
-
-func (x *HelloReply) Encode(d interface{}) ([]byte, error) {
-	switch r := d.(type) {
-	case map[string]interface{}:
-		t, ok := r["message"]
-		if ok {
-			if v, ok := t.(string); ok {
-				x.Message = v
-			} else {
-				fmt.Println("message is not string")
-			}
-		} else {
-			// if required, return error
-			fmt.Println("message is not found")
-		}
-		return proto.Marshal(x)
-	default:
-		return nil, fmt.Errorf("unsupported type %v, must be a map", d)
-	}
-}
-
-func (x *HelloReply) Decode(b []byte) (interface{}, error) {
-	err := proto.Unmarshal(b, x)
-	if err != nil {
-		return nil, err
-	}
-	result := make(map[string]interface{}, 1)
-	result["message"] = x.Message
-	return result, nil
-}
-
-func GetHelloReply() interface{} {
-	return &HelloReply{}
-}
-
-func (x *HelloRequest) Encode(d interface{}) ([]byte, error) {
-	switch r := d.(type) {
-	case map[string]interface{}:
-		t, ok := r["name"]
-		if ok {
-			if v, ok := t.(string); ok {
-				x.Name = v
-			} else {
-				return nil, fmt.Errorf("name is not string")
-			}
-		} else {
-			// if required, return error
-			fmt.Println("message is not found")
-		}
-		return proto.Marshal(x)
-	default:
-		return nil, fmt.Errorf("unsupported type %v, must be a map", d)
-	}
-}
-
-func (x *HelloRequest) Decode(b []byte) (interface{}, error) {
-	err := proto.Unmarshal(b, x)
-	if err != nil {
-		return nil, err
-	}
-	result := make(map[string]interface{}, 1)
-	result["name"] = x.Name
-	return result, nil
-}
-
-func GetHelloRequest() interface{} {
-	return &HelloRequest{}
-}
diff --git a/internal/io/file/file_stream_test.go b/internal/io/file/file_stream_test.go
new file mode 100644
index 00000000..2e276e67
--- /dev/null
+++ b/internal/io/file/file_stream_test.go
@@ -0,0 +1,173 @@
+// Copyright 2023-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// 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 file
+
+import (
+	"os"
+	"reflect"
+	"testing"
+
+	"github.com/lf-edge/ekuiper/v2/internal/compressor"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/message"
+)
+
+func TestFileSinkCompress_Collect(t *testing.T) {
+	tests := []struct {
+		name     string
+		ft       FileType
+		fname    string
+		content  []byte
+		compress string
+	}{
+		{
+			name:    "lines",
+			ft:      LINES_TYPE,
+			fname:   "test_lines",
+			content: []byte("{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+		},
+		{
+			name:    "json",
+			ft:      JSON_TYPE,
+			fname:   "test_json",
+			content: []byte(`[{"key":"value1"},{"key":"value2"}]`),
+		},
+
+		{
+			name:     "lines",
+			ft:       LINES_TYPE,
+			fname:    "test_lines",
+			content:  []byte("{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+			compress: GZIP,
+		},
+
+		{
+			name:     "json",
+			ft:       JSON_TYPE,
+			fname:    "test_json",
+			content:  []byte(`[{"key":"value1"},{"key":"value2"}]`),
+			compress: GZIP,
+		},
+
+		{
+			name:     "lines",
+			ft:       LINES_TYPE,
+			fname:    "test_lines",
+			content:  []byte("{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+			compress: ZSTD,
+		},
+
+		{
+			name:     "json",
+			ft:       JSON_TYPE,
+			fname:    "test_json",
+			content:  []byte(`[{"key":"value1"},{"key":"value2"}]`),
+			compress: ZSTD,
+		},
+	}
+
+	// Create a stream context for testing
+	contextLogger := conf.Log.WithField("rule", "test2")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a temporary file for testing
+			tmpfile, err := os.CreateTemp("", tt.fname)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer os.Remove(tmpfile.Name())
+			// Create a file sink with the temporary file path
+			sink := &fileSink{}
+			f := message.FormatJson
+			if tt.ft == CSV_TYPE {
+				f = message.FormatDelimited
+			}
+			err = sink.Provision(ctx, map[string]interface{}{
+				"path":               tmpfile.Name(),
+				"fileType":           tt.ft,
+				"hasHeader":          true,
+				"format":             f,
+				"rollingNamePattern": "none",
+				"compression":        tt.compress,
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = sink.Connect(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if err := sink.Collect(ctx, []byte("{\"key\":\"value1\"}")); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			// Test collecting another map item
+			if err := sink.Collect(ctx, []byte("{\"key\":\"value2\"}")); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+			if err = sink.Close(ctx); err != nil {
+				t.Errorf("unexpected close error: %s", err)
+			}
+			contents, err := os.ReadFile(tmpfile.Name())
+			if err != nil {
+				t.Fatal(err)
+			}
+			if tt.compress != "" {
+				decompressor, _ := compressor.GetDecompressor(tt.compress)
+				decompress, err := decompressor.Decompress(contents)
+				if err != nil {
+					t.Errorf("%v", err)
+				}
+
+				if !reflect.DeepEqual(decompress, tt.content) {
+					t.Errorf("\nexpected\t %q \nbut got\t\t %q", tt.content, string(contents))
+				}
+			} else {
+				if !reflect.DeepEqual(contents, tt.content) {
+					t.Errorf("\nexpected\t %q \nbut got\t\t %q", tt.content, string(contents))
+				}
+			}
+
+			// Read the contents of the temporary file and check if they match the collected items
+			//r := &FileSource{}
+			//dir := filepath.Dir(tmpfile.Name())
+			//filename := filepath.Base(tmpfile.Name())
+			//p := map[string]interface{}{
+			//	"path":          filepath.Join(dir),
+			//	"decompression": tt.compress,
+			//	"fileType":      tt.ft,
+			//}
+			//
+			//err = r.Configure(filename, p)
+			//if err != nil {
+			//	t.Errorf(err.Error())
+			//	return
+			//}
+			//meta := map[string]interface{}{
+			//	"file": filepath.Join(dir, filename),
+			//}
+			//mc := conf.Clock.(*clock.Mock)
+			//exp := []api.SourceTuple{
+			//	api.NewDefaultSourceTupleWithTime(map[string]interface{}{"key": "value1"}, meta, mc.Now()),
+			//	api.NewDefaultSourceTupleWithTime(map[string]interface{}{"key": "value2"}, meta, mc.Now()),
+			//}
+			//mock.TestSourceOpen(r, exp, t)
+		})
+	}
+}
diff --git a/internal/io/file/sink_test.go b/internal/io/file/sink_test.go
new file mode 100644
index 00000000..8af90b3f
--- /dev/null
+++ b/internal/io/file/sink_test.go
@@ -0,0 +1,710 @@
+// Copyright 2023 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package file
+
+import (
+	"crypto/aes"
+	"crypto/cipher"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/compressor"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/encryptor"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
+	"github.com/lf-edge/ekuiper/v2/pkg/message"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+// Unit test for Configure function
+func TestConfigure(t *testing.T) {
+	ctx := mockContext.NewMockContext("test1", "test")
+	props := map[string]interface{}{
+		"interval": 500,
+		"path":     "test",
+	}
+	m := &fileSink{}
+	err := m.Provision(ctx, props)
+	if err != nil {
+		t.Errorf("Configure() error = %v, wantErr nil", err)
+	}
+	if m.c.Path != "test" {
+		t.Errorf("Configure() Path = %v, want test", m.c.Path)
+	}
+	err = m.Provision(ctx, map[string]interface{}{"interval": 500, "path": ""})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+	err = m.Provision(ctx, map[string]interface{}{"fileType": "csv2"})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+	err = m.Provision(ctx, map[string]interface{}{
+		"interval": 500,
+		"path":     "test",
+		"fileType": "csv",
+	})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+	err = m.Provision(ctx, map[string]interface{}{"interval": 60, "path": "test", "checkInterval": -1})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+	err = m.Provision(ctx, map[string]interface{}{"rollingInterval": -1})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+	err = m.Provision(ctx, map[string]interface{}{"rollingCount": -1})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+	err = m.Provision(ctx, map[string]interface{}{"rollingCount": 0, "rollingInterval": 0})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+	err = m.Provision(ctx, map[string]interface{}{"RollingNamePattern": "test"})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+	err = m.Provision(ctx, map[string]interface{}{"RollingNamePattern": 0})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+
+	for k := range compressionTypes {
+		err = m.Provision(ctx, map[string]interface{}{
+			"interval":           500,
+			"path":               "test",
+			"compression":        k,
+			"rollingNamePattern": "suffix",
+		})
+		if err != nil {
+			t.Errorf("Configure() error = %v, wantErr nil", err)
+		}
+		if m.c.Compression != k {
+			t.Errorf("Configure() Compression = %v, want %v", m.c.Compression, k)
+		}
+	}
+
+	err = m.Provision(ctx, map[string]interface{}{
+		"interval":           500,
+		"path":               "test",
+		"compression":        "",
+		"rollingNamePattern": "suffix",
+	})
+	if err != nil {
+		t.Errorf("Configure() error = %v, wantErr nil", err)
+	}
+	if m.c.Compression != "" {
+		t.Errorf("Configure() Compression = %v, want %v", m.c.Compression, "")
+	}
+
+	err = m.Provision(ctx, map[string]interface{}{
+		"interval":    500,
+		"path":        "test",
+		"compression": "not_exist_algorithm",
+	})
+	if err == nil {
+		t.Errorf("Configure() error = %v, wantErr not nil", err)
+	}
+}
+
+func TestFileSink_Configure(t *testing.T) {
+	defaultCheckInterval := (5 * time.Minute).Milliseconds()
+
+	tests := []struct {
+		name string
+		c    *sinkConf
+		p    map[string]interface{}
+	}{
+		{
+			name: "default configurations",
+			c: &sinkConf{
+				CheckInterval: defaultCheckInterval,
+				Path:          "cache",
+				FileType:      LINES_TYPE,
+				RollingCount:  1000000,
+			},
+			p: map[string]interface{}{},
+		},
+		{
+			name: "new props",
+			c: &sinkConf{
+				CheckInterval:      500,
+				Path:               "test",
+				FileType:           CSV_TYPE,
+				Format:             message.FormatDelimited,
+				Delimiter:          ",",
+				RollingCount:       1000000,
+				RollingNamePattern: "none",
+			},
+			p: map[string]interface{}{
+				"checkInterval":      500,
+				"path":               "test",
+				"fileType":           "csv",
+				"format":             message.FormatDelimited,
+				"rollingNamePattern": "none",
+			},
+		},
+		{ // only set rolling interval
+			name: "rolling",
+			c: &sinkConf{
+				CheckInterval:   defaultCheckInterval,
+				Path:            "cache",
+				FileType:        LINES_TYPE,
+				RollingInterval: 500,
+				RollingCount:    0,
+			},
+			p: map[string]interface{}{
+				"rollingInterval": 500,
+				"rollingCount":    0,
+			},
+		},
+		{
+			name: "fields",
+			c: &sinkConf{
+				CheckInterval:   defaultCheckInterval,
+				Path:            "cache",
+				FileType:        LINES_TYPE,
+				RollingInterval: 500,
+				RollingCount:    0,
+				Fields:          []string{"c", "a", "b"},
+			},
+			p: map[string]interface{}{
+				"rollingInterval": 500,
+				"rollingCount":    0,
+				"fields":          []string{"c", "a", "b"},
+			},
+		},
+	}
+	ctx := mockContext.NewMockContext("test1", "test")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := &fileSink{}
+			if err := m.Provision(ctx, tt.p); err != nil {
+				t.Errorf("fileSink.Configure() error = %v", err)
+				return
+			}
+			if !reflect.DeepEqual(m.c, tt.c) {
+				t.Errorf("fileSink.Configure() = %v, want %v", m.c, tt.c)
+			}
+		})
+	}
+}
+
+// Test single file writing and flush by close
+func TestFileSink_Collect(t *testing.T) {
+	tests := []struct {
+		name     string
+		ft       FileType
+		fname    string
+		content  []byte
+		compress string
+	}{
+		{
+			name:    "lines",
+			ft:      LINES_TYPE,
+			fname:   "test_lines",
+			content: []byte("{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+		},
+		{
+			name:    "json",
+			ft:      JSON_TYPE,
+			fname:   "test_json",
+			content: []byte(`[{"key":"value1"},{"key":"value2"}]`),
+		},
+		{
+			name:    "csv",
+			ft:      CSV_TYPE,
+			fname:   "test_csv",
+			content: []byte("key\n{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+		},
+		{
+			name:     "lines",
+			ft:       LINES_TYPE,
+			fname:    "test_lines",
+			content:  []byte("{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+			compress: GZIP,
+		},
+		{
+			name:     "json",
+			ft:       JSON_TYPE,
+			fname:    "test_json",
+			content:  []byte(`[{"key":"value1"},{"key":"value2"}]`),
+			compress: GZIP,
+		},
+		{
+			name:     "csv",
+			ft:       CSV_TYPE,
+			fname:    "test_csv",
+			content:  []byte("key\n{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+			compress: GZIP,
+		},
+
+		{
+			name:     "lines",
+			ft:       LINES_TYPE,
+			fname:    "test_lines",
+			content:  []byte("{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+			compress: ZSTD,
+		},
+		{
+			name:     "json",
+			ft:       JSON_TYPE,
+			fname:    "test_json",
+			content:  []byte(`[{"key":"value1"},{"key":"value2"}]`),
+			compress: ZSTD,
+		},
+		{
+			name:     "csv",
+			ft:       CSV_TYPE,
+			fname:    "test_csv",
+			content:  []byte("key\n{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+			compress: ZSTD,
+		},
+	}
+
+	ctx := mockContext.NewMockContext("test1", "test")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a temporary file for testing
+			tmpfile, err := os.CreateTemp("", tt.fname)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer os.Remove(tmpfile.Name())
+			// Create a file sink with the temporary file path
+			sink := &fileSink{}
+			f := message.FormatJson
+			if tt.ft == CSV_TYPE {
+				f = message.FormatDelimited
+			}
+			err = sink.Provision(ctx, map[string]interface{}{
+				"path":               tmpfile.Name(),
+				"fileType":           tt.ft,
+				"hasHeader":          true,
+				"format":             f,
+				"rollingNamePattern": "none",
+				"compression":        tt.compress,
+				"fields":             []string{"key"},
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = sink.Connect(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if err := sink.Collect(ctx, []byte("{\"key\":\"value1\"}")); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			// Test collecting another map item
+			if err := sink.Collect(ctx, []byte("{\"key\":\"value2\"}")); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+			if err = sink.Close(ctx); err != nil {
+				t.Errorf("unexpected close error: %s", err)
+			}
+			// Read the contents of the temporary file and check if they match the collected items
+			contents, err := os.ReadFile(tmpfile.Name())
+			if err != nil {
+				t.Fatal(err)
+			}
+			if tt.compress != "" {
+				decompressor, _ := compressor.GetDecompressor(tt.compress)
+				decompress, err := decompressor.Decompress(contents)
+				if err != nil {
+					t.Errorf("%v", err)
+				}
+
+				if !reflect.DeepEqual(decompress, tt.content) {
+					t.Errorf("\nexpected\t %q \nbut got\t\t %q", tt.content, string(contents))
+				}
+			} else {
+				if !reflect.DeepEqual(contents, tt.content) {
+					t.Errorf("\nexpected\t %q \nbut got\t\t %q", tt.content, string(contents))
+				}
+			}
+		})
+	}
+}
+
+// Test file rolling by time
+func TestFileSinkRolling_Collect(t *testing.T) {
+	// Remove existing files
+	err := filepath.Walk(".", func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if filepath.Ext(path) == ".log" {
+			fmt.Println("Deleting file:", path)
+			return os.Remove(path)
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	conf.IsTesting = true
+	tests := []struct {
+		name     string
+		ft       FileType
+		fname    string
+		contents [2][]byte
+		compress string
+	}{
+		{
+			name:  "lines",
+			ft:    LINES_TYPE,
+			fname: "test_lines.log",
+			contents: [2][]byte{
+				[]byte("{\"key\":\"value0\",\"ts\":460}\n{\"key\":\"value1\",\"ts\":910}\n{\"key\":\"value2\",\"ts\":1360}"),
+				[]byte("{\"key\":\"value3\",\"ts\":1810}\n{\"key\":\"value4\",\"ts\":2260}"),
+			},
+		},
+		{
+			name:  "json",
+			ft:    JSON_TYPE,
+			fname: "test_json.log",
+			contents: [2][]byte{
+				[]byte("[{\"key\":\"value0\",\"ts\":460},{\"key\":\"value1\",\"ts\":910},{\"key\":\"value2\",\"ts\":1360}]"),
+				[]byte("[{\"key\":\"value3\",\"ts\":1810},{\"key\":\"value4\",\"ts\":2260}]"),
+			},
+		},
+
+		{
+			name:  "lines",
+			ft:    LINES_TYPE,
+			fname: "test_lines_gzip.log",
+			contents: [2][]byte{
+				[]byte("{\"key\":\"value0\",\"ts\":460}\n{\"key\":\"value1\",\"ts\":910}\n{\"key\":\"value2\",\"ts\":1360}"),
+				[]byte("{\"key\":\"value3\",\"ts\":1810}\n{\"key\":\"value4\",\"ts\":2260}"),
+			},
+			compress: GZIP,
+		},
+		{
+			name:  "json",
+			ft:    JSON_TYPE,
+			fname: "test_json_gzip.log",
+			contents: [2][]byte{
+				[]byte("[{\"key\":\"value0\",\"ts\":460},{\"key\":\"value1\",\"ts\":910},{\"key\":\"value2\",\"ts\":1360}]"),
+				[]byte("[{\"key\":\"value3\",\"ts\":1810},{\"key\":\"value4\",\"ts\":2260}]"),
+			},
+			compress: GZIP,
+		},
+
+		{
+			name:  "lines",
+			ft:    LINES_TYPE,
+			fname: "test_lines_zstd.log",
+			contents: [2][]byte{
+				[]byte("{\"key\":\"value0\",\"ts\":460}\n{\"key\":\"value1\",\"ts\":910}\n{\"key\":\"value2\",\"ts\":1360}"),
+				[]byte("{\"key\":\"value3\",\"ts\":1810}\n{\"key\":\"value4\",\"ts\":2260}"),
+			},
+			compress: ZSTD,
+		},
+		{
+			name:  "json",
+			ft:    JSON_TYPE,
+			fname: "test_json_zstd.log",
+			contents: [2][]byte{
+				[]byte("[{\"key\":\"value0\",\"ts\":460},{\"key\":\"value1\",\"ts\":910},{\"key\":\"value2\",\"ts\":1360}]"),
+				[]byte("[{\"key\":\"value3\",\"ts\":1810},{\"key\":\"value4\",\"ts\":2260}]"),
+			},
+			compress: ZSTD,
+		},
+	}
+
+	// Create a stream context for testing
+	ctx := mockContext.NewMockContext("rule", "testRolling")
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a file sink with the temporary file path
+			sink := &fileSink{}
+			err := sink.Provision(ctx, map[string]interface{}{
+				"path":               tt.fname,
+				"fileType":           tt.ft,
+				"rollingInterval":    1000,
+				"checkInterval":      500,
+				"rollingCount":       0,
+				"rollingNamePattern": "suffix",
+				"compression":        tt.compress,
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			mockclock.ResetClock(10)
+			err = sink.Connect(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			c := mockclock.GetMockClock()
+
+			for i := 0; i < 5; i++ {
+				c.Add(450 * time.Millisecond)
+				m := map[string]interface{}{"key": "value" + strconv.Itoa(i), "ts": c.Now().UnixMilli()}
+				b, err := json.Marshal(m)
+				assert.NoError(t, err)
+				if err := sink.Collect(ctx, b); err != nil {
+					t.Errorf("unexpected error: %s", err)
+				}
+			}
+			c.After(2000 * time.Millisecond)
+			if err = sink.Close(ctx); err != nil {
+				t.Errorf("unexpected close error: %s", err)
+			}
+			// Should write to 2 files
+			for i := 0; i < 2; i++ {
+				// Read the contents of the temporary file and check if they match the collected items
+				var fn string
+				if tt.compress != "" {
+					fn = fmt.Sprintf("test_%s_%s-%d.log", tt.ft, tt.compress, 460+1350*i)
+				} else {
+					fn = fmt.Sprintf("test_%s-%d.log", tt.ft, 460+1350*i)
+				}
+
+				var contents []byte
+				contents, err := os.ReadFile(fn)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if tt.compress != "" {
+					decompressor, _ := compressor.GetDecompressor(tt.compress)
+					contents, err = decompressor.Decompress(contents)
+					if err != nil {
+						t.Errorf("%v", err)
+					}
+				}
+				if !reflect.DeepEqual(contents, tt.contents[i]) {
+					t.Errorf("\nexpected\t %q \nbut got\t\t %q", tt.contents[i], string(contents))
+				}
+			}
+		})
+	}
+}
+
+func TestFileSinkReopen(t *testing.T) {
+	// Remove existing files
+	err := filepath.Walk(".", func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if filepath.Ext(path) == ".log" {
+			fmt.Println("Deleting file:", path)
+			return os.Remove(path)
+		}
+		return nil
+	})
+	assert.NoError(t, err)
+	conf.IsTesting = true
+	tmpfile, err := os.CreateTemp("", "reopen.log")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.Remove(tmpfile.Name())
+	// Create a stream context for testing
+	ctx := mockContext.NewMockContext("test1", "test")
+
+	sink := &fileSink{}
+	err = sink.Provision(ctx, map[string]interface{}{
+		"path":               tmpfile.Name(),
+		"fileType":           LINES_TYPE,
+		"format":             "json",
+		"rollingNamePattern": "none",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = sink.Connect(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Test collecting a map item
+	if err := sink.Collect(ctx, []byte("{\"key\":\"value1\"}")); err != nil {
+		t.Errorf("unexpected error: %s", err)
+	}
+	sink.Close(ctx)
+
+	exp := []byte(`{"key":"value1"}`)
+	contents, err := os.ReadFile(tmpfile.Name())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(contents, exp) {
+		t.Errorf("\nexpected\t %q \nbut got\t\t %q", string(exp), string(contents))
+	}
+
+	sink = &fileSink{}
+	err = sink.Provision(ctx, map[string]interface{}{
+		"path":               tmpfile.Name(),
+		"fileType":           LINES_TYPE,
+		"hasHeader":          true,
+		"format":             "json",
+		"rollingNamePattern": "none",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = sink.Connect(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Test collecting another map item
+	if err := sink.Collect(ctx, []byte("{\"key\":\"value2\"}")); err != nil {
+		t.Errorf("unexpected error: %s", err)
+	}
+	if err = sink.Close(ctx); err != nil {
+		t.Errorf("unexpected close error: %s", err)
+	}
+
+	exp = []byte(`{"key":"value2"}`)
+	contents, err = os.ReadFile(tmpfile.Name())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(contents, exp) {
+		t.Errorf("\nexpected\t %q \nbut got\t\t %q", string(exp), string(contents))
+	}
+}
+
+// Test single file writing and flush by close
+func TestFileCompressAndEncrypt(t *testing.T) {
+	conf.InitConf()
+	tests := []struct {
+		name       string
+		ft         FileType
+		fname      string
+		content    []byte
+		compress   string
+		encryption string
+	}{
+		{
+			name:       "lines with encryption",
+			ft:         LINES_TYPE,
+			fname:      "test_lines",
+			content:    []byte("{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+			encryption: "aes",
+		},
+		{
+			name:       "lines with compress and encryption",
+			ft:         LINES_TYPE,
+			fname:      "test_lines",
+			content:    []byte("{\"key\":\"value1\"}\n{\"key\":\"value2\"}"),
+			compress:   GZIP,
+			encryption: "aes",
+		},
+	}
+
+	// Create a stream context for testing
+	ctx := mockContext.NewMockContext("rule1", "op1")
+	_ = os.Mkdir("tmp", 0o777)
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a temporary file for testing
+			tmpfile, err := os.CreateTemp("tmp", tt.fname)
+			if err != nil {
+				t.Fatal(err)
+			}
+			tmpfile.Close()
+			defer os.Remove(tmpfile.Name())
+			// Create a file sink with the temporary file path
+			sink := &fileSink{}
+			f := message.FormatJson
+			err = sink.Provision(ctx, map[string]interface{}{
+				"path":               tmpfile.Name(),
+				"fileType":           tt.ft,
+				"format":             f,
+				"rollingNamePattern": "none",
+				"compression":        tt.compress,
+				"fields":             []string{"key"},
+				"encryption":         tt.encryption,
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = sink.Connect(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			// Test collecting a map item
+			if err := sink.Collect(ctx, []byte("{\"key\":\"value1\"}")); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			// Test collecting another map item
+			if err := sink.Collect(ctx, []byte("{\"key\":\"value2\"}")); err != nil {
+				t.Errorf("unexpected error: %s", err)
+			}
+
+			if err = sink.Close(ctx); err != nil {
+				t.Errorf("unexpected close error: %s", err)
+			}
+			// Read the contents of the temporary file and check if they match the collected items
+			contents, err := os.ReadFile(tmpfile.Name())
+			if err != nil {
+				t.Fatal(err)
+			}
+			// Decrypt then uncompress
+			key, _, err := encryptor.GetKeyIv()
+			assert.NoError(t, err)
+			revert := Decrypt(key, contents)
+			// uncompress
+			if tt.compress != "" {
+				decompressor, _ := compressor.GetDecompressor(tt.compress)
+				decompress, err := decompressor.Decompress(revert)
+				if err != nil {
+					t.Errorf("%v", err)
+				}
+
+				assert.Equal(t, decompress, tt.content)
+			} else {
+				assert.Equal(t, revert, tt.content)
+			}
+		})
+	}
+}
+
+func Decrypt(key []byte, contents []byte) []byte {
+	// Create a new AES cipher block using the key
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		panic(err)
+	}
+	// Get IV from the encrypted data
+	iv := contents[:aes.BlockSize]
+	// Get the actual encrypted data
+	secret := contents[aes.BlockSize:]
+	// create a new CFB decrypter
+	dstream := cipher.NewCFBDecrypter(block, iv)
+	// decrypt the data
+	decrypted := make([]byte, len(secret))
+	dstream.XORKeyStream(decrypted, secret)
+	return decrypted
+}
diff --git a/internal/processor/stream_test.go b/internal/processor/stream_test.go
index b58fd263..726a7939 100644
--- a/internal/processor/stream_test.go
+++ b/internal/processor/stream_test.go
@@ -284,33 +284,13 @@ func TestInferredStream(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	etcDir := filepath.Join(dataDir, "schemas", "custom")
-	err = os.MkdirAll(etcDir, os.ModePerm)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(etcDir)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}()
-	// build the so file into data/test prior to running the test
-	bytesRead, err := os.ReadFile(filepath.Join(dataDir, "myFormat.so"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = os.WriteFile(filepath.Join(etcDir, "myFormat.so"), bytesRead, 0o755)
-	if err != nil {
-		t.Fatal(err)
-	}
 	petcDir := filepath.Join(dataDir, "schemas", "protobuf")
 	err = os.MkdirAll(petcDir, os.ModePerm)
 	if err != nil {
 		t.Fatal(err)
 	}
 	// Copy test2.proto
-	bytesRead, err = os.ReadFile("../schema/test/test2.proto")
+	bytesRead, err := os.ReadFile("../schema/test/test2.proto")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -343,20 +323,6 @@ func TestInferredStream(t *testing.T) {
 				"name":   {Type: "string"},
 				"author": {Type: "string"},
 			},
-		}, {
-			s: `CREATE STREAM demo2 () WITH (FORMAT="custom", DATASOURCE="demo", SCHEMAID="myFormat.Sample")`,
-			r: map[string]*ast.JsonStreamField{
-				"id":   {Type: "bigint"},
-				"name": {Type: "string"},
-				"age":  {Type: "bigint"},
-				"hobbies": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"indoor":  {Type: "array", Items: &ast.JsonStreamField{Type: "string"}},
-						"outdoor": {Type: "array", Items: &ast.JsonStreamField{Type: "string"}},
-					},
-				},
-			},
 		},
 	}
 
diff --git a/internal/schema/ext_inferer_custom_test.go b/internal/schema/ext_inferer_custom_test.go
deleted file mode 100644
index 8722fa8f..00000000
--- a/internal/schema/ext_inferer_custom_test.go
+++ /dev/null
@@ -1,80 +0,0 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package schema
-
-import (
-	"os"
-	"path/filepath"
-	"testing"
-
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/testx"
-	"github.com/lf-edge/ekuiper/v2/pkg/ast"
-)
-
-func init() {
-	testx.InitEnv("schema")
-}
-
-func TestInferCustom(t *testing.T) {
-	// Prepare test schema file
-	dataDir, err := conf.GetDataLoc()
-	if err != nil {
-		t.Fatal(err)
-	}
-	etcDir := filepath.Join(dataDir, "schemas", "custom")
-	err = os.MkdirAll(etcDir, os.ModePerm)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer func() {
-		err = os.RemoveAll(etcDir)
-		if err != nil {
-			t.Fatal(err)
-		}
-	}()
-	// build the so file into data/test prior to running the test
-	// Copy the helloworld.so
-	bytesRead, err := os.ReadFile(filepath.Join(dataDir, "myFormat.so"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = os.WriteFile(filepath.Join(etcDir, "myFormat.so"), bytesRead, 0o755)
-	if err != nil {
-		t.Fatal(err)
-	}
-	InitRegistry()
-
-	// Test
-	result, err := InferCustom("myFormat", "Sample")
-	if err != nil {
-		t.Errorf("Infer custom format error: %v", err)
-		return
-	}
-	expected := ast.StreamFields{
-		{Name: "id", FieldType: &ast.BasicType{Type: ast.BIGINT}},
-		{Name: "name", FieldType: &ast.BasicType{Type: ast.STRINGS}},
-		{Name: "age", FieldType: &ast.BasicType{Type: ast.BIGINT}},
-		{Name: "hobbies", FieldType: &ast.RecType{
-			StreamFields: []ast.StreamField{
-				{Name: "indoor", FieldType: &ast.ArrayType{Type: ast.STRINGS}},
-				{Name: "outdoor", FieldType: &ast.ArrayType{Type: ast.STRINGS}},
-			},
-		}},
-	}
-	if len(result) != len(expected) {
-		t.Errorf("InferProtobuf result is not expected, got %v, expected %v", result, expected)
-	}
-}
diff --git a/internal/schema/registry_test.go b/internal/schema/registry_test.go
index f8adec20..4983939c 100644
--- a/internal/schema/registry_test.go
+++ b/internal/schema/registry_test.go
@@ -23,8 +23,13 @@ import (
 	"testing"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
 )
 
+func init() {
+	testx.InitEnv("schema")
+}
+
 func TestProtoRegistry(t *testing.T) {
 	// Move test schema file to etc dir
 	etcDir, err := conf.GetDataLoc()
diff --git a/internal/topo/context/default_test.go b/internal/topo/context/default_test.go
index 9485569e..b9d37667 100644
--- a/internal/topo/context/default_test.go
+++ b/internal/topo/context/default_test.go
@@ -26,7 +26,6 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/state"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/transform"
 )
 
 func TestState(t *testing.T) {
@@ -229,34 +228,3 @@ func TestParseTemplate(t *testing.T) {
 		}
 	}
 }
-
-func TestTransition(t *testing.T) {
-	var mockFunc transform.TransFunc = func(d interface{}) ([]byte, bool, error) {
-		return []byte(fmt.Sprintf("%v", d)), true, nil
-	}
-	tests := []struct {
-		data interface{}
-		r    []byte
-	}{
-		{
-			data: "hello",
-			r:    []byte(`hello`),
-		}, {
-			data: "world",
-			r:    []byte(`world`),
-		}, {
-			data: map[string]interface{}{"a": "hello"},
-			r:    []byte(`map[a:hello]`),
-		},
-	}
-
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-	ctx := Background().WithMeta("testTransRule", "op1", &state.MemoryStore{}).(*DefaultContext)
-	nc := WithValue(ctx, TransKey, mockFunc)
-	for i, tt := range tests {
-		r, _, _ := nc.TransformOutput(tt.data)
-		if !reflect.DeepEqual(tt.r, r) {
-			t.Errorf("%d\n\nstmt mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, string(tt.r), string(r))
-		}
-	}
-}
diff --git a/internal/topo/node/dedup_trigger_op_test.go b/internal/topo/node/dedup_trigger_op_test.go
new file mode 100644
index 00000000..d6635b69
--- /dev/null
+++ b/internal/topo/node/dedup_trigger_op_test.go
@@ -0,0 +1,262 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package node
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+func TestDedupTrigger(t *testing.T) {
+	// The test cases are stateful, so we need to run them one by one
+	tests := []struct {
+		name   string
+		args   []int64
+		hg     [][]int64
+		result []map[string]any
+	}{
+		{
+			name:   "initial",
+			args:   []int64{100, 200, 150, 1000},
+			hg:     [][]int64{{100, 200}},
+			result: []map[string]any{{"start_key": "100", "end_key": "200"}},
+		},
+		{
+			name:   "left empty",
+			args:   []int64{50, 70, 160, 1000},
+			hg:     [][]int64{{50, 70}, {100, 200}},
+			result: []map[string]any{{"start_key": "50", "end_key": "70"}},
+		},
+		{
+			name:   "right empty",
+			args:   []int64{250, 270, 170, 1000},
+			hg:     [][]int64{{50, 70}, {100, 200}, {250, 270}},
+			result: []map[string]any{{"start_key": "250", "end_key": "270"}},
+		},
+		{
+			name:   "right overlap",
+			args:   []int64{80, 260, 180, 1000},
+			hg:     [][]int64{{50, 70}, {80, 270}},
+			result: []map[string]any{{"start_key": "80", "end_key": "100"}, {"start_key": "200", "end_key": "250"}},
+		},
+		{
+			name:   "right empty 2",
+			args:   []int64{280, 290, 190, 1000},
+			hg:     [][]int64{{50, 70}, {80, 270}, {280, 290}},
+			result: []map[string]any{{"start_key": "280", "end_key": "290"}},
+		},
+		{
+			name:   "left empty 2",
+			args:   []int64{30, 40, 200, 1000},
+			hg:     [][]int64{{30, 40}, {50, 70}, {80, 270}, {280, 290}},
+			result: []map[string]any{{"start_key": "30", "end_key": "40"}},
+		},
+		{
+			name:   "left overlap",
+			args:   []int64{60, 275, 210, 1000},
+			hg:     [][]int64{{30, 40}, {50, 275}, {280, 290}},
+			result: []map[string]any{{"start_key": "70", "end_key": "80"}, {"start_key": "270", "end_key": "275"}},
+		},
+		{
+			name:   "both overlap",
+			args:   []int64{35, 285, 220, 1000},
+			hg:     [][]int64{{30, 290}},
+			result: []map[string]any{{"start_key": "40", "end_key": "50"}, {"start_key": "275", "end_key": "280"}},
+		},
+		{
+			name:   "inclusion",
+			args:   []int64{25, 300, 230, 1000},
+			hg:     [][]int64{{25, 300}},
+			result: []map[string]any{{"start_key": "25", "end_key": "30"}, {"start_key": "290", "end_key": "300"}},
+		},
+	}
+	ctx := context.NewMockContext("test", "test")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := doTrigger(ctx, tt.args[0], tt.args[1], tt.args[2], tt.args[3])
+			assert.NoError(t, err)
+			assert.Equal(t, tt.result, got)
+			st, err := ctx.GetState("histogram")
+			assert.NoError(t, err)
+			assert.Equal(t, tt.hg, st)
+		})
+	}
+}
+
+func TestDedupTriggerWithExp(t *testing.T) {
+	// The test cases are stateful, so we need to run them one by one
+	tests := []struct {
+		name   string
+		args   []int64
+		hg     [][]int64
+		result []map[string]any
+	}{
+		{
+			name:   "initial",
+			args:   []int64{100, 200, 150, 100},
+			hg:     [][]int64{{100, 200}},
+			result: []map[string]any{{"start_key": "100", "end_key": "200"}},
+		},
+		{
+			name:   "left empty",
+			args:   []int64{50, 70, 160, 100},
+			hg:     [][]int64{{60, 70}, {100, 200}},
+			result: []map[string]any{{"start_key": "60", "end_key": "70"}},
+		},
+		{
+			name:   "right empty",
+			args:   []int64{250, 270, 170, 100},
+			hg:     [][]int64{{60, 70}, {100, 200}, {250, 270}},
+			result: []map[string]any{{"start_key": "250", "end_key": "270"}},
+		},
+		{
+			name:   "right overlap",
+			args:   []int64{80, 260, 180, 100},
+			hg:     [][]int64{{80, 270}},
+			result: []map[string]any{{"start_key": "80", "end_key": "100"}, {"start_key": "200", "end_key": "250"}},
+		},
+		{
+			name:   "right empty 2",
+			args:   []int64{280, 290, 190, 100},
+			hg:     [][]int64{{80, 270}, {280, 290}},
+			result: []map[string]any{{"start_key": "280", "end_key": "290"}},
+		},
+		{
+			name:   "left empty 2",
+			args:   []int64{30, 40, 200, 100},
+			hg:     [][]int64{{80, 270}, {280, 290}},
+			result: nil,
+		},
+		{
+			name:   "left overlap",
+			args:   []int64{60, 275, 210, 100},
+			hg:     [][]int64{{80, 275}, {280, 290}},
+			result: []map[string]any{{"start_key": "270", "end_key": "275"}},
+		},
+		{
+			name:   "both overlap",
+			args:   []int64{35, 285, 220, 100},
+			hg:     [][]int64{{80, 290}},
+			result: []map[string]any{{"start_key": "275", "end_key": "280"}},
+		},
+		{
+			name:   "inclusion",
+			args:   []int64{25, 300, 230, 100},
+			hg:     [][]int64{{80, 300}},
+			result: []map[string]any{{"start_key": "290", "end_key": "300"}},
+		},
+	}
+	ctx := context.NewMockContext("test", "test")
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := doTrigger(ctx, tt.args[0], tt.args[1], tt.args[2], tt.args[3])
+			assert.NoError(t, err)
+			assert.Equal(t, tt.result, got)
+			st, err := ctx.GetState("histogram")
+			assert.NoError(t, err)
+			assert.Equal(t, tt.hg, st)
+		})
+	}
+}
+
+func TestQueue(t *testing.T) {
+	q := make(PriorityQueue, 0)
+	q.Push(&TriggerRequest{start: 100, end: 200})
+	q.Push(&TriggerRequest{start: 50, end: 70})
+	q.Push(&TriggerRequest{start: 250, end: 270})
+	q.Push(&TriggerRequest{start: 80, end: 260})
+	r := q.Pop()
+	assert.Equal(t, int64(50), r.start)
+	r = q.Peek()
+	assert.Equal(t, int64(100), r.start)
+	r = q.Pop()
+	assert.Equal(t, int64(100), r.start)
+	r = q.Pop()
+	assert.Equal(t, int64(80), r.start)
+	r = q.Pop()
+	assert.Equal(t, int64(250), r.start)
+}
+
+func TestExec(t *testing.T) {
+	timex.InitClock()
+	c := mockclock.GetMockClock()
+	node := NewDedupTriggerNode("dt", &def.RuleOption{BufferLength: 100}, "ranges", "begin", "finish", "ts", 99999)
+	ctx := context.NewMockContext("test", "test")
+	resultChan := make(chan any, 100)
+	errChan := make(chan error)
+	node.outputs["output"] = resultChan
+	node.Exec(ctx, errChan)
+	expResults := []any{
+		map[string]any{"begin": int64(90), "finish": int64(180), "ts": int64(180), "ruleId": "new", "ranges": []map[string]any{{"start_key": "90", "end_key": "180"}}},
+		map[string]any{"begin": int64(100), "finish": int64(200), "ts": int64(150), "ruleId": "test", "ranges": []map[string]any{{"start_key": "180", "end_key": "200"}}},
+		map[string]any{"begin": int64(110), "finish": int64(210), "ts": int64(160), "ruleId": "test", "ranges": []map[string]any{{"start_key": "200", "end_key": "210"}}},
+		map[string]any{"begin": int64(1700), "finish": int64(1800), "ts": int64(1800), "ruleId": "new", "ranges": []map[string]any{{"start_key": "1700", "end_key": "1800"}}},
+	}
+	inputData := []map[string]any{
+		{"begin": int64(100), "finish": int64(200), "ts": int64(150), "ruleId": "test"},
+		{"begin": int64(110), "finish": int64(210), "ts": int64(160), "ruleId": "test"},
+		{"begin": int64(90), "finish": int64(180), "ts": int64(180), "ruleId": "new"},
+		{"begin": int64(1700), "finish": int64(1800), "ts": int64(1800), "ruleId": "new"},
+	}
+	for i, data := range inputData {
+		node.input <- &xsql.Tuple{
+			Emitter:   "test",
+			Message:   data,
+			Timestamp: int64(i),
+		}
+		c.Add(10 * time.Millisecond)
+	}
+	go func() {
+		defer func() {
+			close(resultChan)
+		}()
+		for {
+			c.Add(50 * time.Millisecond)
+			mm := node.statManager.GetMetrics()
+			if mm[1] == int64(len(inputData)) {
+				return
+			}
+		}
+	}()
+	var results []any
+loop:
+	for {
+		select {
+		case r, ok := <-resultChan:
+			if !ok {
+				break loop
+			}
+			results = append(results, r)
+		case err := <-errChan:
+			t.Errorf("error: %v", err)
+			break loop
+		case <-time.After(1000 * time.Second):
+			t.Error("timeout")
+			break loop
+		}
+	}
+	for i, r := range results {
+		results[i] = r.(xsql.Row).ToMap()
+	}
+	assert.Equal(t, expResults, results)
+}
diff --git a/internal/topo/operator/preprocessor_test.go b/internal/topo/operator/preprocessor_test.go
index 73ef603a..b82747ce 100644
--- a/internal/topo/operator/preprocessor_test.go
+++ b/internal/topo/operator/preprocessor_test.go
@@ -1070,12 +1070,12 @@ func TestPreprocessorForBinary(t *testing.T) {
 		pp.streamFields = tt.stmt.StreamFields.ToJsonSchema()
 		format := message.FormatJson
 		ccc, _ := converter.GetOrCreateConverter(&ast.Options{FORMAT: format})
-		nCtx := context.WithValue(ctx, context.DecodeKey, ccc)
-		if dm, e := nCtx.Decode(tt.data); e != nil {
+		if dm, e := ccc.Decode(ctx, tt.data); e != nil {
 			log.Fatal(e)
 			return
 		} else {
-			tuple := &xsql.Tuple{Message: dm}
+			dmm := dm.(map[string]any)
+			tuple := &xsql.Tuple{Message: xsql.Message(dmm)}
 			fv, afv := xsql.NewFunctionValuersForOp(nil)
 			result := pp.Apply(ctx, tuple, fv, afv)
 			if !reflect.DeepEqual(tt.result, result) {
diff --git a/internal/topo/planner/planner_sink_test.go b/internal/topo/planner/planner_sink_test.go
index 472c6a1c..9388bd9a 100644
--- a/internal/topo/planner/planner_sink_test.go
+++ b/internal/topo/planner/planner_sink_test.go
@@ -154,6 +154,59 @@ func TestSinkPlan(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "encrypt and compress with stream writer",
+			rule: &def.Rule{
+				Actions: []map[string]any{
+					{
+						"file": map[string]any{
+							"compression": "gzip",
+							"encryption":  "aes",
+						},
+					},
+				},
+				Options: defaultOption,
+			},
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_src1"},
+				Edges: map[string][]any{
+					"source_src1": {
+						"op_file_0_0_transform",
+					},
+					"op_file_0_0_transform": {
+						"op_file_0_1_encode",
+					},
+					"op_file_0_1_encode": {
+						"sink_file_0",
+					},
+				},
+			},
+		},
+		{
+			name: "encrypt and compress with tuple collector",
+			rule: &def.Rule{
+				Actions: []map[string]any{
+					{
+						"memory": map[string]any{
+							"compression": "gzip",
+							"encryption":  "aes",
+						},
+					},
+				},
+				Options: defaultOption,
+			},
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_src1"},
+				Edges: map[string][]any{
+					"source_src1": {
+						"op_memory_0_0_transform",
+					},
+					"op_memory_0_0_transform": {
+						"sink_memory_0",
+					},
+				},
+			},
+		},
 	}
 	for _, c := range tc {
 		t.Run(c.name, func(t *testing.T) {
diff --git a/internal/converter/protobuf/test/wrapper_test.go b/pkg/model/io.go
similarity index 58%
rename from internal/converter/protobuf/test/wrapper_test.go
rename to pkg/model/io.go
index 40869821..38fc8f3a 100644
--- a/internal/converter/protobuf/test/wrapper_test.go
+++ b/pkg/model/io.go
@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// 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.
@@ -12,24 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package main
+package model
 
 import (
-	"fmt"
-	"testing"
+	"io"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 )
 
-func TestWrapper(t *testing.T) {
-	r := HelloReply{}
-	m := map[string]interface{}{"message": "hello"}
-	bytes, err := r.Encode(m)
-	if err != nil {
-		t.Errorf("encode error: %v", err)
-	}
-	fmt.Printf("bytes: %X\n", bytes)
-	mf, err := r.Decode(bytes)
-	if err != nil {
-		t.Errorf("decode error: %v", err)
-	}
-	fmt.Printf("mf: %v\n", mf)
+// Candidate for API. Currently only use internally
+
+type StreamWriter interface {
+	CreateWriter(ctx api.StreamContext, currWriter io.Writer, compression string, encryption string) (io.Writer, error)
 }