Repository: lf-edge/ekuiper
Pull request #2870 (2024-05-23 06:57:07)

feat(source): file source refactor

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

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/io/file/reader/csv.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&filepath=internal%2Fio%2Ffile%2Freader%2Fcsv.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZmlsZS9yZWFkZXIvY3N2Lmdv) | 74.55% | [9 Missing and 5 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [pkg/mock/test\_source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&filepath=pkg%2Fmock%2Ftest_source.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-cGtnL21vY2svdGVzdF9zb3VyY2UuZ28=) | 69.23% | [4 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/file/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&filepath=internal%2Fio%2Ffile%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZmlsZS9zb3VyY2UuZ28=) | 91.43% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/node/source\_node.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Fsource_node.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL3NvdXJjZV9ub2RlLmdv) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/topo/planner/planner\_source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&filepath=internal%2Ftopo%2Fplanner%2Fplanner_source.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9wbGFubmVyL3BsYW5uZXJfc291cmNlLmdv) | 94.74% | [1 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2870?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #2870   +/-   ##
============================================
  Coverage             ?   68.80%           
============================================
  Files                ?      276           
  Lines                ?    26300           
  Branches             ?        0           
============================================
  Hits                 ?    18095           
  Misses               ?     6583           
  Partials             ?     1622           
```



</details>

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


Patch:
diff --git a/internal/io/file/reader/csv.go b/internal/io/file/reader/csv.go
new file mode 100644
index 00000000..27e0bf75
--- /dev/null
+++ b/internal/io/file/reader/csv.go
@@ -0,0 +1,108 @@
+package reader
+
+import (
+	"encoding/csv"
+	"errors"
+	"fmt"
+	"io"
+	"strconv"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func init() {
+	modules.RegisterFileStreamReader("csv", func(ctx api.StreamContext) modules.FileStreamReader {
+		return &CsvReader{}
+	})
+}
+
+type csvConf struct {
+	HasHeader bool     `json:"hasHeader"`
+	Columns   []string `json:"columns"`
+	Delimiter string   `json:"delimiter"`
+}
+
+type CsvReader struct {
+	csvR   *csv.Reader
+	config *csvConf
+
+	cols []string
+}
+
+func (r *CsvReader) Provision(ctx api.StreamContext, props map[string]any) error {
+	c := &csvConf{
+		Delimiter: ",",
+	}
+	e := cast.MapToStruct(props, c)
+	if e != nil {
+		return e
+	}
+	if c.Delimiter == "" {
+		return errors.New("invalid empty delimiter")
+	}
+	r.config = c
+	return nil
+}
+
+func (r *CsvReader) Bind(ctx api.StreamContext, fileStream io.Reader) error {
+	cr := csv.NewReader(fileStream)
+	cr.Comma = rune(r.config.Delimiter[0])
+	cr.TrimLeadingSpace = true
+	cr.FieldsPerRecord = -1
+	cols := r.config.Columns
+	if r.config.HasHeader {
+		var err error
+		ctx.GetLogger().Debug("Has header")
+		cols, err = cr.Read()
+		if err == io.EOF {
+			return fmt.Errorf("header not found")
+		}
+		if err != nil {
+			ctx.GetLogger().Warnf("Read file %s encounter error: %v", "fs.file", err)
+			return err
+		}
+		ctx.GetLogger().Debugf("Got header %v", cols)
+	}
+
+	r.csvR = cr
+	r.cols = cols
+	return nil
+}
+
+func (r *CsvReader) Read(ctx api.StreamContext) (any, error) {
+	record, err := r.csvR.Read()
+	if err == io.EOF {
+		return nil, err
+	}
+	if err != nil {
+		ctx.GetLogger().Warnf("Read file %s encounter error: %v", "fs.file", err)
+		return nil, err
+	}
+	ctx.GetLogger().Debugf("Read" + strings.Join(record, ","))
+
+	var m map[string]interface{}
+	if r.cols == nil {
+		m = make(map[string]interface{}, len(record))
+		for i, v := range record {
+			m["cols"+strconv.Itoa(i)] = v
+		}
+	} else {
+		m = make(map[string]interface{}, len(r.cols))
+		for i, v := range r.cols {
+			m[v] = record[i]
+		}
+	}
+
+	return m, nil
+}
+
+func (r *CsvReader) IsBytesReader() bool {
+	return false
+}
+
+func (r *CsvReader) Close(ctx api.StreamContext) error {
+	return nil
+}
diff --git a/internal/io/file/reader/lines.go b/internal/io/file/reader/lines.go
new file mode 100644
index 00000000..5c437cf8
--- /dev/null
+++ b/internal/io/file/reader/lines.go
@@ -0,0 +1,46 @@
+package reader
+
+import (
+	"bufio"
+	"io"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func init() {
+	modules.RegisterFileStreamReader("lines", func(ctx api.StreamContext) modules.FileStreamReader {
+		return &LinesReader{}
+	})
+}
+
+type LinesReader struct {
+	scanner *bufio.Scanner
+}
+
+func (r *LinesReader) Provision(ctx api.StreamContext, props map[string]any) error {
+	return nil
+}
+
+func (r *LinesReader) Bind(ctx api.StreamContext, fileStream io.Reader) error {
+	scanner := bufio.NewScanner(fileStream)
+	scanner.Split(bufio.ScanLines)
+	r.scanner = scanner
+	return nil
+}
+
+func (r *LinesReader) Read(ctx api.StreamContext) (any, error) {
+	succ := r.scanner.Scan()
+	if !succ {
+		return nil, io.EOF
+	}
+	return r.scanner.Bytes(), nil
+}
+
+func (r *LinesReader) IsBytesReader() bool {
+	return true
+}
+
+func (r *LinesReader) Close(ctx api.StreamContext) error {
+	return nil
+}
diff --git a/internal/io/file/source.go b/internal/io/file/source.go
index 25483b3c..1118d351 100644
--- a/internal/io/file/source.go
+++ b/internal/io/file/source.go
@@ -26,13 +26,16 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	_ "github.com/lf-edge/ekuiper/v2/internal/io/file/reader"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 type SourceConfig struct {
 	FileName         string        `json:"datasource"`
-	FileType         FileType      `json:"fileType"`
+	FileType         string        `json:"fileType"`
 	Path             string        `json:"path"`
 	Interval         time.Duration `json:"interval"`
 	IsTable          bool          `json:"isTable"`
@@ -42,28 +45,25 @@ type SourceConfig struct {
 	MoveTo           string        `json:"moveTo"`
 	IgnoreStartLines int           `json:"ignoreStartLines"`
 	IgnoreEndLines   int           `json:"ignoreEndLines"`
-	// For csv only
-	HasHeader bool     `json:"hasHeader"`
-	Columns   []string `json:"columns"`
-	Delimiter string   `json:"delimiter"`
 	// Only use for planning
 	Decompression string `json:"decompression"`
 }
 
 // Source load data from file system.
-// If file type is lines, it reads line by line.
+// Depending on file types, it may read line by line like lines, csv.
 // Otherwise, it reads the file as a whole and send to company reader node to read and split.
-// The planner need to plan according to the file type.(currently, it is hardcoded to deal with file source only)
+// The planner need to plan according to the file type.
 type Source struct {
 	file   string
 	isDir  bool
 	config *SourceConfig
+	reader modules.FileStreamReader
 	eof    api.EOFIngest
 }
 
 func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 	cfg := &SourceConfig{
-		FileType: JSON_TYPE,
+		FileType: "json",
 	}
 	err := cast.MapToStruct(props, cfg)
 	if err != nil {
@@ -72,8 +72,15 @@ func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 	if cfg.FileType == "" {
 		return errors.New("missing or invalid property fileType, must be 'json'")
 	}
-	if _, ok := fileTypes[cfg.FileType]; !ok {
-		return fmt.Errorf("invalid property fileType: %s", cfg.FileType)
+	reader, ok := modules.GetFileStreamReader(ctx, cfg.FileType)
+	if ok {
+		err = reader.Provision(ctx, props)
+		if err != nil {
+			return err
+		}
+		fs.reader = reader
+	} else {
+		ctx.GetLogger().Warnf("file type %s is not stream reader, will send out the whole file", cfg.FileType)
 	}
 	if cfg.Path == "" {
 		return errors.New("missing property Path")
@@ -125,10 +132,6 @@ func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 			}
 		}
 	}
-	if cfg.Delimiter == "" {
-		cfg.Delimiter = ","
-	}
-
 	fs.config = cfg
 	return nil
 }
@@ -138,10 +141,10 @@ func (fs *Source) Connect(ctx api.StreamContext) error {
 	return nil
 }
 
-// Subscribe file source always ingest bytes
+// Subscribe file source may ingest bytes or tuple
 // For stream source, it ingest one line
 // For batch source, it ingest the whole file, thus it need a reader node to coordinate and read the content into lines/array
-func (fs *Source) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
+func (fs *Source) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest) error {
 	fs.Load(ctx, ingest, ingestError)
 	if fs.config.Interval > 0 {
 		ticker := time.NewTicker(fs.config.Interval)
@@ -175,7 +178,7 @@ func (fs *Source) Close(ctx api.StreamContext) error {
 	return nil
 }
 
-func (fs *Source) Load(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) {
+func (fs *Source) Load(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest) {
 	if fs.isDir {
 		ctx.GetLogger().Debugf("Monitor dir %s", fs.file)
 		entries, err := os.ReadDir(fs.file)
@@ -220,7 +223,7 @@ func (fs *Source) Load(ctx api.StreamContext, ingest api.BytesIngest, ingestErro
 	ctx.GetLogger().Debug("All tuples sent")
 }
 
-func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.BytesIngest, ingestError api.ErrorIngest) {
+func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.TupleIngest, ingestError api.ErrorIngest) {
 	var (
 		err error
 		r   io.Reader
@@ -241,25 +244,25 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Bytes
 	}
 	meta := map[string]any{"file": file}
 	// Read line or read all
-	switch fs.config.FileType {
-	case LINES_TYPE: // read line by line
-		scanner := bufio.NewScanner(r)
-		scanner.Split(bufio.ScanLines)
-		for scanner.Scan() {
+	if fs.reader != nil {
+		err = fs.reader.Bind(ctx, r)
+		if err != nil {
+			ingestError(ctx, err)
+			return
+		}
+		for {
+			line, err := fs.reader.Read(ctx)
+			if err != nil {
+				break
+			}
 			rcvTime := timex.GetNow()
-			fmt.Println("received", rcvTime.UnixMilli())
-			line := scanner.Bytes()
 			ingest(ctx, line, meta, rcvTime)
 			if fs.config.SendInterval > 0 {
 				time.Sleep(fs.config.SendInterval)
 			}
 		}
-		// Check for any errors during scanning
-		if err := scanner.Err(); err != nil {
-			ingestError(ctx, err)
-			return
-		}
-	default: // read all
+		_ = fs.reader.Close(ctx)
+	} else {
 		rcvTime := timex.GetNow()
 		content, err := os.ReadFile(file)
 		if err != nil {
@@ -270,6 +273,7 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Bytes
 			ingest(ctx, content, meta, rcvTime)
 		}
 	}
+
 	ctx.GetLogger().Debugf("Finish loading from file %s", file)
 	switch fs.config.ActionAfterRead {
 	case 1:
@@ -337,11 +341,28 @@ func ignoreLines(ctx api.StreamContext, reader io.Reader, ignoreStartLines int,
 	return r
 }
 
+func (fs *Source) Info() (i model.NodeInfo) {
+	if fs.reader == nil { // output batch raw, so need encrypt/decompress as a whole and then decode as a whole
+		i.NeedBatchDecode = true
+		i.NeedDecode = true
+	} else if fs.reader.IsBytesReader() { // decrypt/decompress in scan and output raw
+		i.NeedDecode = true
+		i.HasCompress = true
+		i.HasInterval = true
+	} else { // decrypt/decompress in scan and output decoded tuple
+		i.HasCompress = true
+		i.HasInterval = true
+	}
+	return
+}
+
 func GetSource() api.Source {
 	return &Source{}
 }
 
 var (
-	_ api.BytesSource = &Source{}
+	// ingest possibly []byte and tuple
+	_ api.TupleSource = &Source{}
 	_ api.Bounded     = &Source{}
+	_ model.InfoNode  = &Source{}
 )
diff --git a/internal/io/mqtt/connection.go b/internal/io/mqtt/connection.go
index cfd12bf0..95132e1f 100644
--- a/internal/io/mqtt/connection.go
+++ b/internal/io/mqtt/connection.go
@@ -58,7 +58,7 @@ type SubscriptionInfo struct {
 }
 
 func (conn *Connection) Publish(topic string, qos byte, retained bool, payload any) error {
-	if !conn.connected.Load() {
+	if conn == nil || !conn.connected.Load() {
 		return errorx.NewIOErr("mqtt client is not connected")
 	}
 	token := conn.Client.Publish(topic, qos, retained, payload)
diff --git a/internal/topo/node/concurrent.go b/internal/topo/node/concurrent.go
index b9cdfec3..f15131ba 100644
--- a/internal/topo/node/concurrent.go
+++ b/internal/topo/node/concurrent.go
@@ -15,6 +15,8 @@
 package node
 
 import (
+	"time"
+
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 )
@@ -25,6 +27,10 @@ import (
 type workerFunc func(ctx api.StreamContext, item any) []any
 
 func runWithOrder(ctx api.StreamContext, node *defaultSinkNode, numWorkers int, wf workerFunc) {
+	runWithOrderAndInterval(ctx, node, numWorkers, wf, 0)
+}
+
+func runWithOrderAndInterval(ctx api.StreamContext, node *defaultSinkNode, numWorkers int, wf workerFunc, sendInterval time.Duration) {
 	workerChans := make([]chan any, numWorkers)
 	workerOutChans := make([]chan []any, numWorkers)
 	for i := range workerChans {
@@ -38,14 +44,14 @@ func runWithOrder(ctx api.StreamContext, node *defaultSinkNode, numWorkers int,
 	}
 	// start merger goroutine
 	output := make(chan any)
-	go merge(ctx, node, output, workerOutChans...)
+	go merge(ctx, node, sendInterval, output, workerOutChans...)
 
 	// Distribute input data to workers
 	distribute(ctx, node, numWorkers, workerChans)
 }
 
 // Merge multiple channels into one preserving the order
-func merge(ctx api.StreamContext, node *defaultSinkNode, output chan any, channels ...chan []any) {
+func merge(ctx api.StreamContext, node *defaultSinkNode, sendInterval time.Duration, output chan any, channels ...chan []any) {
 	defer close(output)
 	// Start a goroutine for each input channel
 	for {
@@ -64,6 +70,9 @@ func merge(ctx api.StreamContext, node *defaultSinkNode, output chan any, channe
 					default:
 						node.statManager.IncTotalRecordsOut()
 					}
+					if sendInterval > 0 {
+						time.Sleep(sendInterval)
+					}
 				}
 				node.statManager.IncTotalMessagesProcessed(1)
 			case <-ctx.Done():
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index 8d663085..0b3944b9 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -16,6 +16,7 @@ package node
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/converter"
@@ -23,14 +24,23 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
 )
 
+// DecodeOp manages the format decoding (employ schema) and sending frequency (for batch decode, like a json array)
 type DecodeOp struct {
 	*defaultSinkNode
 	converter message.Converter
 	sLayer    *schemaLayer.SchemaLayer
+	// When receiving list, send them one by one, this is the sending interval between each
+	// Typically set by file source
+	sendInterval time.Duration
+}
+
+type dconf struct {
+	SendInterval time.Duration `json:"sendInterval"`
 }
 
 func (o *DecodeOp) AttachSchema(ctx api.StreamContext, dataSource string, schema map[string]*ast.JsonStreamField, isWildcard bool) {
@@ -57,7 +67,7 @@ func (o *DecodeOp) DetachSchema(ctx api.StreamContext, ruleId string) {
 	}
 }
 
-func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, options *ast.Options, isWildcard, isSchemaless bool, schema map[string]*ast.JsonStreamField) (*DecodeOp, error) {
+func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, options *ast.Options, isWildcard, isSchemaless bool, schema map[string]*ast.JsonStreamField, props map[string]any) (*DecodeOp, error) {
 	options.Schema = nil
 	options.IsWildCard = isWildcard
 	options.IsSchemaLess = isSchemaless
@@ -71,10 +81,16 @@ func NewDecodeOp(name, StreamName string, ruleId string, rOpt *def.RuleOption, o
 		msg := fmt.Sprintf("cannot get converter from format %s, schemaId %s: %v", options.FORMAT, options.SCHEMAID, err)
 		return nil, fmt.Errorf(msg)
 	}
+	dc := &dconf{}
+	e := cast.MapToStruct(props, dc)
+	if e != nil {
+		return nil, e
+	}
 	return &DecodeOp{
 		defaultSinkNode: newDefaultSinkNode(name, rOpt),
 		converter:       converterTool,
 		sLayer:          schemaLayer.NewSchemaLayer(ruleId, StreamName, schema, isWildcard),
+		sendInterval:    dc.SendInterval,
 	}, nil
 }
 
@@ -86,7 +102,7 @@ func (o *DecodeOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 			o.Close()
 		}()
 		err := infra.SafeRun(func() error {
-			runWithOrder(ctx, o.defaultSinkNode, o.concurrency, o.Worker)
+			runWithOrderAndInterval(ctx, o.defaultSinkNode, o.concurrency, o.Worker, o.sendInterval)
 			return nil
 		})
 		if err != nil {
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index fb3fe4bc..6b67f3fd 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -87,6 +87,10 @@ func (m *SourceNode) ingestAnyTuple(ctx api.StreamContext, data any, meta map[st
 		for _, mm := range mess {
 			m.ingestMap(mm, meta, ts)
 		}
+	// expected from file which send out any tuple type
+	case []byte:
+		tuple := &xsql.RawTuple{Emitter: m.name, Rawdata: mess, Timestamp: ts, Metadata: meta}
+		m.Broadcast(tuple)
 	// Source tuples are expected from memory
 	case *xsql.Tuple:
 		m.ingestTuple(mess, ts)
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index b8f3becc..798722f2 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -26,6 +26,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/topo/operator"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
 )
 
 func transformSourceNode(ctx api.StreamContext, t *DataSourcePlan, mockSourcesProp map[string]map[string]any, ruleId string, options *def.RuleOption, index int) (node.DataSourceNode, []node.OperatorNode, int, error) {
@@ -100,8 +101,14 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 		}
 		srcConnNode = srcSubtopo
 	}
-
-	_, isBytesSource := ss.(api.BytesSource)
+	// Make sure it is provisioned. Already done in NewSourceNode
+	info := checkByteSource(ss)
+	if info.HasCompress {
+		sp.Decompression = ""
+	}
+	if info.HasInterval {
+		delete(props, "sendInterval")
+	}
 
 	if err != nil {
 		return nil, nil, 0, err
@@ -110,7 +117,7 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 	var ops []node.OperatorNode
 
 	if sp.Decompression != "" {
-		if isBytesSource {
+		if info.NeedBatchDecode {
 			dco, err := node.NewDecompressOp(fmt.Sprintf("%d_decompress", index), options, sp.Decompression)
 			if err != nil {
 				return nil, nil, 0, err
@@ -118,13 +125,13 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 			index++
 			ops = append(ops, dco)
 		} else {
-			return nil, nil, 0, fmt.Errorf("source %s does not support decompression", t.name)
+			ctx.GetLogger().Warnf("source %s does not support decompression", t.name)
 		}
 	}
 
-	if isBytesSource {
+	if info.NeedDecode {
 		// Create the decode node
-		decodeNode, err := node.NewDecodeOp(fmt.Sprintf("%d_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields)
+		decodeNode, err := node.NewDecodeOp(fmt.Sprintf("%d_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
 		if err != nil {
 			return nil, nil, 0, err
 		}
@@ -155,3 +162,17 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 	}
 	return srcConnNode, ops, 0, nil
 }
+
+func checkByteSource(ss api.Source) model.NodeInfo {
+	switch st := ss.(type) {
+	case model.InfoNode:
+		return st.Info()
+	case api.BytesSource:
+		return model.NodeInfo{
+			NeedDecode:      true,
+			NeedBatchDecode: true,
+		}
+	default:
+		return model.NodeInfo{}
+	}
+}
diff --git a/pkg/model/io.go b/pkg/model/io.go
index 38fc8f3a..fc23b255 100644
--- a/pkg/model/io.go
+++ b/pkg/model/io.go
@@ -25,3 +25,19 @@ import (
 type StreamWriter interface {
 	CreateWriter(ctx api.StreamContext, currWriter io.Writer, compression string, encryption string) (io.Writer, error)
 }
+
+type StreamReader interface {
+	CreateWriter(ctx api.StreamContext, currWriter io.Writer, compression string, encryption string) (io.Writer, error)
+}
+
+// InfoNode explain the node itself. Mainly used for planner to decide the split of source/sink
+type InfoNode interface {
+	Info() NodeInfo
+}
+
+type NodeInfo struct {
+	NeedDecode      bool
+	NeedBatchDecode bool // like decrypt, decompress as a whole
+	HasCompress     bool
+	HasInterval     bool
+}
diff --git a/pkg/modules/file.go b/pkg/modules/file.go
new file mode 100644
index 00000000..d8623c0b
--- /dev/null
+++ b/pkg/modules/file.go
@@ -0,0 +1,71 @@
+// 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 modules
+
+import (
+	"io"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+)
+
+type RollHook interface {
+	Provision(ctx api.StreamContext, props map[string]any) error
+	RollDone(ctx api.StreamContext, filePath string) error
+	api.Closable
+}
+
+type RollHookProvider func() RollHook
+
+var fileRollHooks = map[string]RollHookProvider{}
+
+func RegisterFileRollHook(name string, provider RollHookProvider) {
+	fileRollHooks[name] = provider
+}
+
+func GetFileRollHook(name string) (RollHook, bool) {
+	if p, ok := fileRollHooks[name]; ok {
+		return p(), ok
+	}
+	return nil, false
+}
+
+// FileStreamReader reads a type of file line by line. Avoid to load the full file
+// If need to load full file, just extend converter to decode the full bytes
+type FileStreamReader interface {
+	// Provision Set up the static properties
+	Provision(ctx api.StreamContext, props map[string]any) error
+	// Bind set the file stream. Make sure the previous read has done
+	Bind(ctx api.StreamContext, fileStream io.Reader) error
+	// Read the next record. Returns EOF when the input has reached its end.
+	Read(ctx api.StreamContext) (any, error)
+	// IsBytesReader If is bytes reader, Read must return []byte, otherwise return map or []map
+	IsBytesReader() bool
+	api.Closable
+}
+
+type FileStreamReaderProvider func(ctx api.StreamContext) FileStreamReader
+
+var fileStreamReaders = map[string]FileStreamReaderProvider{}
+
+func RegisterFileStreamReader(name string, provider FileStreamReaderProvider) {
+	fileStreamReaders[name] = provider
+}
+
+func GetFileStreamReader(ctx api.StreamContext, name string) (FileStreamReader, bool) {
+	if p, ok := fileStreamReaders[name]; ok {
+		return p(ctx), true
+	}
+	return nil, false
+}
diff --git a/pkg/modules/file_roll_hook.go b/pkg/modules/file_roll_hook.go
deleted file mode 100644
index 1546a71f..00000000
--- a/pkg/modules/file_roll_hook.go
+++ /dev/null
@@ -1,38 +0,0 @@
-// 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 modules
-
-import "github.com/lf-edge/ekuiper/contract/v2/api"
-
-type RollHook interface {
-	Provision(ctx api.StreamContext, props map[string]any) error
-	RollDone(ctx api.StreamContext, filePath string) error
-	api.Closable
-}
-
-type RollHookProvider func() RollHook
-
-var fileRollHooks = map[string]RollHookProvider{}
-
-func RegisterFileRollHook(name string, provider RollHookProvider) {
-	fileRollHooks[name] = provider
-}
-
-func GetFileRollHook(name string) (RollHook, bool) {
-	if p, ok := fileRollHooks[name]; ok {
-		return p(), ok
-	}
-	return nil, false
-}


Test patch:
diff --git a/internal/io/file/source_test.go b/internal/io/file/source_test.go
index 76402b3e..d09970e2 100644
--- a/internal/io/file/source_test.go
+++ b/internal/io/file/source_test.go
@@ -67,10 +67,9 @@ func TestSourceProvision(t *testing.T) {
 				"path":       path,
 			},
 			c: &SourceConfig{
-				FileName:  name,
-				Path:      path,
-				FileType:  JSON_TYPE,
-				Delimiter: ",",
+				FileName: name,
+				Path:     path,
+				FileType: string(JSON_TYPE),
 			},
 		},
 		{
@@ -92,13 +91,14 @@ func TestSourceProvision(t *testing.T) {
 			e: "missing or invalid property fileType, must be 'json'",
 		},
 		{
-			name: "invalid file type",
+			name: "invalid reader prop",
 			props: map[string]any{
 				"datasource": name,
 				"path":       path,
-				"fileType":   "aaa",
+				"fileType":   "csv",
+				"hasHeader":  "uvw",
 			},
-			e: "invalid property fileType: aaa",
+			e: "1 error(s) decoding:\n\n* 'hasHeader' expected type 'bool', got unconvertible type 'string', value: 'uvw'",
 		},
 		{
 			name: "missing path",
@@ -134,8 +134,7 @@ func TestSourceProvision(t *testing.T) {
 			c: &SourceConfig{
 				FileName:         "",
 				Path:             relPath,
-				FileType:         JSON_TYPE,
-				Delimiter:        ",",
+				FileType:         string(JSON_TYPE),
 				IgnoreStartLines: 0,
 				IgnoreEndLines:   0,
 			},
@@ -179,8 +178,7 @@ func TestSourceProvision(t *testing.T) {
 			c: &SourceConfig{
 				FileName:        name,
 				Path:            path,
-				FileType:        JSON_TYPE,
-				Delimiter:       ",",
+				FileType:        string(JSON_TYPE),
 				ActionAfterRead: 2,
 				MoveTo:          filepath.Join(path, "ddd"),
 			},
@@ -227,7 +225,7 @@ func TestLines(t *testing.T) {
 		model.NewDefaultRawTuple([]byte("{\"id\": 3,\"name\": \"John Smith\"}"), meta, mc.Now()),
 		model.NewDefaultRawTuple([]byte("[{\"id\": 4,\"name\": \"John Smith\"},{\"id\": 5,\"name\": \"John Smith\"}]"), meta, mc.Now()),
 	}
-	r := GetSource().(api.BytesSource)
+	r := GetSource().(api.TupleSource)
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":            path,
 		"fileType":        "lines",
@@ -238,6 +236,40 @@ func TestLines(t *testing.T) {
 	})
 }
 
+func TestCSVBatch(t *testing.T) {
+	path, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	path = filepath.Join(path, "test")
+
+	meta1 := map[string]any{
+		"file": filepath.Join(path, "csv", "a.csv"),
+	}
+	meta2 := map[string]any{
+		"file": filepath.Join(path, "csv", "b.csv"),
+	}
+	mc := timex.Clock
+	exp := []api.MessageTuple{
+		model.NewDefaultSourceTuple(map[string]any{"@": "#", "id": "1", "ts": "1670170500", "value": "161.927872"}, meta1, mc.Now()),
+		model.NewDefaultSourceTuple(map[string]any{"@": "#", "id": "2", "ts": "1670170900", "value": "176"}, meta1, mc.Now()),
+		model.NewDefaultSourceTuple(map[string]any{"id": "33", "ts": "1670270500", "humidity": "89"}, meta2, mc.Now()),
+		model.NewDefaultSourceTuple(map[string]any{"id": "44", "ts": "1670270900", "humidity": "76"}, meta2, mc.Now()),
+	}
+	r := GetSource().(api.TupleSource)
+	mock.TestSourceConnector(t, r, map[string]any{
+		"path":             path,
+		"fileType":         "csv",
+		"datasource":       "csv",
+		"hasHeader":        true,
+		"delimiter":        "\t",
+		"ignoreStartLines": 3,
+		"ignoreEndLines":   1,
+	}, exp, func() {
+		// do nothing
+	})
+}
+
 func TestBatch(t *testing.T) {
 	path, err := os.Getwd()
 	if err != nil {
@@ -252,7 +284,7 @@ func TestBatch(t *testing.T) {
 	exp := []api.MessageTuple{
 		model.NewDefaultRawTuple([]byte("[{\"id\": 1,\"name\": \"John Doe\",\"height\": 1.82},{\"id\": 2,\"name\": \"Jane Doe\",\"height\": 1.65}]"), meta, mc.Now()),
 	}
-	r := GetSource().(api.BytesSource)
+	r := GetSource().(api.TupleSource)
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":       path,
 		"fileType":   "json",
@@ -281,7 +313,7 @@ func TestIgnoreLines(t *testing.T) {
 		model.NewDefaultRawTuple([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta, mc.Now()),
 		model.NewDefaultRawTuple([]byte("{\"id\": 3,\"name\": \"John Smith\"}"), meta, mc.Now()),
 	}
-	r := GetSource().(api.BytesSource)
+	r := GetSource().(api.TupleSource)
 	r.(api.Bounded).SetEofIngest(func(ctx api.StreamContext) {
 		fmt.Printf("eof")
 	})
@@ -328,7 +360,7 @@ func TestIntervalAndDir(t *testing.T) {
 		model.NewDefaultRawTuple([]byte("[{\"id\": 4,\"name\": \"John Smith\"},{\"id\": 5,\"name\": \"John Smith\"}]"), meta, ts),
 		model.NewDefaultRawTuple([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta, ts),
 	}
-	r := GetSource().(api.BytesSource)
+	r := GetSource().(api.TupleSource)
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":             path,
 		"fileType":         "lines",
diff --git a/internal/io/file/test/csv/a.csv b/internal/io/file/test/csv/a.csv
new file mode 100644
index 00000000..16d0da3a
--- /dev/null
+++ b/internal/io/file/test/csv/a.csv
@@ -0,0 +1,7 @@
+//	some irrelevant content
+<!	Version=1.0	Code=UTF-8	!>
+<special content>
+@	id	ts	value
+#	1	1670170500	161.927872
+#	2	1670170900	176
+<special content>
diff --git a/internal/io/file/test/csv/b.csv b/internal/io/file/test/csv/b.csv
new file mode 100644
index 00000000..de354274
--- /dev/null
+++ b/internal/io/file/test/csv/b.csv
@@ -0,0 +1,7 @@
+//	some irrelevant content
+<!	Version=1.0	Code=UTF-8	!>
+<special content>
+id	ts	humidity
+33	1670270500	89
+44	1670270900	76
+<special content>
diff --git a/internal/topo/node/decode_op_test.go b/internal/topo/node/decode_op_test.go
index e8dcf5c5..2480b134 100644
--- a/internal/topo/node/decode_op_test.go
+++ b/internal/topo/node/decode_op_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 func TestJSON(t *testing.T) {
@@ -38,7 +39,9 @@ func TestJSON(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json"}, false, true, nil)
+			op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{
+				"sendInterval": "10ms",
+			})
 			assert.NoError(t, err)
 			out := make(chan any, 100)
 			err = op.AddOutput(out, "test")
@@ -68,7 +71,7 @@ func TestJSON(t *testing.T) {
 				{errors.New("unsupported decode result: hello")},
 				{errors.New("unsupported data received: invalid")},
 			}
-
+			timex.Add(2 * time.Second)
 			for i, c := range cases {
 				op.input <- c
 				for _, e := range expects[i] {
@@ -89,7 +92,7 @@ func TestJSON(t *testing.T) {
 // Concurrency 10 - BenchmarkThrougput-16           1000000000               0.1553 ns/op
 // This is useful when a node is much slower
 func BenchmarkThrougput(b *testing.B) {
-	op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: 10, Debug: true}, &ast.Options{FORMAT: "mock"}, false, true, nil)
+	op, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: 10, Debug: true}, &ast.Options{FORMAT: "mock"}, false, true, nil, nil)
 	assert.NoError(b, err)
 	out := make(chan any, 100)
 	err = op.AddOutput(out, "test")
@@ -130,7 +133,7 @@ func TestJSONWithSchema(t *testing.T) {
 				"a": {
 					Type: "bigint",
 				},
-			})
+			}, nil)
 			assert.NoError(t, err)
 			out := make(chan any, 100)
 			err = op.AddOutput(out, "test")
@@ -220,7 +223,13 @@ func TestJSONWithSchema(t *testing.T) {
 }
 
 func TestValidate(t *testing.T) {
-	_, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "cann"}, false, true, nil)
+	_, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "cann"}, false, true, nil, nil)
 	assert.Error(t, err)
 	assert.Equal(t, "cannot get converter from format cann, schemaId : format type cann not supported", err.Error())
+	_, err = NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "none"})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "1 error(s) decoding:\n\n* error decoding 'sendInterval': time: invalid duration \"none\"")
+	do, err := NewDecodeOp("test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "12s"})
+	assert.NoError(t, err)
+	assert.Equal(t, 12*time.Second, do.sendInterval)
 }
diff --git a/internal/topo/planner/planner_source_test.go b/internal/topo/planner/planner_source_test.go
new file mode 100644
index 00000000..a395006b
--- /dev/null
+++ b/internal/topo/planner/planner_source_test.go
@@ -0,0 +1,292 @@
+// 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 (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/meta"
+	"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/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+)
+
+func TestPlanTopo(t *testing.T) {
+	kv, err := store.GetKV("stream")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	streamSqls := map[string]string{
+		"src1":     `CREATE STREAM src1 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt");`,
+		"src2":     `CREATE STREAM src2 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt", SHARED="true");`,
+		"src3":     `CREATE STREAM src3 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSel");`,
+		"src4":     `CREATE STREAM src4 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSel",SHARED="true");`,
+		"filesrc1": `CREATE STREAM fs1 () WITH (FORMAT="json", TYPE="file",CONF_KEY="lines");`,
+		"filesrc2": `CREATE STREAM fs2 () WITH (FORMAT="delimited", TYPE="file",CONF_KEY="csv");`,
+		"filesrc3": `CREATE STREAM fs3 () WITH (FORMAT="json",TYPE="file",CONF_KEY="json");`,
+	}
+	for name, sql := range streamSqls {
+		s, err := json.Marshal(&xsql.StreamInfo{
+			StreamType: ast.TypeStream,
+			Statement:  sql,
+		})
+		assert.NoError(t, err)
+		err = kv.Set(name, string(s))
+		assert.NoError(t, err)
+	}
+	confs := []struct {
+		conf map[string]any
+		p    string
+		k    string
+	}{
+		{
+			conf: map[string]any{
+				"connectionSelector": "mqtt.localConnection",
+			},
+			p: "mqtt",
+			k: "testSel",
+		},
+		{
+			conf: map[string]any{
+				"fileType":      "csv",
+				"decompression": "gzip",
+			},
+			p: "file",
+			k: "csv",
+		},
+		{
+			conf: map[string]any{
+				"fileType":      "lines",
+				"decompression": "gzip",
+			},
+			p: "file",
+			k: "lines",
+		},
+		{
+			conf: map[string]any{
+				"decompression": "gzip",
+			},
+			p: "file",
+			k: "json",
+		},
+	}
+	meta.InitYamlConfigManager()
+	dataDir, _ := conf.GetDataLoc()
+	err = os.MkdirAll(filepath.Join(dataDir, "sources"), 0o755)
+	assert.NoError(t, err)
+
+	for _, cc := range confs {
+		p, k := cc.p, cc.k
+		bs, err := json.Marshal(cc.conf)
+		assert.NoError(t, err)
+		err = meta.AddSourceConfKey(p, k, "", bs)
+		assert.NoError(t, err)
+		// intended to run at last
+		defer func() {
+			err = meta.DelSourceConfKey(p, k, "")
+			assert.NoError(t, err)
+		}()
+	}
+
+	tests := []struct {
+		name string
+		sql  string
+		topo *def.PrintableTopo
+	}{
+		{
+			name: "testMqttSplit",
+			sql:  `SELECT * FROM src1`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_src1"},
+				Edges: map[string][]any{
+					"source_src1": {
+						"op_2_decoder",
+					},
+					"op_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "testSharedMqttSplit",
+			sql:  `SELECT * FROM src2`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_src2"},
+				Edges: map[string][]any{
+					"source_src2": {
+						"op_src2_2_decoder",
+					},
+					"op_src2_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "testSharedConnSplit",
+			sql:  `SELECT * FROM src3`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_mqtt.localConnection/topic1"},
+				Edges: map[string][]any{
+					"source_mqtt.localConnection/topic1": {
+						"op_2_decoder",
+					},
+					"op_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "testSharedNodeWithSharedConnSplit",
+			sql:  `SELECT * FROM src4`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_mqtt.localConnection/topic1"},
+				Edges: map[string][]any{
+					"source_mqtt.localConnection/topic1": {
+						"op_src4_2_decoder",
+					},
+					"op_src4_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "test lines file",
+			sql:  `SELECT * FROM filesrc1`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_fs1"},
+				Edges: map[string][]any{
+					"source_fs1": {
+						"op_2_decoder",
+					},
+					"op_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "test csv file",
+			sql:  `SELECT * FROM filesrc2`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_fs2"},
+				Edges: map[string][]any{
+					"source_fs2": {
+						"op_2_project",
+					},
+					"op_2_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+		{
+			name: "test json file",
+			sql:  `SELECT * FROM filesrc3`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_fs3"},
+				Edges: map[string][]any{
+					"source_fs3": {
+						"op_2_decompress",
+					},
+					"op_2_decompress": {
+						"op_3_decoder",
+					},
+					"op_3_decoder": {
+						"op_4_project",
+					},
+					"op_4_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tp, err := PlanSQLWithSourcesAndSinks(def.GetDefaultRule(tt.name, tt.sql), nil)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.topo, tp.GetTopo())
+		})
+	}
+}
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index fe9a927d..6ead3104 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -4785,11 +4785,11 @@ func TestTransformSourceNode(t *testing.T) {
 	props := nodeConf.GetSourceConf("mqtt", &ast.Options{TYPE: "mqtt", DATASOURCE: "topic1"})
 	srcNode, err := node.NewSourceNode(tp.GetContext(), "test", &mqtt.SourceConnector{}, props, &def.RuleOption{SendError: false})
 	assert.NoError(t, err)
-	decodeNode, err := node.NewDecodeOp("2_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema)
+	decodeNode, err := node.NewDecodeOp("2_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
 	assert.NoError(t, err)
 	decomNode, err := node.NewDecompressOp("2_decompressor", &def.RuleOption{SendError: false}, "gzip")
 	assert.NoError(t, err)
-	decodeNode2, err := node.NewDecodeOp("3_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema)
+	decodeNode2, err := node.NewDecodeOp("3_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
 	assert.NoError(t, err)
 	props2 := nodeConf.GetSourceConf("mqtt", &ast.Options{TYPE: "mqtt", CONF_KEY: "testCom", DATASOURCE: "topic2"})
 	srcNode2, err := node.NewSourceNode(tp.GetContext(), "test", &mqtt.SourceConnector{}, props2, &def.RuleOption{SendError: false})
@@ -4967,151 +4967,3 @@ func TestGetLogicalPlanForExplain(t *testing.T) {
 		}
 	}
 }
-
-func TestPlanTopo(t *testing.T) {
-	kv, err := store.GetKV("stream")
-	if err != nil {
-		t.Error(err)
-		return
-	}
-	streamSqls := map[string]string{
-		"src1": `CREATE STREAM src1 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt");`,
-		"src2": `CREATE STREAM src2 () WITH (DATASOURCE="src1", FORMAT="json", TYPE="mqtt", SHARED="true");`,
-		"src3": `CREATE STREAM src3 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSel");`,
-		"src4": `CREATE STREAM src4 () WITH (DATASOURCE="topic1", FORMAT="json", TYPE="mqtt", CONF_KEY="testSel",SHARED="true");`,
-	}
-	for name, sql := range streamSqls {
-		s, err := json.Marshal(&xsql.StreamInfo{
-			StreamType: ast.TypeStream,
-			Statement:  sql,
-		})
-		assert.NoError(t, err)
-		err = kv.Set(name, string(s))
-		assert.NoError(t, err)
-	}
-	// add connectionSelector for meta
-	a1 := map[string]interface{}{
-		"connectionSelector": "mqtt.localConnection",
-	}
-	bs, err := json.Marshal(a1)
-	assert.NoError(t, err)
-	meta.InitYamlConfigManager()
-	dataDir, _ := conf.GetDataLoc()
-	err = os.MkdirAll(filepath.Join(dataDir, "sources"), 0o755)
-	assert.NoError(t, err)
-	err = meta.AddSourceConfKey("mqtt", "testSel", "", bs)
-	assert.NoError(t, err)
-	defer func() {
-		err = meta.DelSourceConfKey("mqtt", "testSel", "")
-		assert.NoError(t, err)
-	}()
-	tests := []struct {
-		name string
-		sql  string
-		topo *def.PrintableTopo
-	}{
-		{
-			name: "testMqttSplit",
-			sql:  `SELECT * FROM src1`,
-			topo: &def.PrintableTopo{
-				Sources: []string{"source_src1"},
-				Edges: map[string][]any{
-					"source_src1": {
-						"op_2_decoder",
-					},
-					"op_2_decoder": {
-						"op_3_project",
-					},
-					"op_3_project": {
-						"op_logToMemory_0_0_transform",
-					},
-					"op_logToMemory_0_0_transform": {
-						"op_logToMemory_0_1_encode",
-					},
-					"op_logToMemory_0_1_encode": {
-						"sink_logToMemory_0",
-					},
-				},
-			},
-		},
-		{
-			name: "testSharedMqttSplit",
-			sql:  `SELECT * FROM src2`,
-			topo: &def.PrintableTopo{
-				Sources: []string{"source_src2"},
-				Edges: map[string][]any{
-					"source_src2": {
-						"op_src2_2_decoder",
-					},
-					"op_src2_2_decoder": {
-						"op_3_project",
-					},
-					"op_3_project": {
-						"op_logToMemory_0_0_transform",
-					},
-					"op_logToMemory_0_0_transform": {
-						"op_logToMemory_0_1_encode",
-					},
-					"op_logToMemory_0_1_encode": {
-						"sink_logToMemory_0",
-					},
-				},
-			},
-		},
-		{
-			name: "testSharedConnSplit",
-			sql:  `SELECT * FROM src3`,
-			topo: &def.PrintableTopo{
-				Sources: []string{"source_mqtt.localConnection/topic1"},
-				Edges: map[string][]any{
-					"source_mqtt.localConnection/topic1": {
-						"op_2_decoder",
-					},
-					"op_2_decoder": {
-						"op_3_project",
-					},
-					"op_3_project": {
-						"op_logToMemory_0_0_transform",
-					},
-					"op_logToMemory_0_0_transform": {
-						"op_logToMemory_0_1_encode",
-					},
-					"op_logToMemory_0_1_encode": {
-						"sink_logToMemory_0",
-					},
-				},
-			},
-		},
-		{
-			name: "testSharedNodeWithSharedConnSplit",
-			sql:  `SELECT * FROM src4`,
-			topo: &def.PrintableTopo{
-				Sources: []string{"source_mqtt.localConnection/topic1"},
-				Edges: map[string][]any{
-					"source_mqtt.localConnection/topic1": {
-						"op_src4_2_decoder",
-					},
-					"op_src4_2_decoder": {
-						"op_3_project",
-					},
-					"op_3_project": {
-						"op_logToMemory_0_0_transform",
-					},
-					"op_logToMemory_0_0_transform": {
-						"op_logToMemory_0_1_encode",
-					},
-					"op_logToMemory_0_1_encode": {
-						"sink_logToMemory_0",
-					},
-				},
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			tp, err := PlanSQLWithSourcesAndSinks(def.GetDefaultRule(tt.name, tt.sql), nil)
-			assert.NoError(t, err)
-			assert.Equal(t, tt.topo, tp.GetTopo())
-		})
-	}
-}
diff --git a/pkg/mock/test_source.go b/pkg/mock/test_source.go
index 1b22feb7..62fff1ad 100644
--- a/pkg/mock/test_source.go
+++ b/pkg/mock/test_source.go
@@ -78,7 +78,22 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 				panic(err)
 			})
 		case api.TupleSource:
-			panic("added later")
+			err = ss.Subscribe(ctx, func(ctx api.StreamContext, message any, meta map[string]any, ts time.Time) {
+				switch mt := message.(type) {
+				case []byte:
+					result = append(result, model.NewDefaultRawTuple(mt, meta, ts))
+				case map[string]any:
+					result = append(result, model.NewDefaultSourceTuple(mt, meta, ts))
+				default:
+					panic("not supported yet")
+				}
+				limit--
+				if limit == 0 {
+					wg.Done()
+				}
+			}, func(ctx api.StreamContext, err error) {
+				panic(err)
+			})
 		default:
 			panic("wrong source type")
 		}