Repository: lf-edge/ekuiper
Pull request #2815 (2024-04-26 09:14:00)

feat: separate schema layer from decoder


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/converter.go b/internal/converter/converter.go
index c900a420..eb5c7f91 100644
--- a/internal/converter/converter.go
+++ b/internal/converter/converter.go
@@ -55,7 +55,7 @@ func GetOrCreateConverter(options *ast.Options) (c message.Converter, err error)
 		if options.RuleID == "" || options.StreamName == "" {
 			return json.GetConverter()
 		}
-		return json.NewFastJsonConverter(options.RuleID, options.StreamName, options.Schema, options.IsWildCard, options.IsSchemaLess), nil
+		return json.NewFastJsonConverter(options.Schema, options.IsWildCard, options.IsSchemaLess), nil
 	}
 
 	schemaFile := ""
diff --git a/internal/converter/json/converter.go b/internal/converter/json/converter.go
index e1f9ef94..36854eb3 100644
--- a/internal/converter/json/converter.go
+++ b/internal/converter/json/converter.go
@@ -22,7 +22,6 @@ 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"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
@@ -58,91 +57,24 @@ func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 type FastJsonConverter struct {
 	sync.RWMutex
 	isSchemaLess bool
-	// ruleID -> schema
-	schemaMap map[string]map[string]*ast.JsonStreamField
-	// ruleID -> StreamName
-	streamMap map[string]string
-	schema    map[string]*ast.JsonStreamField
-	// ruleID -> wildcard
-	wildcardMap map[string]struct{}
+	isWildcard   bool
+	schema       map[string]*ast.JsonStreamField
 }
 
-func NewFastJsonConverter(ruleID, streamName string, schema map[string]*ast.JsonStreamField, isWildcard, isSchemaLess bool) *FastJsonConverter {
+func NewFastJsonConverter(schema map[string]*ast.JsonStreamField, isWildcard, isSchemaLess bool) *FastJsonConverter {
 	f := &FastJsonConverter{
-		schemaMap:    make(map[string]map[string]*ast.JsonStreamField),
 		schema:       schema,
-		wildcardMap:  make(map[string]struct{}),
 		isSchemaLess: isSchemaLess,
-		streamMap:    map[string]string{},
+		isWildcard:   isWildcard,
 	}
-	f.schemaMap[ruleID] = schema
-	f.streamMap[ruleID] = streamName
-	if isWildcard {
-		f.wildcardMap[ruleID] = struct{}{}
-	}
-	f.storeSchema()
 	return f
 }
 
-func (c *FastJsonConverter) MergeSchema(ruleID, dataSource string, newSchema map[string]*ast.JsonStreamField, isWildcard bool) error {
-	c.Lock()
-	defer c.Unlock()
-	_, ok := c.schemaMap[ruleID]
-	if ok {
-		return nil
-	}
-	c.schemaMap[ruleID] = newSchema
-	c.streamMap[ruleID] = dataSource
-	if isWildcard {
-		c.wildcardMap[ruleID] = struct{}{}
-	} else {
-		mergedSchema, err := mergeSchema(c.schema, newSchema)
-		if err != nil {
-			return err
-		}
-		c.schema = mergedSchema
-	}
-	c.storeSchema()
-	return nil
-}
-
-func (c *FastJsonConverter) DetachSchema(ruleID string) error {
-	var err error
+func (c *FastJsonConverter) ResetSchema(schema map[string]*ast.JsonStreamField, isWildcard bool) {
 	c.Lock()
 	defer c.Unlock()
-	_, ok := c.schemaMap[ruleID]
-	if ok {
-		merge.RemoveRuleSchema(ruleID)
-		delete(c.streamMap, ruleID)
-		delete(c.wildcardMap, ruleID)
-		delete(c.schemaMap, ruleID)
-		newSchema := make(map[string]*ast.JsonStreamField)
-		for _, schema := range c.schemaMap {
-			newSchema, err = mergeSchema(newSchema, schema)
-			if err != nil {
-				return err
-			}
-		}
-		c.schema = newSchema
-		c.storeSchema()
-	}
-	return nil
-}
-
-func (c *FastJsonConverter) storeSchema() {
-	if len(c.wildcardMap) > 0 {
-		for ruleID := range c.schemaMap {
-			merge.AddRuleSchema(ruleID, c.streamMap[ruleID], nil, true)
-		}
-		return
-	}
-	for ruleID := range c.schemaMap {
-		merge.AddRuleSchema(ruleID, c.streamMap[ruleID], c.schema, false)
-	}
-}
-
-func mergeSchema(originSchema, newSchema map[string]*ast.JsonStreamField) (map[string]*ast.JsonStreamField, error) {
-	return merge.MergeSchema(originSchema, newSchema)
+	c.schema = schema
+	c.isWildcard = isWildcard
 }
 
 func (c *FastJsonConverter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
@@ -157,7 +89,7 @@ func (c *FastJsonConverter) Decode(ctx api.StreamContext, b []byte) (m any, err
 	}()
 	c.RLock()
 	defer c.RUnlock()
-	if len(c.wildcardMap) > 0 {
+	if c.isWildcard {
 		return converter.Decode(ctx, b)
 	}
 	return c.decodeWithSchema(b, c.schema)
diff --git a/internal/converter/schema/schema.go b/internal/converter/schema/schema.go
new file mode 100644
index 00000000..95fec429
--- /dev/null
+++ b/internal/converter/schema/schema.go
@@ -0,0 +1,119 @@
+// 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 schema
+
+import (
+	"sync"
+
+	"github.com/lf-edge/ekuiper/v2/internal/converter/merge"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+type SchemaLayer struct {
+	sync.RWMutex
+	schema map[string]*ast.JsonStreamField
+	// ruleID -> schema
+	schemaMap map[string]map[string]*ast.JsonStreamField
+	// ruleID -> StreamName
+	streamMap map[string]string
+	// ruleID -> wildcard
+	wildcardMap map[string]struct{}
+}
+
+func NewSchemaLayer(ruleID, streamName string, schema map[string]*ast.JsonStreamField, isWildcard bool) *SchemaLayer {
+	s := &SchemaLayer{
+		schemaMap:   make(map[string]map[string]*ast.JsonStreamField),
+		streamMap:   make(map[string]string),
+		wildcardMap: make(map[string]struct{}),
+	}
+	s.schemaMap[ruleID] = schema
+	s.streamMap[ruleID] = streamName
+	if isWildcard {
+		s.wildcardMap[ruleID] = struct{}{}
+	}
+	s.schema = schema
+	s.storeSchema()
+	return s
+}
+
+func (s *SchemaLayer) storeSchema() {
+	if len(s.wildcardMap) > 0 {
+		for ruleID := range s.schemaMap {
+			merge.AddRuleSchema(ruleID, s.streamMap[ruleID], nil, true)
+		}
+		return
+	}
+	for ruleID := range s.schemaMap {
+		merge.AddRuleSchema(ruleID, s.streamMap[ruleID], s.schema, false)
+	}
+}
+
+func (s *SchemaLayer) MergeSchema(ruleID, dataSource string, newSchema map[string]*ast.JsonStreamField, isWildcard bool) error {
+	s.Lock()
+	defer s.Unlock()
+	_, ok := s.schemaMap[ruleID]
+	if ok {
+		return nil
+	}
+	s.schemaMap[ruleID] = newSchema
+	s.streamMap[ruleID] = dataSource
+	if isWildcard {
+		s.wildcardMap[ruleID] = struct{}{}
+	} else {
+		mergedSchema, err := mergeSchema(s.schema, newSchema)
+		if err != nil {
+			return err
+		}
+		s.schema = mergedSchema
+	}
+	s.storeSchema()
+	return nil
+}
+
+func (s *SchemaLayer) DetachSchema(ruleID string) error {
+	var err error
+	s.Lock()
+	defer s.Unlock()
+	_, ok := s.schemaMap[ruleID]
+	if ok {
+		merge.RemoveRuleSchema(ruleID)
+		delete(s.streamMap, ruleID)
+		delete(s.wildcardMap, ruleID)
+		delete(s.schemaMap, ruleID)
+		newSchema := make(map[string]*ast.JsonStreamField)
+		for _, schema := range s.schemaMap {
+			newSchema, err = mergeSchema(newSchema, schema)
+			if err != nil {
+				return err
+			}
+		}
+		s.schema = newSchema
+		s.storeSchema()
+	}
+	return nil
+}
+
+func (s *SchemaLayer) GetSchema() (map[string]*ast.JsonStreamField, bool) {
+	s.RLock()
+	defer s.RUnlock()
+	if len(s.wildcardMap) > 0 {
+		return nil, true
+	}
+	return s.schema, false
+}
+
+func mergeSchema(originSchema, newSchema map[string]*ast.JsonStreamField) (map[string]*ast.JsonStreamField, error) {
+	return merge.MergeSchema(originSchema, newSchema)
+}
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/io/file/constant.go b/internal/io/file/constant.go
new file mode 100644
index 00000000..b1166c21
--- /dev/null
+++ b/internal/io/file/constant.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 file
+
+type FileType string
+
+const (
+	JSON_TYPE  FileType = "json"
+	CSV_TYPE   FileType = "csv"
+	LINES_TYPE FileType = "lines"
+)
+
+const (
+	GZIP = "gzip"
+	ZSTD = "zstd"
+)
+
+var fileTypes = map[FileType]struct{}{
+	JSON_TYPE:  {},
+	CSV_TYPE:   {},
+	LINES_TYPE: {},
+}
+
+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/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index 0a9b05c0..cbab54f2 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -20,6 +20,7 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/converter"
+	schemaLayer "github.com/lf-edge/ekuiper/v2/internal/converter/schema"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
@@ -30,22 +31,27 @@ import (
 type DecodeOp struct {
 	*defaultSinkNode
 	converter message.Converter
+	sLayer    *schemaLayer.SchemaLayer
 }
 
 func (o *DecodeOp) AttachSchema(ctx api.StreamContext, dataSource string, schema map[string]*ast.JsonStreamField, isWildcard bool) {
 	ctx.GetLogger().Infof("attach schema to shared stream")
-	if fastDecoder, ok := o.converter.(message.SchemaMergeAbleConverter); ok {
-		if err := fastDecoder.MergeSchema(ctx.GetRuleId(), dataSource, schema, isWildcard); err != nil {
+	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
+		if err := o.sLayer.MergeSchema(ctx.GetRuleId(), dataSource, schema, isWildcard); err != nil {
 			ctx.GetLogger().Warnf("merge schema to shared stream failed, err: %v", err)
+		} else {
+			fastDecoder.ResetSchema(o.sLayer.GetSchema())
 		}
 	}
 }
 
 func (o *DecodeOp) DetachSchema(ruleId string) {
 	conf.Log.Infof("detach schema for shared stream rule %v", ruleId)
-	if fastDecoder, ok := o.converter.(message.SchemaMergeAbleConverter); ok {
-		if err := fastDecoder.DetachSchema(ruleId); err != nil {
+	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
+		if err := o.sLayer.DetachSchema(ruleId); err != nil {
 			conf.Log.Warnf("detach schema for shared stream rule %v failed, err:%v", ruleId, err)
+		} else {
+			fastDecoder.ResetSchema(o.sLayer.GetSchema())
 		}
 	}
 }
@@ -67,6 +73,7 @@ func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, o
 	return &DecodeOp{
 		defaultSinkNode: newDefaultSinkNode(name, rOpt),
 		converter:       converterTool,
+		sLayer:          schemaLayer.NewSchemaLayer(ruleId, StreamName, schema, isWildcard),
 	}, nil
 }
 
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/pkg/message/artifacts.go b/pkg/message/artifacts.go
index a8b90e54..c602ddf6 100644
--- a/pkg/message/artifacts.go
+++ b/pkg/message/artifacts.go
@@ -40,9 +40,8 @@ type Converter interface {
 // The columns information is defined in the source side, like file source
 type ConverterProvider func(schemaFileName string, SchemaMessageName string, delimiter string) (Converter, error)
 
-type SchemaMergeAbleConverter interface {
-	MergeSchema(key, datasource string, newSchema map[string]*ast.JsonStreamField, isWildcard bool) error
-	DetachSchema(key string) error
+type SchemaResetAbleConverter interface {
+	ResetSchema(schema map[string]*ast.JsonStreamField, isWildcard bool)
 }
 
 type ColumnSetter interface {
diff --git a/pkg/model/io.go b/pkg/model/io.go
new file mode 100644
index 00000000..38fc8f3a
--- /dev/null
+++ b/pkg/model/io.go
@@ -0,0 +1,27 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package model
+
+import (
+	"io"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+)
+
+// Candidate for API. Currently only use internally
+
+type StreamWriter interface {
+	CreateWriter(ctx api.StreamContext, currWriter io.Writer, compression string, encryption string) (io.Writer, error)
+}


Test patch:
diff --git a/internal/converter/json/convert_bench_test.go b/internal/converter/json/convert_bench_test.go
index 732a1171..4735469f 100644
--- a/internal/converter/json/convert_bench_test.go
+++ b/internal/converter/json/convert_bench_test.go
@@ -107,7 +107,7 @@ func benchmarkByFiles(filePath string, b *testing.B, schema map[string]*ast.Json
 		b.Fatalf(err.Error())
 	}
 	if schema != nil {
-		f := NewFastJsonConverter("", "", schema, false, false)
+		f := NewFastJsonConverter(schema, false, false)
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
 			f.Decode(ctx, payload)
diff --git a/internal/converter/json/converter_test.go b/internal/converter/json/converter_test.go
index 62014b37..f86e1bcd 100644
--- a/internal/converter/json/converter_test.go
+++ b/internal/converter/json/converter_test.go
@@ -17,7 +17,6 @@ package json
 import (
 	"encoding/base64"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"os"
 	"path"
@@ -26,7 +25,6 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"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"
@@ -218,7 +216,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema, false, false)
 		v, err := f.Decode(ctx, tc.payload)
 		require.NoError(t, err)
 		require.Equal(t, v, tc.require)
@@ -229,7 +227,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema, false, false)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -389,7 +387,7 @@ func TestFastJsonConverterWithSchemaError(t *testing.T) {
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema, false, false)
 		_, err := f.Decode(ctx, tc.payload)
 		require.Error(t, err)
 		require.Equal(t, err.Error(), tc.err.Error())
@@ -400,7 +398,7 @@ func TestFastJsonEncode(t *testing.T) {
 	a := make(map[string]int)
 	a["a"] = 1
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter("", "", nil, false, false)
+	f := NewFastJsonConverter(nil, false, false)
 	v, err := f.Encode(ctx, a)
 	require.NoError(t, err)
 	require.Equal(t, v, []byte(`{"a":1}`))
@@ -433,7 +431,7 @@ func TestArrayWithArray(t *testing.T) {
 		},
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter("", "", schema, false, false)
+	f := NewFastJsonConverter(schema, false, false)
 	v, err := f.Decode(ctx, payload)
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -614,7 +612,7 @@ func TestTypeNull(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema, false, false)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -624,7 +622,7 @@ func TestTypeNull(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter("", "", tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema, false, false)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -641,7 +639,7 @@ func TestConvertBytea(t *testing.T) {
 		},
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter("", "", schema, false, false)
+	f := NewFastJsonConverter(schema, false, false)
 	v, err := f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -657,7 +655,7 @@ func TestConvertBytea(t *testing.T) {
 			},
 		},
 	}
-	f = NewFastJsonConverter("", "", schema, false, false)
+	f = NewFastJsonConverter(schema, false, false)
 	v, err = f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -665,213 +663,11 @@ func TestConvertBytea(t *testing.T) {
 	})
 }
 
-func TestMergeSchema(t *testing.T) {
-	testcases := []struct {
-		originSchema map[string]*ast.JsonStreamField
-		newSchema    map[string]*ast.JsonStreamField
-		resultSchema map[string]*ast.JsonStreamField
-		err          error
-	}{
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"b": nil,
-			},
-			resultSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-				"b": nil,
-			},
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-			},
-			resultSchema: map[string]*ast.JsonStreamField{
-				"a": nil,
-			},
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "array",
-					Items: &ast.JsonStreamField{
-						Type: "struct",
-						Properties: map[string]*ast.JsonStreamField{
-							"b": {
-								Type: "bigint",
-							},
-						},
-					},
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "array",
-					Items: &ast.JsonStreamField{
-						Type: "struct",
-						Properties: map[string]*ast.JsonStreamField{
-							"b": {
-								Type: "string",
-							},
-						},
-					},
-				},
-			},
-			err: errors.New("column field type b between current[bigint] and new[string] are not equal"),
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "array",
-					Items: &ast.JsonStreamField{
-						Type: "bigint",
-					},
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "array",
-					Items: &ast.JsonStreamField{
-						Type: "string",
-					},
-				},
-			},
-			err: errors.New("array column field type a between current[bigint] and new[string] are not equal"),
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"b": {
-							Type: "bigint",
-						},
-					},
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"b": {
-							Type: "string",
-						},
-					},
-				},
-			},
-			err: errors.New("column field type b between current[bigint] and new[string] are not equal"),
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "bigint",
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"b": {
-					Type: "string",
-				},
-			},
-			resultSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "bigint",
-				},
-				"b": {
-					Type: "string",
-				},
-			},
-		},
-		{
-			originSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"b": {
-							Type: "bigint",
-						},
-					},
-				},
-			},
-			newSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"c": {
-							Type: "string",
-						},
-					},
-				},
-			},
-			resultSchema: map[string]*ast.JsonStreamField{
-				"a": {
-					Type: "struct",
-					Properties: map[string]*ast.JsonStreamField{
-						"b": {
-							Type: "bigint",
-						},
-						"c": {
-							Type: "string",
-						},
-					},
-				},
-			},
-		},
-	}
-	for _, tc := range testcases {
-		f := NewFastJsonConverter("1", "", tc.originSchema, false, false)
-		err := f.MergeSchema("2", "", tc.newSchema, false)
-		if tc.err == nil {
-			require.NoError(t, err)
-			require.Equal(t, tc.resultSchema, f.schema)
-		} else {
-			require.Equal(t, tc.err, err)
-		}
-	}
-}
-
-func TestMergeWildcardSchema(t *testing.T) {
-	originSchema := map[string]*ast.JsonStreamField{
-		"a": {
-			Type: "bigint",
-		},
-	}
-	f := NewFastJsonConverter("1", "", originSchema, false, false)
-	require.NoError(t, f.MergeSchema("2", "", nil, true))
-	newSchema := map[string]*ast.JsonStreamField{
-		"b": {
-			Type: "bigint",
-		},
-	}
-	require.NoError(t, f.MergeSchema("3", "", newSchema, false))
-	data := map[string]interface{}{
-		"a": float64(1),
-		"b": float64(2),
-		"c": float64(3),
-	}
-	ctx := mockContext.NewMockContext("test", "op1")
-	bs, _ := json.Marshal(data)
-	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(ctx, bs)
-	require.NoError(t, err)
-	require.Equal(t, map[string]interface{}{
-		"a": int64(1),
-		"b": int64(2),
-	}, d)
-}
-
 func TestSchemaless(t *testing.T) {
 	originSchema := map[string]*ast.JsonStreamField{
 		"a": nil,
 	}
-	f := NewFastJsonConverter("1", "", originSchema, false, true)
+	f := NewFastJsonConverter(originSchema, false, true)
 	testcases := []struct {
 		data   map[string]interface{}
 		expect map[string]interface{}
@@ -926,44 +722,10 @@ func TestJsonError(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
 	// fastjson
-	c := NewFastJsonConverter("", "", nil, false, true)
+	c := NewFastJsonConverter(nil, false, true)
 	_, err = c.Decode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok = err.(errorx.ErrorWithCode)
 	require.True(t, ok)
 	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
 }
-
-func TestAttachDetachSchema(t *testing.T) {
-	f := NewFastJsonConverter("rule1", "demo", nil, true, true)
-	err := f.MergeSchema("rule2", "demo", map[string]*ast.JsonStreamField{
-		"a": nil,
-	}, false)
-	require.NoError(t, err)
-	r := merge.GetRuleSchema("rule1")
-	er := merge.RuleSchemaResponse{
-		Schema: map[string]map[string]*ast.JsonStreamField{
-			"demo": nil,
-		},
-		Wildcard: map[string]bool{
-			"demo": true,
-		},
-	}
-	require.Equal(t, r, er)
-	r = merge.GetRuleSchema("rule2")
-	require.Equal(t, r, er)
-	// detach rule
-	require.NoError(t, f.DetachSchema("rule1"))
-	r = merge.GetRuleSchema("rule2")
-	er = merge.RuleSchemaResponse{
-		Schema: map[string]map[string]*ast.JsonStreamField{
-			"demo": {
-				"a": nil,
-			},
-		},
-		Wildcard: map[string]bool{
-			"demo": false,
-		},
-	}
-	require.Equal(t, r, er)
-}
diff --git a/internal/converter/schema/schema_test.go b/internal/converter/schema/schema_test.go
new file mode 100644
index 00000000..bb7b736b
--- /dev/null
+++ b/internal/converter/schema/schema_test.go
@@ -0,0 +1,246 @@
+// 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 schema
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/converter/merge"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+func TestMergeSchema(t *testing.T) {
+	testcases := []struct {
+		originSchema map[string]*ast.JsonStreamField
+		newSchema    map[string]*ast.JsonStreamField
+		resultSchema map[string]*ast.JsonStreamField
+		err          error
+	}{
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"b": nil,
+			},
+			resultSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+				"b": nil,
+			},
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+			},
+			resultSchema: map[string]*ast.JsonStreamField{
+				"a": nil,
+			},
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "array",
+					Items: &ast.JsonStreamField{
+						Type: "struct",
+						Properties: map[string]*ast.JsonStreamField{
+							"b": {
+								Type: "bigint",
+							},
+						},
+					},
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "array",
+					Items: &ast.JsonStreamField{
+						Type: "struct",
+						Properties: map[string]*ast.JsonStreamField{
+							"b": {
+								Type: "string",
+							},
+						},
+					},
+				},
+			},
+			err: errors.New("column field type b between current[bigint] and new[string] are not equal"),
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "array",
+					Items: &ast.JsonStreamField{
+						Type: "bigint",
+					},
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "array",
+					Items: &ast.JsonStreamField{
+						Type: "string",
+					},
+				},
+			},
+			err: errors.New("array column field type a between current[bigint] and new[string] are not equal"),
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"b": {
+							Type: "bigint",
+						},
+					},
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"b": {
+							Type: "string",
+						},
+					},
+				},
+			},
+			err: errors.New("column field type b between current[bigint] and new[string] are not equal"),
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "bigint",
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "string",
+				},
+			},
+			resultSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "bigint",
+				},
+				"b": {
+					Type: "string",
+				},
+			},
+		},
+		{
+			originSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"b": {
+							Type: "bigint",
+						},
+					},
+				},
+			},
+			newSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"c": {
+							Type: "string",
+						},
+					},
+				},
+			},
+			resultSchema: map[string]*ast.JsonStreamField{
+				"a": {
+					Type: "struct",
+					Properties: map[string]*ast.JsonStreamField{
+						"b": {
+							Type: "bigint",
+						},
+						"c": {
+							Type: "string",
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, tc := range testcases {
+		sLayer := NewSchemaLayer("r1", "s1", tc.originSchema, false)
+		err := sLayer.MergeSchema("r2", "s1", tc.newSchema, false)
+		if tc.err == nil {
+			require.NoError(t, err)
+		} else {
+			require.Equal(t, tc.err, err)
+		}
+	}
+}
+
+func TestMergeWildcardSchema(t *testing.T) {
+	originSchema := map[string]*ast.JsonStreamField{
+		"a": {
+			Type: "bigint",
+		},
+	}
+	f := NewSchemaLayer("1", "", originSchema, false)
+	require.NoError(t, f.MergeSchema("2", "", nil, true))
+	newSchema := map[string]*ast.JsonStreamField{
+		"b": {
+			Type: "bigint",
+		},
+	}
+	require.NoError(t, f.MergeSchema("3", "", newSchema, false))
+	s, isWildcard := f.GetSchema()
+	require.Nil(t, s)
+	require.True(t, isWildcard)
+}
+
+func TestAttachDetachSchema(t *testing.T) {
+	f := NewSchemaLayer("rule1", "demo", nil, true)
+	err := f.MergeSchema("rule2", "demo", map[string]*ast.JsonStreamField{
+		"a": nil,
+	}, false)
+	require.NoError(t, err)
+	r := merge.GetRuleSchema("rule1")
+	er := merge.RuleSchemaResponse{
+		Schema: map[string]map[string]*ast.JsonStreamField{
+			"demo": nil,
+		},
+		Wildcard: map[string]bool{
+			"demo": true,
+		},
+	}
+	require.Equal(t, r, er)
+	r = merge.GetRuleSchema("rule2")
+	require.Equal(t, r, er)
+	// detach rule
+	require.NoError(t, f.DetachSchema("rule1"))
+	r = merge.GetRuleSchema("rule2")
+	er = merge.RuleSchemaResponse{
+		Schema: map[string]map[string]*ast.JsonStreamField{
+			"demo": {
+				"a": nil,
+			},
+		},
+		Wildcard: map[string]bool{
+			"demo": false,
+		},
+	}
+	require.Equal(t, r, er)
+}
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/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) {