Repository: lf-edge/ekuiper
Pull request #2885 (2024-05-29 06:37:58)

refactor(api): source distinct pull and push

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2885?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 `59.45946%` with `45 lines` in your changes are missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@7b3b670`). [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/2885?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/topo/node/source\_node.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2885?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) | 18.75% | [25 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2885?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/2885?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=) | 67.92% | [12 Missing and 5 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2885?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/2885?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) | 88.89% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2885?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    #2885   +/-   ##
============================================
  Coverage             ?   68.78%           
============================================
  Files                ?      279           
  Lines                ?    26864           
  Branches             ?        0           
============================================
  Hits                 ?    18477           
  Misses               ?     6733           
  Partials             ?     1654           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2885?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/contract/api/source.go b/contract/api/source.go
index 67279fb6..64c02e8e 100644
--- a/contract/api/source.go
+++ b/contract/api/source.go
@@ -16,35 +16,45 @@ package api
 
 import "time"
 
-// The source capabilities are split to several functionality
+// The source capabilities are split to several trait
 // Implementations can implement part of them and combine
 
-// Source is the interface that wraps the basic Source method.
-// The lifecycle of a source: Provision -> Connect -> Subscribe -> Close
+// Source is the raw interface that wraps the basic Source method. It cannot be used independently, must implement more traits.
+// The lifecycle of a source: Provision -> Connect -> Subscribe/Pull -> Close
 type Source interface {
 	Nodelet
 	Connector
 }
 
-type ErrorIngest func(ctx StreamContext, err error)
-type BytesIngest func(ctx StreamContext, payload []byte, meta map[string]any, ts time.Time)
+/// Source interfaces to be implemented. With raw source interface and a selected mandatory trait
 
+// BytesSource receives the bytes payload pushed by the external source
 type BytesSource interface {
 	Source
 	Subscribe(ctx StreamContext, ingest BytesIngest, ingestError ErrorIngest) error
 }
 
-// TupleIngest reads in a structural data or its list.
-// It supports map and []map for now
-type TupleIngest func(ctx StreamContext, data any, meta map[string]any, ts time.Time)
-
+// TupleSource receives the non-bytes payload pushed by the external source
 type TupleSource interface {
 	Source
 	Subscribe(ctx StreamContext, ingest TupleIngest, ingestError ErrorIngest) error
 }
 
-type EOFIngest func(ctx StreamContext)
+// PullBytesSource fetch the bytes payload in an interval from the external source. Interval property must be defined
+type PullBytesSource interface {
+	Source
+	Pull(ctx StreamContext, trigger time.Time, ingest BytesIngest, ingestError ErrorIngest)
+}
+
+// PullTupleSource fetch the non-bytes payload in an interval from the external source. Interval property must be defined
+type PullTupleSource interface {
+	Source
+	Pull(ctx StreamContext, trigger time.Time, ingest TupleIngest, ingestError ErrorIngest)
+}
+
+/// Other optional traits
 
+// Bounded means the source can have an end.
 type Bounded interface {
 	SetEofIngest(eof EOFIngest)
 }
@@ -56,6 +66,7 @@ type Rewindable interface {
 	ResetOffset(input map[string]any) error
 }
 
+// LookupSource is a source feature to query the source on demand (TO be modified later)
 type LookupSource interface {
 	// Open creates the connection to the external data source
 	Open(ctx StreamContext) error
@@ -66,3 +77,13 @@ type LookupSource interface {
 	Lookup(ctx StreamContext, fields []string, keys []string, values []interface{}) (MessageTupleList, error)
 	Closable
 }
+
+/// helper function definition
+
+type ErrorIngest func(ctx StreamContext, err error)
+type BytesIngest func(ctx StreamContext, payload []byte, meta map[string]any, ts time.Time)
+
+// TupleIngest reads in a structural data or its list.
+// It supports map and []map for now
+type TupleIngest func(ctx StreamContext, data any, meta map[string]any, ts time.Time)
+type EOFIngest func(ctx StreamContext)
diff --git a/internal/io/file/source.go b/internal/io/file/source.go
index 1118d351..b5c5fea2 100644
--- a/internal/io/file/source.go
+++ b/internal/io/file/source.go
@@ -74,6 +74,10 @@ func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 	}
 	reader, ok := modules.GetFileStreamReader(ctx, cfg.FileType)
 	if ok {
+		// TODO support later. If decompression is set, we need to read in the whole file
+		if cfg.Decompression != "" {
+			return fmt.Errorf("decompression is not supported for %s file type", cfg.FileType)
+		}
 		err = reader.Provision(ctx, props)
 		if err != nil {
 			return err
@@ -82,6 +86,7 @@ func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 	} 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")
 	}
@@ -141,31 +146,11 @@ func (fs *Source) Connect(ctx api.StreamContext) error {
 	return nil
 }
 
-// Subscribe file source may ingest bytes or tuple
+// Pull 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.TupleIngest, ingestError api.ErrorIngest) error {
+func (fs *Source) Pull(ctx api.StreamContext, _ time.Time, ingest api.TupleIngest, ingestError api.ErrorIngest) {
 	fs.Load(ctx, ingest, ingestError)
-	if fs.config.Interval > 0 {
-		ticker := time.NewTicker(fs.config.Interval)
-		logger := ctx.GetLogger()
-		defer ticker.Stop()
-		for {
-			select {
-			case <-ticker.C:
-				logger.Debugf("Load file source again at %v", timex.GetNowInMilli())
-				fs.Load(ctx, ingest, ingestError)
-			case <-ctx.Done():
-				return nil
-			}
-		}
-	} else { // If not pulling is set, it is a bounded source
-		if fs.eof != nil {
-			ctx.GetLogger().Debugf("send eof")
-			fs.eof(ctx)
-		}
-	}
-	return nil
 }
 
 func (fs *Source) SetEofIngest(eof api.EOFIngest) {
@@ -362,7 +347,7 @@ func GetSource() api.Source {
 
 var (
 	// ingest possibly []byte and tuple
-	_ api.TupleSource = &Source{}
-	_ api.Bounded     = &Source{}
-	_ model.InfoNode  = &Source{}
+	_ api.PullTupleSource = &Source{}
+	_ api.Bounded         = &Source{}
+	_ model.InfoNode      = &Source{}
 )
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index 6b67f3fd..d6d6b1bb 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -21,8 +21,9 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
-	"github.com/lf-edge/ekuiper/v2/pkg/stat"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 // SourceNode is a node that connects to an external source
@@ -31,7 +32,12 @@ import (
 type SourceNode struct {
 	*defaultNode
 
-	s api.Source
+	s        api.Source
+	interval time.Duration
+}
+
+type c struct {
+	Interval time.Duration `json:"interval"`
 }
 
 // NewSourceNode creates a SourceConnectorNode
@@ -41,12 +47,25 @@ func NewSourceNode(ctx api.StreamContext, name string, ss api.Source, props map[
 		return nil, err
 	}
 	ctx.GetLogger().Infof("provision source %s with props %+v", name, props)
+	cc := &c{}
+	err = cast.MapToStruct(props, cc)
+	if err != nil {
+		return nil, err
+	}
 	m := &SourceNode{
 		defaultNode: newDefaultNode(name, rOpt),
 		s:           ss,
+		interval:    cc.Interval,
+	}
+	switch st := ss.(type) {
+	case api.Bounded:
+		st.SetEofIngest(m.ingestEof)
 	}
-	if bs, ok := ss.(api.Bounded); ok {
-		bs.SetEofIngest(m.ingestEof)
+	switch ss.(type) {
+	case api.PullTupleSource, api.PullBytesSource:
+		if cc.Interval < 1 {
+			return nil, fmt.Errorf("interval should be larger than 1ms for pull source")
+		}
 	}
 	return m, nil
 }
@@ -58,9 +77,6 @@ func NewSourceNode(ctx api.StreamContext, name string, ss api.Source, props map[
 // Open will be invoked by topo. It starts reading data.
 func (m *SourceNode) Open(ctx api.StreamContext, ctrlCh chan<- error) {
 	m.prepareExec(ctx, ctrlCh, "source")
-	if able, ok := m.s.(stat.StatsAble); ok {
-		able.SetupStats(m.statManager)
-	}
 	go m.Run(ctx, ctrlCh)
 }
 
@@ -145,6 +161,8 @@ func (m *SourceNode) Run(ctx api.StreamContext, ctrlCh chan<- error) {
 			err = ss.Subscribe(ctx, m.ingestBytes, m.ingestError)
 		case api.TupleSource:
 			err = ss.Subscribe(ctx, m.ingestAnyTuple, m.ingestError)
+		case api.PullBytesSource, api.PullTupleSource:
+			err = m.runPull(ctx)
 		}
 		if err != nil {
 			return err
@@ -157,6 +175,33 @@ func (m *SourceNode) Run(ctx api.StreamContext, ctrlCh chan<- error) {
 	<-ctx.Done()
 }
 
+func (m *SourceNode) runPull(ctx api.StreamContext) error {
+	m.doPull(ctx, timex.GetNow())
+	ticker := timex.GetTicker(m.interval)
+	go func() {
+		defer ticker.Stop()
+		for {
+			select {
+			case tc := <-ticker.C:
+				ctx.GetLogger().Debugf("source pull at %v", tc.UnixMilli())
+				m.doPull(ctx, tc)
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	return nil
+}
+
+func (m *SourceNode) doPull(ctx api.StreamContext, tc time.Time) {
+	switch ss := m.s.(type) {
+	case api.PullBytesSource:
+		ss.Pull(ctx, tc, m.ingestBytes, m.ingestError)
+	case api.PullTupleSource:
+		ss.Pull(ctx, tc, m.ingestAnyTuple, m.ingestError)
+	}
+}
+
 func (m *SourceNode) Close() {
 	m.defaultNode.Close()
 }
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index 6834f94d..ab44a35b 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -16,6 +16,7 @@ package planner
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/binder/io"
@@ -71,12 +72,6 @@ func transformSourceNode(ctx api.StreamContext, t *DataSourcePlan, mockSourcesPr
 	return nil, nil, 0, fmt.Errorf("unknown stream type %d", t.streamStmt.StreamType)
 }
 
-type SourcePropsForSplit struct {
-	Decompression string `json:"decompression"`
-	SelId         string `json:"connectionSelector"`
-	PayloadFormat string `json:"payloadFormat"`
-}
-
 func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, options *def.RuleOption, index int, ruleId string, pp node.UnOperation) (node.DataSourceNode, []node.OperatorNode, int, error) {
 	// Get all props
 	props := nodeConf.GetSourceConf(t.streamStmt.Options.TYPE, t.streamStmt.Options)
@@ -102,35 +97,25 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 		}
 		srcConnNode = srcSubtopo
 	}
-	// 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
 	}
 	index++
 	var ops []node.OperatorNode
 
-	if sp.Decompression != "" {
-		if info.NeedBatchDecode {
-			dco, err := node.NewDecompressOp(fmt.Sprintf("%d_decompress", index), options, sp.Decompression)
-			if err != nil {
-				return nil, nil, 0, err
-			}
-			index++
-			ops = append(ops, dco)
-		} else {
-			ctx.GetLogger().Warnf("source %s does not support decompression", t.name)
+	// Need to check after source has provisioned, so do not put it before provision
+	featureSet := checkFeatures(ss, sp, props)
+
+	if featureSet.needCompression {
+		dco, err := node.NewDecompressOp(fmt.Sprintf("%d_decompress", index), options, sp.Decompression)
+		if err != nil {
+			return nil, nil, 0, err
 		}
+		index++
+		ops = append(ops, dco)
 	}
 
-	if info.NeedDecode {
+	if featureSet.needDecode {
 		// Create the decode node
 		decodeNode, err := node.NewDecodeOp(ctx, false, fmt.Sprintf("%d_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
 		if err != nil {
@@ -140,7 +125,7 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 		ops = append(ops, decodeNode)
 	}
 
-	if sp.PayloadFormat != "" {
+	if featureSet.needPayloadDecode {
 		// Create the decode node
 		payloadDecodeNode, err := node.NewDecodeOp(ctx, true, fmt.Sprintf("%d_payload_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
 		if err != nil {
@@ -174,6 +159,35 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 	return srcConnNode, ops, 0, nil
 }
 
+type SourcePropsForSplit struct {
+	Decompression string        `json:"decompression"`
+	SelId         string        `json:"connectionSelector"`
+	PayloadFormat string        `json:"payloadFormat"`
+	Interval      time.Duration `json:"interval"`
+}
+
+type traits struct {
+	needConnection    bool
+	needCompression   bool
+	needDecode        bool
+	needPayloadDecode bool
+}
+
+// function to return if a sub node is needed
+func checkFeatures(ss api.Source, sp *SourcePropsForSplit, props map[string]any) traits {
+	info := checkByteSource(ss)
+	// TODO here is a hack for file source send interval. If it is sent in file, do not need to process sendInterval in decode
+	if info.HasInterval {
+		delete(props, "sendInterval")
+	}
+	return traits{
+		needConnection:    sp.SelId != "",
+		needCompression:   sp.Decompression != "" && (!info.HasCompress || info.NeedBatchDecode),
+		needDecode:        info.NeedDecode,
+		needPayloadDecode: sp.PayloadFormat != "",
+	}
+}
+
 func checkByteSource(ss api.Source) model.NodeInfo {
 	switch st := ss.(type) {
 	case model.InfoNode:
diff --git a/pkg/model/source_tuple.go b/pkg/model/source_tuple.go
index 4cb409cc..f5171412 100644
--- a/pkg/model/source_tuple.go
+++ b/pkg/model/source_tuple.go
@@ -16,6 +16,8 @@ package model
 
 import (
 	"time"
+
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 type DefaultSourceTuple struct {
@@ -34,6 +36,14 @@ func NewDefaultRawTuple(raw []byte, meta map[string]any, ts time.Time) *DefaultS
 	}
 }
 
+func NewDefaultRawTupleIgnoreTs(raw []byte, meta map[string]any) *DefaultSourceTuple {
+	return &DefaultSourceTuple{
+		meta: meta,
+		raw:  raw,
+		time: timex.Maxtime,
+	}
+}
+
 func NewDefaultSourceTuple(message map[string]any, meta map[string]any, timestamp time.Time) *DefaultSourceTuple {
 	return &DefaultSourceTuple{
 		message: message,
diff --git a/pkg/stat/stat.go b/pkg/stat/stat.go
deleted file mode 100644
index 15005e67..00000000
--- a/pkg/stat/stat.go
+++ /dev/null
@@ -1,21 +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 stat
-
-import "github.com/lf-edge/ekuiper/v2/internal/topo/node/metric"
-
-type StatsAble interface {
-	SetupStats(stats metric.StatManager)
-}


Test patch:
diff --git a/internal/io/file/source_test.go b/internal/io/file/source_test.go
index d09970e2..a9e83de4 100644
--- a/internal/io/file/source_test.go
+++ b/internal/io/file/source_test.go
@@ -183,6 +183,16 @@ func TestSourceProvision(t *testing.T) {
 				MoveTo:          filepath.Join(path, "ddd"),
 			},
 		},
+		{
+			name: "no decompression for stream typs",
+			props: map[string]any{
+				"datasource":    name,
+				"path":          path,
+				"fileType":      LINES_TYPE,
+				"decompression": "gzip",
+			},
+			e: "decompression is not supported for lines file type",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -225,7 +235,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.TupleSource)
+	r := GetSource()
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":            path,
 		"fileType":        "lines",
@@ -256,7 +266,7 @@ func TestCSVBatch(t *testing.T) {
 		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)
+	r := GetSource()
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":             path,
 		"fileType":         "csv",
@@ -284,7 +294,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.TupleSource)
+	r := GetSource()
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":       path,
 		"fileType":   "json",
@@ -313,7 +323,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.TupleSource)
+	r := GetSource()
 	r.(api.Bounded).SetEofIngest(func(ctx api.StreamContext) {
 		fmt.Printf("eof")
 	})
@@ -343,8 +353,6 @@ func TestIgnoreLines(t *testing.T) {
 }
 
 func TestIntervalAndDir(t *testing.T) {
-	ts0 := timex.GetNow()
-	ts := ts0.Add(2 * time.Second)
 	path, err := os.Getwd()
 	if err != nil {
 		t.Fatal(err)
@@ -355,12 +363,12 @@ func TestIntervalAndDir(t *testing.T) {
 		"file": filepath.Join(path, "test.lines"),
 	}
 	exp := []api.MessageTuple{
-		model.NewDefaultRawTuple([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta, ts),
-		model.NewDefaultRawTuple([]byte("{\"id\": 3,\"name\": \"John Smith\"}"), meta, ts),
-		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),
+		model.NewDefaultRawTupleIgnoreTs([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta),
+		model.NewDefaultRawTupleIgnoreTs([]byte("{\"id\": 3,\"name\": \"John Smith\"}"), meta),
+		model.NewDefaultRawTupleIgnoreTs([]byte("[{\"id\": 4,\"name\": \"John Smith\"},{\"id\": 5,\"name\": \"John Smith\"}]"), meta),
+		model.NewDefaultRawTupleIgnoreTs([]byte("{\"id\": 2,\"name\": \"Jane Doe\"}"), meta),
 	}
-	r := GetSource().(api.TupleSource)
+	r := GetSource()
 	mock.TestSourceConnector(t, r, map[string]any{
 		"path":             path,
 		"fileType":         "lines",
@@ -368,9 +376,12 @@ func TestIntervalAndDir(t *testing.T) {
 		"sendInterval":     "100ms",
 		"parallel":         true,
 		"ignoreStartLines": 1,
+		// only for test
+		"ignoreTs": true,
 	}, exp, func() {
-		fmt.Println("before add")
-		timex.Add(2 * time.Second)
-		fmt.Println("after add")
+		for i := 0; i < 10; i++ {
+			timex.Add(2 * time.Second)
+			time.Sleep(100 * time.Millisecond)
+		}
 	})
 }
diff --git a/internal/topo/node/source_node_test.go b/internal/topo/node/source_node_test.go
index 0496701a..ba14831f 100644
--- a/internal/topo/node/source_node_test.go
+++ b/internal/topo/node/source_node_test.go
@@ -118,6 +118,27 @@ func TestNewError(t *testing.T) {
 	})
 	assert.Error(t, err)
 	assert.Equal(t, "datasource name cannot be empty", err.Error())
+	_, err = NewSourceNode(ctx, "mock_connector", sc, map[string]any{"interval": "invalid", "datasource": "demo"}, &def.RuleOption{
+		BufferLength: 1024,
+		SendError:    true,
+	})
+	assert.Error(t, err)
+	assert.Equal(t, "1 error(s) decoding:\n\n* error decoding 'interval': time: invalid duration \"invalid\"", err.Error())
+
+	var pc api.PullTupleSource = &MockPullSource{}
+	_, err = NewSourceNode(ctx, "mock_connector", pc, map[string]any{"datasource": "demo"}, &def.RuleOption{
+		BufferLength: 1024,
+		SendError:    true,
+	})
+	assert.Error(t, err)
+	assert.Equal(t, "interval should be larger than 1ms for pull source", err.Error())
+
+	_, err = NewSourceNode(ctx, "mock_connector", pc, map[string]any{"datasource": "demo", "interval": "1s"}, &def.RuleOption{
+		BufferLength: 1024,
+		SendError:    true,
+	})
+	assert.NoError(t, err)
+	assert.True(t, pc.(*MockPullSource).set)
 }
 
 func TestConnError(t *testing.T) {
@@ -159,6 +180,86 @@ func TestConnError(t *testing.T) {
 	assert.Equal(t, "data is nil", errResult.Error())
 }
 
+func TestPull(t *testing.T) {
+	mc := mockclock.GetMockClock()
+	expects := []any{
+		&xsql.Tuple{
+			Metadata:  map[string]any{"topic": "demo"},
+			Timestamp: mc.Now(),
+			Emitter:   "mock_connector",
+			Message:   map[string]any{"index": 1},
+		},
+		&xsql.RawTuple{
+			Emitter:   "mock_connector",
+			Metadata:  map[string]any{"topic": "demo"},
+			Timestamp: mc.Now().Add(time.Second),
+			Rawdata:   []byte{2},
+		},
+		&xsql.Tuple{
+			Timestamp: mc.Now().Add(2 * time.Second),
+			Emitter:   "mock_connector",
+			Message:   map[string]any{"index": 3},
+		},
+		&xsql.Tuple{
+			Timestamp: mc.Now().Add(3 * time.Second),
+			Emitter:   "mock_connector",
+			Message:   map[string]any{"index": 4},
+		},
+		&xsql.Tuple{
+			Timestamp: mc.Now().Add(4 * time.Second),
+			Emitter:   "mock_connector",
+			Message:   map[string]any{"index": 5},
+			Metadata:  map[string]any{"topic": "demo"},
+		},
+	}
+	var sc api.PullTupleSource = &MockPullSource{}
+	ctx := mockContext.NewMockContext("rule1", "src1")
+	errCh := make(chan error)
+	scn, err := NewSourceNode(ctx, "mock_connector", sc, map[string]any{"datasource": "demo", "interval": "1s"}, &def.RuleOption{
+		BufferLength: 1024,
+		SendError:    true,
+	})
+	assert.NoError(t, err)
+	result := make(chan any, 10)
+	err = scn.AddOutput(result, "testResult")
+	assert.NoError(t, err)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	limit := len(expects)
+	actual := make([]any, 0, limit)
+	go func() {
+		defer wg.Done()
+		ticker := time.After(2000 * time.Second)
+		for {
+			select {
+			case sg := <-errCh:
+				switch et := sg.(type) {
+				case error:
+					assert.Fail(t, et.Error())
+					return
+				default:
+					fmt.Println("ctrlCh", et)
+				}
+			case tuple := <-result:
+				actual = append(actual, tuple)
+				limit--
+				if limit <= 0 {
+					return
+				}
+			case <-ticker:
+				assert.Fail(t, "timeout")
+				return
+			}
+		}
+	}()
+	scn.Open(ctx, errCh)
+	time.Sleep(10 * time.Millisecond)
+	timex.Add(10 * time.Second)
+	wg.Wait()
+	assert.Equal(t, expects, actual)
+}
+
 type MockSourceConnector struct {
 	data       [][]byte
 	topic      string
@@ -207,3 +308,58 @@ func (m *MockSourceConnector) Subscribe(ctx api.StreamContext, ingest api.BytesI
 	}()
 	return nil
 }
+
+type MockPullSource struct {
+	set       bool
+	pullTimes int
+}
+
+func (m *MockPullSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	return nil
+}
+
+func (m *MockPullSource) Close(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *MockPullSource) Connect(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *MockPullSource) Pull(ctx api.StreamContext, trigger time.Time, ingest api.TupleIngest, ingestError api.ErrorIngest) {
+	m.pullTimes++
+	var mess any
+	switch m.pullTimes % 5 {
+	case 0:
+		mess = map[string]any{
+			"index": m.pullTimes,
+		}
+	case 1:
+		mess = []map[string]any{
+			{
+				"index": m.pullTimes,
+			},
+		}
+	case 2:
+		mess = []byte{byte(m.pullTimes)}
+	case 3:
+		mess = &xsql.Tuple{
+			Message: map[string]any{
+				"index": m.pullTimes,
+			},
+		}
+	case 4:
+		mess = []*xsql.Tuple{
+			{
+				Message: map[string]any{
+					"index": m.pullTimes,
+				},
+			},
+		}
+	}
+	ingest(ctx, mess, map[string]any{"topic": "demo"}, trigger)
+}
+
+func (m *MockPullSource) SetEofIngest(eof api.EOFIngest) {
+	m.set = true
+}
diff --git a/internal/topo/planner/planner_source_test.go b/internal/topo/planner/planner_source_test.go
index d00715b5..d8befd7f 100644
--- a/internal/topo/planner/planner_source_test.go
+++ b/internal/topo/planner/planner_source_test.go
@@ -69,17 +69,17 @@ func TestPlanTopo(t *testing.T) {
 		{
 			conf: map[string]any{
 				"fileType":      "csv",
-				"decompression": "gzip",
 				"payloadFormat": "json",
 				"payloadField":  "col1",
+				"interval":      "20s",
 			},
 			p: "file",
 			k: "csv",
 		},
 		{
 			conf: map[string]any{
-				"fileType":      "lines",
-				"decompression": "gzip",
+				"fileType": "lines",
+				"interval": "20s",
 			},
 			p: "file",
 			k: "lines",
@@ -87,6 +87,7 @@ func TestPlanTopo(t *testing.T) {
 		{
 			conf: map[string]any{
 				"decompression": "gzip",
+				"interval":      "20s",
 			},
 			p: "file",
 			k: "json",
diff --git a/pkg/mock/test_source.go b/pkg/mock/test_source.go
index 62fff1ad..f998a62c 100644
--- a/pkg/mock/test_source.go
+++ b/pkg/mock/test_source.go
@@ -24,12 +24,19 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
 var count atomic.Value
 
+type ic struct {
+	Interval time.Duration `json:"interval"`
+	IgnoreTs bool          `json:"ignoreTs"`
+}
+
 func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expected any, sender func()) {
 	// init
 	c := count.Load()
@@ -38,9 +45,12 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 		c = 0
 	}
 	count.Store(c.(int) + 1)
+	cc := &ic{}
+	err := cast.MapToStruct(props, cc)
+	assert.NoError(t, err)
 	// provision
 	ctx, cancel := mockContext.NewMockContext(fmt.Sprintf("rule%d", c), "op1").WithCancel()
-	err := r.Provision(ctx, props)
+	err = r.Provision(ctx, props)
 	assert.NoError(t, err)
 	// connect, subscribe and read data
 	err = r.Connect(ctx)
@@ -65,35 +75,79 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 		result []api.MessageTuple
 	)
 	wg.Add(1)
+	ingestBytes := func(ctx api.StreamContext, payload []byte, meta map[string]any, ts time.Time) {
+		if cc.IgnoreTs {
+			result = append(result, model.NewDefaultRawTupleIgnoreTs(payload, meta))
+		} else {
+			result = append(result, model.NewDefaultRawTuple(payload, meta, ts))
+		}
+
+		limit--
+		if limit == 0 {
+			wg.Done()
+		}
+	}
+	ingestTuples := func(ctx api.StreamContext, message any, meta map[string]any, ts time.Time) {
+		switch mt := message.(type) {
+		case []byte:
+			if cc.IgnoreTs {
+				result = append(result, model.NewDefaultRawTupleIgnoreTs(mt, meta))
+			} else {
+				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()
+		}
+	}
 	go func() {
 		switch ss := r.(type) {
 		case api.BytesSource:
-			err = ss.Subscribe(ctx, func(ctx api.StreamContext, payload []byte, meta map[string]any, ts time.Time) {
-				result = append(result, model.NewDefaultRawTuple(payload, meta, ts))
-				limit--
-				if limit == 0 {
-					wg.Done()
-				}
-			}, func(ctx api.StreamContext, err error) {
+			err = ss.Subscribe(ctx, ingestBytes, func(ctx api.StreamContext, err error) {
 				panic(err)
 			})
 		case api.TupleSource:
-			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) {
+			err = ss.Subscribe(ctx, ingestTuples, func(ctx api.StreamContext, err error) {
 				panic(err)
 			})
+		case api.PullBytesSource, api.PullTupleSource:
+			switch ss := r.(type) {
+			case api.PullBytesSource:
+				ss.Pull(ctx, timex.GetNow(), ingestBytes, func(ctx api.StreamContext, err error) {
+					panic(err)
+				})
+			case api.PullTupleSource:
+				ss.Pull(ctx, timex.GetNow(), ingestTuples, func(ctx api.StreamContext, err error) {
+					panic(err)
+				})
+			}
+			ticker := timex.GetTicker(cc.Interval)
+			go func() {
+				defer ticker.Stop()
+				for {
+					select {
+					case tc := <-ticker.C:
+						ctx.GetLogger().Debugf("source pull at %v", tc.UnixMilli())
+						switch ss := r.(type) {
+						case api.PullBytesSource:
+							ss.Pull(ctx, tc, ingestBytes, func(ctx api.StreamContext, err error) {
+								panic(err)
+							})
+						case api.PullTupleSource:
+							ss.Pull(ctx, tc, ingestTuples, func(ctx api.StreamContext, err error) {
+								panic(err)
+							})
+						}
+					case <-ctx.Done():
+						return
+					}
+				}
+			}()
 		default:
 			panic("wrong source type")
 		}
@@ -103,7 +157,7 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 		assert.NoError(t, err)
 	}()
 
-	ticker := time.After(60 * time.Second)
+	ticker := time.After(60000 * time.Second)
 	finished := make(chan struct{})
 	go func() {
 		wg.Wait()