Repository: lf-edge/ekuiper
Pull request #2945 (2024-06-24 07:35:40)

feat: support sql sink 

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2945?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 `35.46512%` with `111 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@33842f9`). [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/2945?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/sql/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2945?src=pr&el=tree&filepath=internal%2Fio%2Fsql%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vc3FsL3NpbmsuZ28=) | 30.57% | [96 Missing and 13 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2945?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/bump/bump3.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2945?src=pr&el=tree&filepath=internal%2Fserver%2Fbump%2Fbump3.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL2J1bXAvYnVtcDMuZ28=) | 85.71% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2945?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    #2945   +/-   ##
============================================
  Coverage             ?   69.62%           
============================================
  Files                ?      313           
  Lines                ?    29971           
  Branches             ?        0           
============================================
  Hits                 ?    20867           
  Misses               ?     7287           
  Partials             ?     1817           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2945?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/binder/io/builtin.go b/internal/binder/io/builtin.go
index 91f9f547..d3efbf0a 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -22,7 +22,6 @@ import (
 	mqttCon "github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/io/neuron"
 	"github.com/lf-edge/ekuiper/v2/internal/io/sink"
-	"github.com/lf-edge/ekuiper/v2/internal/io/sql"
 	sqlCon "github.com/lf-edge/ekuiper/v2/internal/io/sql/client"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 	"github.com/lf-edge/ekuiper/v2/pkg/nng"
@@ -37,7 +36,6 @@ func init() {
 	modules.RegisterSource("neuron", neuron.GetSource)
 	// modules.RegisterSource("websocket", func() api.Source { return &websocket.WebsocketSource{} })
 	// modules.RegisterSource("simulator", func() api.Source { return &simulator.Source{} })
-	modules.RegisterSource("sql", sql.GetSource)
 
 	modules.RegisterSink("log", sink.NewLogSink)
 	modules.RegisterSink("logToMemory", sink.NewLogSinkToMemory)
diff --git a/internal/binder/io/ext_full.go b/internal/binder/io/ext_full.go
index 79365734..cecaa64b 100644
--- a/internal/binder/io/ext_full.go
+++ b/internal/binder/io/ext_full.go
@@ -16,6 +16,11 @@
 
 package io
 
+import (
+	"github.com/lf-edge/ekuiper/v2/internal/io/sql"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
 func init() {
 	//modules.RegisterSource("random", func() api.Source { return random.GetSource() })
 	//modules.RegisterSource("video", func() api.Source { return video.GetSource() })
@@ -31,4 +36,6 @@ func init() {
 	// sinks["tdengine"] = func() api.Sink { return tdengine.GetSink() }
 	// sinks["zmq"] = func() api.Sink { return zmqSink.GetSink() }
 	// sources["zmq"] = func() api.Source { return zmq.GetSource() }
+	modules.RegisterSource("sql", sql.GetSource)
+	modules.RegisterSink("sql", sql.GetSink)
 }
diff --git a/internal/io/file/source.go b/internal/io/file/source.go
index 8e55a920..4d88ae37 100644
--- a/internal/io/file/source.go
+++ b/internal/io/file/source.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-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.
@@ -203,16 +203,10 @@ func (fs *Source) Load(ctx api.StreamContext, ingest api.TupleIngest, ingestErro
 	} else {
 		fs.parseFile(ctx, fs.file, ingest, ingestError)
 	}
-	// TODO deal with table case
-	//if fs.config.IsTable {
-	//	select {
-	//	case consumer <- api.NewDefaultSourceTupleWithTime(nil, nil, rcvTime):
-	//		// do nothing
-	//	case <-ctx.Done():
-	//		return nil
-	//	}
-	//}
-	ctx.GetLogger().Debug("All tuples sent")
+	if fs.config.Interval == 0 && fs.eof != nil {
+		fs.eof(ctx)
+		ctx.GetLogger().Debug("All tuples sent")
+	}
 }
 
 func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.TupleIngest, ingestError api.ErrorIngest) {
diff --git a/internal/io/sql/sink.go b/internal/io/sql/sink.go
new file mode 100644
index 00000000..3ac421a8
--- /dev/null
+++ b/internal/io/sql/sink.go
@@ -0,0 +1,272 @@
+// 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.
+
+//go:build full
+// +build full
+
+package sql
+
+import (
+	"fmt"
+	"reflect"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/sql/client"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+)
+
+type SQLSinkConnector struct {
+	config *sqlSinkConfig
+	conn   *client.SQLConnection
+	props  map[string]any
+}
+
+type sqlSinkConfig struct {
+	DBUrl        string   `json:"dburl"`
+	Table        string   `json:"table"`
+	Fields       []string `json:"fields"`
+	RowKindField string   `json:"rowKindField"`
+	KeyField     string   `json:"keyField"`
+}
+
+func (c *sqlSinkConfig) buildInsertSql(ctx api.StreamContext, mapData map[string]interface{}) ([]string, string, error) {
+	keys, vals, err := c.getKeyValues(ctx, mapData)
+	if err != nil {
+		return keys, "", err
+	}
+	sqlStr := "(" + strings.Join(vals, ",") + ")"
+	return keys, sqlStr, nil
+}
+
+func (c *sqlSinkConfig) getKeyValues(ctx api.StreamContext, mapData map[string]interface{}) ([]string, []string, error) {
+	if 0 == len(mapData) {
+		return nil, nil, fmt.Errorf("data is empty.")
+	}
+	logger := ctx.GetLogger()
+	var keys, vals []string
+
+	if len(c.Fields) != 0 {
+		for _, k := range c.Fields {
+			keys = append(keys, k)
+			if v, ok := mapData[k]; ok && v != nil {
+				if reflect.String == reflect.TypeOf(v).Kind() {
+					vals = append(vals, fmt.Sprintf("'%v'", v))
+				} else {
+					vals = append(vals, fmt.Sprintf(`%v`, v))
+				}
+			} else {
+				logger.Warn("not found field:", k)
+				vals = append(vals, fmt.Sprintf(`NULL`))
+			}
+		}
+	} else {
+		for k, v := range mapData {
+			keys = append(keys, k)
+			if reflect.String == reflect.TypeOf(v).Kind() {
+				vals = append(vals, fmt.Sprintf("'%v'", v))
+			} else {
+				vals = append(vals, fmt.Sprintf(`%v`, v))
+			}
+		}
+	}
+	return keys, vals, nil
+}
+
+func (s *SQLSinkConnector) Provision(ctx api.StreamContext, configs map[string]any) error {
+	c := &sqlSinkConfig{}
+	err := cast.MapToStruct(configs, c)
+	if err != nil {
+		return err
+	}
+	if c.DBUrl == "" {
+		return fmt.Errorf("property dburl is required")
+	}
+	if c.Table == "" {
+		return fmt.Errorf("property Table is required")
+	}
+	if c.RowKindField != "" && c.KeyField == "" {
+		return fmt.Errorf("keyField is required when rowKindField is set")
+	}
+	s.config = c
+	s.props = configs
+	return nil
+}
+
+func (s *SQLSinkConnector) Connect(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Connecting to sql server")
+	var cli *client.SQLConnection
+	var err error
+	id := s.config.DBUrl
+	conn, err := connection.FetchConnection(ctx, id, "sql", s.props)
+	if err != nil {
+		return err
+	}
+	cli = conn.(*client.SQLConnection)
+	s.conn = cli
+	return err
+}
+
+func (s *SQLSinkConnector) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Closing sql sink connector url:%v", s.config.DBUrl)
+	if s.conn != nil {
+		id := s.config.DBUrl
+		connection.DetachConnection(ctx, id, s.props)
+		s.conn.DetachSub(ctx, s.props)
+	}
+	return nil
+}
+
+func (s *SQLSinkConnector) Collect(ctx api.StreamContext, item api.MessageTuple) (err error) {
+	return s.collect(ctx, item.ToMap())
+}
+
+func (s *SQLSinkConnector) collect(ctx api.StreamContext, item map[string]any) (err error) {
+	if len(s.config.RowKindField) < 1 {
+		var keys []string = nil
+		var values []string = nil
+		var vars string
+		keys, vars, err = s.config.buildInsertSql(ctx, item)
+		if err != nil {
+			return err
+		}
+		values = append(values, vars)
+		if keys != nil {
+			sqlStr := buildInsertSQL(s.config.Table, keys, values)
+			return s.writeToDB(ctx, sqlStr)
+		}
+		return nil
+	}
+	return s.save(ctx, s.config.Table, item)
+}
+
+func (s *SQLSinkConnector) CollectList(ctx api.StreamContext, items api.MessageTupleList) (err error) {
+	return s.collectList(ctx, items.ToMaps())
+}
+
+func (s *SQLSinkConnector) collectList(ctx api.StreamContext, items []map[string]any) (err error) {
+	var keys []string = nil
+	var values []string = nil
+	var vars string
+	if len(s.config.RowKindField) < 1 {
+		for _, mapData := range items {
+			keys, vars, err = s.config.buildInsertSql(ctx, mapData)
+			if err != nil {
+				return err
+			}
+			values = append(values, vars)
+		}
+		if keys != nil {
+			sqlStr := buildInsertSQL(s.config.Table, keys, values)
+			return s.writeToDB(ctx, sqlStr)
+		}
+		return nil
+	}
+	for _, el := range items {
+		err := s.save(ctx, s.config.Table, el)
+		if err != nil {
+			ctx.GetLogger().Error(err)
+		}
+	}
+	return nil
+}
+
+// save save updatable data only to db
+func (s *SQLSinkConnector) save(ctx api.StreamContext, table string, data map[string]interface{}) error {
+	rowkind := ast.RowkindInsert
+	c, ok := data[s.config.RowKindField]
+	if ok {
+		rowkind, ok = c.(string)
+		if !ok {
+			return fmt.Errorf("rowkind field %s is not a string in data %v", s.config.RowKindField, data)
+		}
+		if rowkind != ast.RowkindInsert && rowkind != ast.RowkindUpdate && rowkind != ast.RowkindDelete {
+			return fmt.Errorf("invalid rowkind %s", rowkind)
+		}
+	}
+	var sqlStr string
+	switch rowkind {
+	case ast.RowkindInsert:
+		keys, vars, err := s.config.buildInsertSql(ctx, data)
+		if err != nil {
+			return err
+		}
+		values := []string{vars}
+		if keys != nil {
+			sqlStr = buildInsertSQL(table, keys, values)
+		}
+	case ast.RowkindUpdate:
+		keyval, ok := data[s.config.KeyField]
+		if !ok {
+			return fmt.Errorf("field %s does not exist in data %v", s.config.KeyField, data)
+		}
+		keys, vals, err := s.config.getKeyValues(ctx, data)
+		if err != nil {
+			return err
+		}
+		sqlStr = fmt.Sprintf("UPDATE %s SET ", table)
+		for i, key := range keys {
+			if i != 0 {
+				sqlStr += ","
+			}
+			sqlStr += fmt.Sprintf("%s=%s", key, vals[i])
+		}
+		if _, ok := keyval.(string); ok {
+			sqlStr += fmt.Sprintf(" WHERE %s = '%s';", s.config.KeyField, keyval)
+		} else {
+			sqlStr += fmt.Sprintf(" WHERE %s = %v;", s.config.KeyField, keyval)
+		}
+	case ast.RowkindDelete:
+		keyval, ok := data[s.config.KeyField]
+		if !ok {
+			return fmt.Errorf("field %s does not exist in data %v", s.config.KeyField, data)
+		}
+		if _, ok := keyval.(string); ok {
+			sqlStr = fmt.Sprintf("DELETE FROM %s WHERE %s = '%s';", table, s.config.KeyField, keyval)
+		} else {
+			sqlStr = fmt.Sprintf("DELETE FROM %s WHERE %s = %v;", table, s.config.KeyField, keyval)
+		}
+	default:
+		return fmt.Errorf("invalid rowkind %s", rowkind)
+	}
+	return s.writeToDB(ctx, sqlStr)
+}
+
+func (s *SQLSinkConnector) writeToDB(ctx api.StreamContext, sqlStr string) error {
+	ctx.GetLogger().Debugf(sqlStr)
+	r, err := s.conn.GetDB().Exec(sqlStr)
+	if err != nil {
+		return errorx.NewIOErr(err.Error())
+	}
+	d, err := r.RowsAffected()
+	if err != nil {
+		ctx.GetLogger().Errorf("get rows affected error: %s", err.Error())
+	}
+	ctx.GetLogger().Debugf("Rows affected: %d", d)
+	return nil
+}
+
+func buildInsertSQL(table string, keys []string, values []string) string {
+	sql := fmt.Sprintf("INSERT INTO %s (%s) values ", table, strings.Join(keys, ",")) + strings.Join(values, ",") + ";"
+	return sql
+}
+
+func GetSink() api.Sink {
+	return &SQLSinkConnector{}
+}
+
+var _ api.TupleCollector = &SQLSinkConnector{}
diff --git a/internal/io/sql/source.go b/internal/io/sql/source.go
index 54565ab4..5ff91f9d 100644
--- a/internal/io/sql/source.go
+++ b/internal/io/sql/source.go
@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//go:build full
+// +build full
+
 package sql
 
 import (
diff --git a/internal/server/bump/bump3.go b/internal/server/bump/bump3.go
index f6309fb8..ae79054b 100644
--- a/internal/server/bump/bump3.go
+++ b/internal/server/bump/bump3.go
@@ -24,9 +24,29 @@ import (
 )
 
 func bumpFrom2TO3() error {
+	rewriteSQLSinkConfiguration()
 	return rewriteSQLSourceConfiguration()
 }
 
+func rewriteSQLSinkConfiguration() error {
+	keyProps, err := conf.GetCfgFromKVStorage("sinks", "sql", "")
+	if err != nil {
+		return err
+	}
+	for key, props := range keyProps {
+		dbURLRaw, ok := props["url"]
+		if ok {
+			dbURL, ok := dbURLRaw.(string)
+			if ok {
+				props["dburl"] = dbURL
+				_, _, confKey, _ := extractKey(key)
+				conf.WriteCfgIntoKVStorage("sinks", "sql", confKey, props)
+			}
+		}
+	}
+	return nil
+}
+
 func rewriteSQLSourceConfiguration() error {
 	keyProps, err := conf.GetCfgFromKVStorage("sources", "sql", "")
 	if err != nil {
diff --git a/internal/topo/node/concurrent.go b/internal/topo/node/concurrent.go
index 9625be4f..43ca2dfc 100644
--- a/internal/topo/node/concurrent.go
+++ b/internal/topo/node/concurrent.go
@@ -69,7 +69,6 @@ func merge(ctx api.StreamContext, node *defaultSinkNode, sendInterval time.Durat
 						time.Sleep(sendInterval)
 					}
 				}
-				node.statManager.IncTotalMessagesProcessed(1)
 			case <-ctx.Done():
 				ctx.GetLogger().Infof("merge done")
 				return
@@ -112,6 +111,7 @@ func worker(ctx api.StreamContext, node *defaultSinkNode, i int, wf workerFunc,
 			default:
 				node.statManager.IncTotalRecordsIn()
 				result = wf(ctx, item)
+				node.statManager.IncTotalMessagesProcessed(1)
 			}
 			select {
 			case output <- result:
diff --git a/internal/topo/node/join_align_node.go b/internal/topo/node/join_align_node.go
index b30b4574..2226f6c6 100644
--- a/internal/topo/node/join_align_node.go
+++ b/internal/topo/node/join_align_node.go
@@ -27,19 +27,27 @@ import (
 // The input for batch table MUST be *WindowTuples
 type JoinAlignNode struct {
 	*defaultSinkNode
-	// states
+	// table states
 	batch map[string][]*xsql.Tuple
+	size  map[string]int
 }
 
 const BatchKey = "$$batchInputs"
 
-func NewJoinAlignNode(name string, emitters []string, options *def.RuleOption) (*JoinAlignNode, error) {
+func NewJoinAlignNode(name string, emitters []string, sizes []int, options *def.RuleOption) (*JoinAlignNode, error) {
 	batch := make(map[string][]*xsql.Tuple, len(emitters))
-	for _, e := range emitters {
-		batch[e] = nil
+	size := make(map[string]int, len(emitters))
+	for i, e := range emitters {
+		s := sizes[i]
+		if s >= 9999 {
+			s = 100
+		}
+		batch[e] = make([]*xsql.Tuple, 0, s)
+		size[e] = sizes[i]
 	}
 	n := &JoinAlignNode{
 		batch: batch,
+		size:  size,
 	}
 	n.defaultSinkNode = newDefaultSinkNode(name, options)
 	return n, nil
@@ -84,28 +92,21 @@ func (n *JoinAlignNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 					n.statManager.ProcessTimeStart()
 					switch d := data.(type) {
 					case *xsql.Tuple:
-						log.Debugf("JoinAlignNode receive tuple input %s", d)
-						n.alignBatch(ctx, d)
-					case *xsql.WindowTuples:
-						if d.WindowRange != nil { // real window
-							log.Debugf("JoinAlignNode receive window input %s", d)
-							n.alignBatch(ctx, d)
-						} else { // table window
-							log.Debugf("JoinAlignNode receive batch source %s", d)
-							if et, ok := d.Content[0].(xsql.EmittedData); ok {
-								emitter := et.GetEmitter()
-								// Buffer and update batch inputs
-								_, ok := n.batch[emitter]
-								if !ok {
-									e := fmt.Errorf("run JoinAlignNode error: receive batch input from unknown emitter %[1]T(%[1]v)", d)
-									n.Broadcast(e)
-									n.statManager.IncTotalExceptions(e.Error())
-									break
-								}
-								n.batch[emitter] = convertToTupleSlice(d.Content)
-								_ = ctx.PutState(BatchKey, n.batch)
+						log.Debugf("JoinAlignNode receive tuple input %v", d)
+						if b, ok := n.batch[d.Emitter]; ok {
+							s := n.size[d.Emitter]
+							if len(b) >= s {
+								b = b[s-len(b)+1:]
 							}
+							b = append(b, d)
+							n.batch[d.Emitter] = b
+							_ = ctx.PutState(BatchKey, n.batch)
+						} else {
+							n.alignBatch(ctx, d)
 						}
+					case *xsql.WindowTuples:
+						log.Debugf("JoinAlignNode receive window input %v", d)
+						n.alignBatch(ctx, d)
 					default:
 						e := fmt.Errorf("run JoinAlignNode error: invalid input type but got %[1]T(%[1]v)", d)
 						n.Broadcast(e)
@@ -123,14 +124,6 @@ func (n *JoinAlignNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 	}()
 }
 
-func convertToTupleSlice(content []xsql.Row) []*xsql.Tuple {
-	tuples := make([]*xsql.Tuple, len(content))
-	for i, v := range content {
-		tuples[i] = v.(*xsql.Tuple)
-	}
-	return tuples
-}
-
 func (n *JoinAlignNode) alignBatch(_ api.StreamContext, input any) {
 	n.statManager.ProcessTimeStart()
 	var w *xsql.WindowTuples
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index d439cd19..3ae4c2f8 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -61,12 +61,6 @@ func NewSourceNode(ctx api.StreamContext, name string, ss api.Source, props map[
 	case api.Bounded:
 		st.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
 }
 
@@ -110,6 +104,7 @@ func (m *SourceNode) ingestAnyTuple(ctx api.StreamContext, data any, meta map[st
 	case []byte:
 		tuple := &xsql.RawTuple{Emitter: m.name, Rawdata: mess, Timestamp: ts, Metadata: meta}
 		m.Broadcast(tuple)
+		m.statManager.IncTotalRecordsOut()
 	// Source tuples are expected from memory
 	case *xsql.Tuple:
 		m.ingestTuple(mess, ts)
@@ -219,22 +214,24 @@ func (m *SourceNode) runPull(ctx api.StreamContext) error {
 	if err != nil {
 		return err
 	}
-	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())
-				e := m.doPull(ctx, tc)
-				if e != nil {
-					m.ingestError(ctx, e)
+	if m.interval > 0 {
+		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())
+					e := m.doPull(ctx, tc)
+					if e != nil {
+						m.ingestError(ctx, e)
+					}
+				case <-ctx.Done():
+					return
 				}
-			case <-ctx.Done():
-				return
 			}
-		}
-	}()
+		}()
+	}
 	return nil
 }
 
diff --git a/internal/topo/operator/table_processor.go b/internal/topo/operator/table_processor.go
deleted file mode 100644
index 97f4f3f1..00000000
--- a/internal/topo/operator/table_processor.go
+++ /dev/null
@@ -1,104 +0,0 @@
-// Copyright 2021-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 operator
-
-import (
-	"fmt"
-
-	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/xsql"
-	"github.com/lf-edge/ekuiper/v2/pkg/ast"
-)
-
-type TableProcessor struct {
-	// Pruned stream fields. Could be streamField(with data type info) or string
-	defaultFieldProcessor
-
-	checkSchema  bool
-	isBatchInput bool // whether the inputs are batched, such as file which sends multiple messages at a batch. If batch input, only fires when EOF is received. This is mutual exclusive with retainSize.
-	retainSize   int  // how many(maximum) messages to be retained for each output
-	emitterName  string
-	// States
-	output       *xsql.WindowTuples // current batched message collection
-	batchEmitted bool               // If batch input, this is the signal for whether the last batch has emitted. If true, reinitialize.
-}
-
-func NewTableProcessor(isSchemaless bool, name string, fields map[string]*ast.JsonStreamField, options *ast.Options) (*TableProcessor, error) {
-	p := &TableProcessor{emitterName: name, batchEmitted: true, retainSize: 1}
-	if !isSchemaless && options.STRICT_VALIDATION {
-		p.defaultFieldProcessor = defaultFieldProcessor{
-			streamFields: fields, timestampFormat: options.TIMESTAMP_FORMAT,
-		}
-		p.checkSchema = true
-	}
-	if options.RETAIN_SIZE > 0 {
-		p.retainSize = options.RETAIN_SIZE
-		p.isBatchInput = false
-	} else if isBatch(options.TYPE) {
-		p.isBatchInput = true
-		p.retainSize = 0
-	}
-	return p, nil
-}
-
-// Apply
-//
-//	input: *xsql.Tuple or BatchCount
-//	output: WindowTuples
-func (p *TableProcessor) Apply(ctx api.StreamContext, data interface{}, _ *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) interface{} {
-	logger := ctx.GetLogger()
-	tuple, ok := data.(*xsql.Tuple)
-	if !ok {
-		return fmt.Errorf("expect *xsql.Tuple data type")
-	}
-	logger.Debugf("preprocessor receive %v", tuple)
-	if p.batchEmitted {
-		p.output = &xsql.WindowTuples{
-			Content: make([]xsql.Row, 0),
-		}
-		p.batchEmitted = false
-	}
-	if tuple.Message != nil {
-		if p.checkSchema {
-			err := p.validateAndConvert(tuple)
-			if err != nil {
-				return fmt.Errorf("error in preprocessor: %s", err)
-			}
-		}
-		var newTuples []xsql.Row
-		_ = p.output.Range(func(i int, r xsql.ReadonlyRow) (bool, error) {
-			if p.retainSize > 0 && p.output.Len() == p.retainSize && i == 0 {
-				return true, nil
-			}
-			newTuples = append(newTuples, r.(xsql.Row))
-			return true, nil
-		})
-		newTuples = append(newTuples, tuple)
-		p.output = &xsql.WindowTuples{
-			Content: newTuples,
-		}
-		if !p.isBatchInput {
-			return p.output
-		}
-	} else if p.isBatchInput { // EOF
-		p.batchEmitted = true
-		return p.output
-	}
-	return nil
-}
-
-func isBatch(t string) bool {
-	return t == "file" || t == ""
-}
diff --git a/internal/topo/planner/joinAlignPlan.go b/internal/topo/planner/joinAlignPlan.go
index 2e5c03fb..cb11dff9 100644
--- a/internal/topo/planner/joinAlignPlan.go
+++ b/internal/topo/planner/joinAlignPlan.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// Copyright 2021-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.
@@ -16,9 +16,16 @@ package planner
 
 import "github.com/lf-edge/ekuiper/v2/pkg/ast"
 
+const (
+	DefaultRetainSize = 1
+	MaxRetainSize     = 9999
+)
+
 type JoinAlignPlan struct {
 	baseLogicalPlan
 	Emitters []string
+	// retain size for each table emitter
+	Sizes []int
 }
 
 func (p *JoinAlignPlan) BuildExplainInfo() {
diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
index b21a79ce..179e5028 100644
--- a/internal/topo/planner/planner.go
+++ b/internal/topo/planner/planner.go
@@ -256,7 +256,7 @@ func buildOps(lp LogicalPlan, tp *topo.Topo, options *def.RuleOption, sources ma
 	case *LookupPlan:
 		op, err = planLookupSource(tp.GetContext(), t, options)
 	case *JoinAlignPlan:
-		op, err = node.NewJoinAlignNode(fmt.Sprintf("%d_join_aligner", newIndex), t.Emitters, options)
+		op, err = node.NewJoinAlignNode(fmt.Sprintf("%d_join_aligner", newIndex), t.Emitters, t.Sizes, options)
 	case *JoinPlan:
 		op = Transform(&operator.JoinOp{Joins: t.joins, From: t.from}, fmt.Sprintf("%d_join", newIndex), options)
 	case *FilterPlan:
@@ -318,6 +318,7 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *def.RuleOption, store kv.
 		lookupTableChildren map[string]*ast.Options
 		scanTableChildren   []LogicalPlan
 		scanTableEmitters   []string
+		scanTableSizes      []int
 		streamEmitters      []string
 		w                   *ast.Window
 		ds                  ast.Dimensions
@@ -349,6 +350,18 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *def.RuleOption, store kv.
 			} else {
 				scanTableChildren = append(scanTableChildren, p)
 				scanTableEmitters = append(scanTableEmitters, string(sInfo.stmt.Name))
+				tableSize := sInfo.stmt.Options.RETAIN_SIZE
+				if tableSize == 0 {
+					switch sInfo.stmt.Options.TYPE {
+					// If retainSize is not set, file table will try to read all the content in it
+					case "", "file":
+						// TODO use interface to determine if the table is batch like file
+						tableSize = MaxRetainSize
+					default:
+						tableSize = DefaultRetainSize
+					}
+				}
+				scanTableSizes = append(scanTableSizes, tableSize)
 			}
 		}
 	}
@@ -438,6 +451,7 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *def.RuleOption, store kv.
 			if len(scanTableChildren) > 0 {
 				p = JoinAlignPlan{
 					Emitters: scanTableEmitters,
+					Sizes:    scanTableSizes,
 				}.Init()
 				p.SetChildren(append(children, scanTableChildren...))
 				children = []LogicalPlan{p}
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index 65d83d08..ef685b5e 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -36,40 +36,31 @@ func transformSourceNode(ctx api.StreamContext, t *DataSourcePlan, mockSourcesPr
 	if isMock {
 		t.streamStmt.Options.TYPE = "simulator"
 	}
-	switch t.streamStmt.StreamType {
-	case ast.TypeStream:
-		strType := t.streamStmt.Options.TYPE
-		if strType == "" {
+	strType := t.streamStmt.Options.TYPE
+	if strType == "" {
+		switch t.streamStmt.StreamType {
+		case ast.TypeStream:
 			strType = "mqtt"
-			t.streamStmt.Options.TYPE = strType
+		case ast.TypeTable:
+			strType = "file"
 		}
-		si, err := io.Source(strType)
-		if err != nil {
-			return nil, nil, 0, err
-		}
-		if si == nil {
-			return nil, nil, 0, fmt.Errorf("source type %s not found", strType)
-		}
-		var pp node.UnOperation
-		if t.iet || (!isSchemaless && (t.streamStmt.Options.STRICT_VALIDATION || t.isBinary)) {
-			pp, err = operator.NewPreprocessor(isSchemaless, t.streamFields, t.allMeta, t.metaFields, t.iet, t.timestampField, t.timestampFormat, t.isBinary, t.streamStmt.Options.STRICT_VALIDATION)
-			if err != nil {
-				return nil, nil, 0, err
-			}
-		}
-		return splitSource(ctx, t, si, options, index, ruleId, pp)
-	case ast.TypeTable:
-		si, err := io.Source(t.streamStmt.Options.TYPE)
-		if err != nil {
-			return nil, nil, 0, err
-		}
-		pp, err := operator.NewTableProcessor(isSchemaless, string(t.name), t.streamFields, t.streamStmt.Options)
+		t.streamStmt.Options.TYPE = strType
+	}
+	si, err := io.Source(strType)
+	if err != nil {
+		return nil, nil, 0, err
+	}
+	if si == nil {
+		return nil, nil, 0, fmt.Errorf("source type %s not found", strType)
+	}
+	var pp node.UnOperation
+	if t.iet || (!isSchemaless && (t.streamStmt.Options.STRICT_VALIDATION || t.isBinary)) {
+		pp, err = operator.NewPreprocessor(isSchemaless, t.streamFields, t.allMeta, t.metaFields, t.iet, t.timestampField, t.timestampFormat, t.isBinary, t.streamStmt.Options.STRICT_VALIDATION)
 		if err != nil {
 			return nil, nil, 0, err
 		}
-		return splitSource(ctx, t, si, options, index, ruleId, pp)
 	}
-	return nil, nil, 0, fmt.Errorf("unknown stream type %d", t.streamStmt.StreamType)
+	return splitSource(ctx, t, si, options, index, ruleId, pp)
 }
 
 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) {


Test patch:
diff --git a/internal/io/sql/sink_test.go b/internal/io/sql/sink_test.go
new file mode 100644
index 00000000..07957d80
--- /dev/null
+++ b/internal/io/sql/sink_test.go
@@ -0,0 +1,286 @@
+// 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.
+
+//go:build full
+// +build full
+
+package sql
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/io/sql/testx"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestSQLSinkCollect(t *testing.T) {
+	connection.InitConnectionManager4Test()
+	ctx := mockContext.NewMockContext("1", "2")
+	s, err := testx.SetupEmbeddedMysqlServer(address, port)
+	require.NoError(t, err)
+	defer func() {
+		s.Close()
+	}()
+	tableName := "t"
+	dburl := fmt.Sprintf("mysql://root:@%v:%v/test", address, port)
+	testcases := []struct {
+		props map[string]any
+		data  map[string]any
+		a     int
+		b     int
+	}{
+		{
+			props: map[string]interface{}{
+				"dburl": dburl,
+				"table": tableName,
+			},
+			data: map[string]any{
+				"a": 2,
+				"b": 2,
+			},
+			a: 2,
+			b: 2,
+		},
+		{
+			props: map[string]interface{}{
+				"dburl":  dburl,
+				"table":  tableName,
+				"fields": []string{"a", "b"},
+			},
+			data: map[string]any{
+				"a": 3,
+				"b": 3,
+				"c": 3,
+			},
+			a: 3,
+			b: 3,
+		},
+		{
+			props: map[string]interface{}{
+				"dburl":        dburl,
+				"table":        tableName,
+				"fields":       []string{"a", "b"},
+				"rowKindField": "action",
+				"keyField":     "a",
+			},
+			data: map[string]any{
+				"a":      4,
+				"b":      4,
+				"c":      4,
+				"action": "insert",
+			},
+			a: 4,
+			b: 4,
+		},
+	}
+	for _, tc := range testcases {
+		sqlSink := &SQLSinkConnector{}
+		require.NoError(t, sqlSink.Provision(ctx, tc.props))
+		require.NoError(t, sqlSink.Connect(ctx))
+		require.NoError(t, sqlSink.collect(ctx, tc.data))
+		rows, err := sqlSink.conn.GetDB().Query(fmt.Sprintf("select a,b from t where a = %v and b = %v", tc.a, tc.b))
+		require.NoError(t, err)
+		count := 0
+		for rows.Next() {
+			count++
+			var a int
+			var b int
+			require.NoError(t, rows.Scan(&a, &b))
+			require.Equal(t, tc.a, a)
+			require.Equal(t, tc.b, b)
+		}
+		sqlSink.Close(ctx)
+		require.Equal(t, 1, count)
+	}
+
+	// insert 5,5 6,6
+	sqlSink := &SQLSinkConnector{}
+	require.NoError(t, sqlSink.Provision(ctx, map[string]interface{}{
+		"dburl": dburl,
+		"table": tableName,
+	}))
+	require.NoError(t, sqlSink.Connect(ctx))
+	require.NoError(t, sqlSink.collectList(ctx, []map[string]any{
+		{
+			"a": 5,
+			"b": 5,
+		},
+		{
+			"a": 6,
+			"b": 6,
+		},
+	}))
+
+	rows, err := sqlSink.conn.GetDB().Query("select a,b from t where a >=5 and b >=5")
+	require.NoError(t, err)
+	var got [][]int
+	for rows.Next() {
+		var a int
+		var b int
+		require.NoError(t, rows.Scan(&a, &b))
+		got = append(got, []int{a, b})
+	}
+	sqlSink.Close(ctx)
+	require.Equal(t, [][]int{{5, 5}, {6, 6}}, got)
+
+	// insert 7,7 8,8
+	sqlSink = &SQLSinkConnector{}
+	require.NoError(t, sqlSink.Provision(ctx, map[string]interface{}{
+		"dburl":        dburl,
+		"table":        tableName,
+		"rowKindField": "action",
+		"keyField":     "a",
+		"fields":       []string{"a", "b"},
+	}))
+	require.NoError(t, sqlSink.Connect(ctx))
+	require.NoError(t, sqlSink.collectList(ctx, []map[string]any{
+		{
+			"a":      7,
+			"b":      7,
+			"action": "insert",
+		},
+		{
+			"a":      8,
+			"b":      8,
+			"action": "insert",
+		},
+	}))
+
+	got = [][]int{}
+	rows, err = sqlSink.conn.GetDB().Query("select a,b from t where a >=7 and b >=7")
+	require.NoError(t, err)
+	for rows.Next() {
+		var a int
+		var b int
+		require.NoError(t, rows.Scan(&a, &b))
+		got = append(got, []int{a, b})
+	}
+	sqlSink.Close(ctx)
+	require.Equal(t, [][]int{{7, 7}, {8, 8}}, got)
+}
+
+func TestSQLProvisionErr(t *testing.T) {
+	ctx := mockContext.NewMockContext("1", "2")
+	sqlSink := &SQLSinkConnector{}
+	require.Error(t, sqlSink.Provision(ctx, map[string]interface{}{}))
+	require.Error(t, sqlSink.Provision(ctx, map[string]interface{}{
+		"dburl": "123",
+	}))
+	require.Error(t, sqlSink.Provision(ctx, map[string]interface{}{
+		"dburl":        "123",
+		"table":        "123",
+		"rowKindField": "123",
+	}))
+}
+
+func TestSQLSinkConfigKV(t *testing.T) {
+	ctx := mockContext.NewMockContext("1", "2")
+	config := &sqlSinkConfig{
+		Fields: []string{"a"},
+	}
+	_, _, err := config.getKeyValues(ctx, nil)
+	require.Error(t, err)
+
+	keys, values, err := config.getKeyValues(ctx, map[string]interface{}{
+		"a": "value",
+	})
+	require.NoError(t, err)
+	require.Equal(t, []string{"a"}, keys)
+	require.Equal(t, []string{"'value'"}, values)
+
+	config = &sqlSinkConfig{
+		Fields: []string{"a"},
+	}
+	keys, values, err = config.getKeyValues(ctx, map[string]interface{}{
+		"b": "value",
+	})
+	require.NoError(t, err)
+	require.NoError(t, err)
+	require.Equal(t, []string{"a"}, keys)
+	require.Equal(t, []string{"NULL"}, values)
+
+	config = &sqlSinkConfig{}
+	keys, values, err = config.getKeyValues(ctx, map[string]interface{}{
+		"a": "value",
+	})
+	require.NoError(t, err)
+	require.NoError(t, err)
+	require.Equal(t, []string{"a"}, keys)
+	require.Equal(t, []string{"'value'"}, values)
+}
+
+func TestSQLSinkAction(t *testing.T) {
+	connection.InitConnectionManager4Test()
+	ctx := mockContext.NewMockContext("1", "2")
+	s, err := testx.SetupEmbeddedMysqlServer(address, port)
+	require.NoError(t, err)
+	defer func() {
+		s.Close()
+	}()
+	tableName := "t"
+	dburl := fmt.Sprintf("mysql://root:@%v:%v/test", address, port)
+	sqlSink := &SQLSinkConnector{}
+	require.NoError(t, sqlSink.Provision(ctx, map[string]interface{}{
+		"dburl":        dburl,
+		"table":        tableName,
+		"fields":       []string{"a", "b"},
+		"rowKindField": "action",
+		"keyField":     "a",
+	}))
+	require.NoError(t, sqlSink.Connect(ctx))
+	// update
+	require.NoError(t, sqlSink.collect(ctx, map[string]any{
+		"a":      1,
+		"b":      2,
+		"action": "update",
+	}))
+	got := [][]int{}
+	rows, err := sqlSink.conn.GetDB().Query("select a,b from t where a=1")
+	require.NoError(t, err)
+	for rows.Next() {
+		var a int
+		var b int
+		require.NoError(t, rows.Scan(&a, &b))
+		got = append(got, []int{a, b})
+	}
+	require.Equal(t, [][]int{{1, 2}}, got)
+	// delete
+	require.NoError(t, sqlSink.collect(ctx, map[string]any{
+		"a":      1,
+		"b":      2,
+		"action": "delete",
+	}))
+	got = [][]int{}
+	rows, err = sqlSink.conn.GetDB().Query("select a,b from t where a=1")
+	require.NoError(t, err)
+	for rows.Next() {
+		var a int
+		var b int
+		require.NoError(t, rows.Scan(&a, &b))
+		got = append(got, []int{a, b})
+	}
+	require.Equal(t, [][]int{}, got)
+
+	// invalid
+	require.Error(t, sqlSink.collect(ctx, map[string]any{
+		"a":      1,
+		"b":      2,
+		"action": "mock",
+	}))
+	require.NoError(t, sqlSink.Close(ctx))
+}
diff --git a/internal/io/sql/source_test.go b/internal/io/sql/source_test.go
index dd53a658..e69d3596 100644
--- a/internal/io/sql/source_test.go
+++ b/internal/io/sql/source_test.go
@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//go:build full
+// +build full
+
 package sql
 
 import (
diff --git a/internal/server/bump/bump3_test.go b/internal/server/bump/bump3_test.go
index beddaa64..7c3187fe 100644
--- a/internal/server/bump/bump3_test.go
+++ b/internal/server/bump/bump3_test.go
@@ -27,6 +27,19 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/store"
 )
 
+func TestRewriteSinkConf(t *testing.T) {
+	conf.IsTesting = true
+	require.NoError(t, conf.WriteCfgIntoKVStorage("sinks", "sql", "conf1", map[string]interface{}{
+		"url": "123",
+	}))
+	require.NoError(t, rewriteSQLSinkConfiguration())
+	d, err := conf.GetCfgFromKVStorage("sinks", "sql", "conf1")
+	require.NoError(t, err)
+	v, ok := d["sinks.sql.conf1"]["dburl"]
+	require.True(t, ok)
+	require.Equal(t, "123", v)
+}
+
 func TestRewriteSQLConf(t *testing.T) {
 	testcases := []struct {
 		oldCfg   *OriginSqlSourceCfg
diff --git a/internal/topo/node/join_align_node_test.go b/internal/topo/node/join_align_node_test.go
new file mode 100644
index 00000000..85732c60
--- /dev/null
+++ b/internal/topo/node/join_align_node_test.go
@@ -0,0 +1,187 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package node
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestAlignTable(t *testing.T) {
+	tests := []struct {
+		name string
+		in   []any
+		out  []any
+	}{
+		{
+			name: "stream send",
+			in: []any{
+				&xsql.Tuple{
+					Emitter: "stream1",
+					Message: map[string]any{"id": 1, "a": 1},
+				},
+				&xsql.Tuple{
+					Emitter: "table1",
+					Message: map[string]any{"id": 1, "t1": "data"},
+				},
+				&xsql.Tuple{
+					Emitter: "stream1",
+					Message: map[string]any{"id": 1, "a": 2},
+				},
+				"unknown",
+			},
+			out: []any{
+				&xsql.WindowTuples{
+					Content: []xsql.Row{
+						&xsql.Tuple{
+							Emitter: "stream1",
+							Message: map[string]any{"id": 1, "a": 1},
+						},
+					},
+				},
+				&xsql.WindowTuples{
+					Content: []xsql.Row{
+						&xsql.Tuple{
+							Emitter: "stream1",
+							Message: map[string]any{"id": 1, "a": 2},
+						},
+						&xsql.Tuple{
+							Emitter: "table1",
+							Message: map[string]any{"id": 1, "t1": "data"},
+						},
+					},
+				},
+				errors.New("run JoinAlignNode error: invalid input type but got string(unknown)"),
+			},
+		},
+		{
+			name: "retain exceed",
+			in: []any{
+				&xsql.Tuple{
+					Emitter: "table1",
+					Message: map[string]any{"id": 1, "t1": "data2"},
+				},
+				&xsql.Tuple{
+					Emitter: "table1",
+					Message: map[string]any{"id": 1, "t1": "data3"},
+				},
+				&xsql.Tuple{
+					Emitter: "table1",
+					Message: map[string]any{"id": 1, "t1": "data4"},
+				},
+				&xsql.Tuple{
+					Emitter: "stream1",
+					Message: map[string]any{"id": 1, "a": 3},
+				},
+			},
+			out: []any{
+				&xsql.WindowTuples{
+					Content: []xsql.Row{
+						&xsql.Tuple{
+							Emitter: "stream1",
+							Message: map[string]any{"id": 1, "a": 3},
+						},
+						&xsql.Tuple{
+							Emitter: "table1",
+							Message: map[string]any{"id": 1, "t1": "data3"},
+						},
+						&xsql.Tuple{
+							Emitter: "table1",
+							Message: map[string]any{"id": 1, "t1": "data4"},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "multiple table",
+			in: []any{
+				&xsql.Tuple{
+					Emitter: "table2",
+					Message: map[string]any{"id": 1, "t2": "dd1"},
+				},
+				&xsql.Tuple{
+					Emitter: "table1",
+					Message: map[string]any{"id": 1, "t1": "data5"},
+				},
+				&xsql.Tuple{
+					Emitter: "table2",
+					Message: map[string]any{"id": 1, "t2": "dd2"},
+				},
+				&xsql.WindowTuples{
+					Content: []xsql.Row{
+						&xsql.Tuple{
+							Emitter: "stream1",
+							Message: map[string]any{"id": 1, "a": 4},
+						},
+					},
+				},
+			},
+			out: []any{
+				&xsql.WindowTuples{
+					Content: []xsql.Row{
+						&xsql.Tuple{
+							Emitter: "stream1",
+							Message: map[string]any{"id": 1, "a": 4},
+						},
+						&xsql.Tuple{
+							Emitter: "table1",
+							Message: map[string]any{"id": 1, "t1": "data5"},
+						},
+						&xsql.Tuple{
+							Emitter: "table2",
+							Message: map[string]any{"id": 1, "t2": "dd1"},
+						},
+						&xsql.Tuple{
+							Emitter: "table2",
+							Message: map[string]any{"id": 1, "t2": "dd2"},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			n, e := NewJoinAlignNode("align", []string{"table1", "table2"}, []int{2, 9999}, &def.RuleOption{
+				SendError: true,
+			})
+			assert.NoError(t, e)
+			out := make(chan any, 100)
+			e = n.AddOutput(out, "test")
+			assert.NoError(t, e)
+			ctx := mockContext.NewMockContext("test", "test")
+			errCh := make(chan error)
+			n.Exec(ctx, errCh)
+			defer n.Close()
+			for _, in := range tt.in {
+				n.input <- in
+			}
+			r := make([]any, 0, len(tt.out))
+			for i := 0; i < len(tt.out); i++ {
+				rr := <-out
+				r = append(r, rr)
+			}
+			assert.Equal(t, tt.out, r)
+		})
+	}
+}
diff --git a/internal/topo/node/source_node_test.go b/internal/topo/node/source_node_test.go
index 8d3fda45..3a4bf77b 100644
--- a/internal/topo/node/source_node_test.go
+++ b/internal/topo/node/source_node_test.go
@@ -127,13 +127,6 @@ func TestNewError(t *testing.T) {
 	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,
diff --git a/internal/topo/operator/table_processor_test.go b/internal/topo/operator/table_processor_test.go
deleted file mode 100644
index b70e4ed2..00000000
--- a/internal/topo/operator/table_processor_test.go
+++ /dev/null
@@ -1,147 +0,0 @@
-// Copyright 2021-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 operator
-
-import (
-	"encoding/json"
-	"fmt"
-	"reflect"
-	"testing"
-
-	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
-	"github.com/lf-edge/ekuiper/v2/internal/xsql"
-	"github.com/lf-edge/ekuiper/v2/pkg/ast"
-)
-
-func TestTableProcessor_Apply(t *testing.T) {
-	tests := []struct {
-		stmt   *ast.StreamStmt
-		data   []byte
-		result interface{}
-	}{
-		{
-			stmt: &ast.StreamStmt{
-				Name: ast.StreamName("demo"),
-				StreamFields: []ast.StreamField{
-					{Name: "a", FieldType: &ast.ArrayType{
-						Type: ast.STRUCT,
-						FieldType: &ast.RecType{
-							StreamFields: []ast.StreamField{
-								{Name: "b", FieldType: &ast.BasicType{Type: ast.STRINGS}},
-							},
-						},
-					}},
-				},
-			},
-			data: []byte(`[{"a": [{"b" : "hello1"}, {"b" : "hello2"}]},{"a": [{"b" : "hello2"}, {"b" : "hello3"}]},{"a": [{"b" : "hello3"}, {"b" : "hello4"}]}]`),
-			result: &xsql.WindowTuples{
-				Content: []xsql.Row{
-					&xsql.Tuple{
-						Message: xsql.Message{
-							"a": []interface{}{
-								map[string]interface{}{"b": "hello1"},
-								map[string]interface{}{"b": "hello2"},
-							},
-						},
-						Emitter: "demo",
-					},
-					&xsql.Tuple{
-						Message: xsql.Message{
-							"a": []interface{}{
-								map[string]interface{}{"b": "hello2"},
-								map[string]interface{}{"b": "hello3"},
-							},
-						},
-						Emitter: "demo",
-					},
-					&xsql.Tuple{
-						Message: xsql.Message{
-							"a": []interface{}{
-								map[string]interface{}{"b": "hello3"},
-								map[string]interface{}{"b": "hello4"},
-							},
-						},
-						Emitter: "demo",
-					},
-				},
-			},
-		}, {
-			stmt: &ast.StreamStmt{
-				Name:         ast.StreamName("demo"),
-				StreamFields: nil,
-			},
-			data: []byte(`[{"a": {"b" : "hello", "c": {"d": 35.2}}},{"a": {"b" : "world", "c": {"d": 65.2}}}]`),
-			result: &xsql.WindowTuples{
-				Content: []xsql.Row{
-					&xsql.Tuple{
-						Message: xsql.Message{
-							"a": map[string]interface{}{
-								"b": "hello",
-								"c": map[string]interface{}{
-									"d": 35.2,
-								},
-							},
-						},
-						Emitter: "demo",
-					},
-					&xsql.Tuple{
-						Message: xsql.Message{
-							"a": map[string]interface{}{
-								"b": "world",
-								"c": map[string]interface{}{
-									"d": 65.2,
-								},
-							},
-						},
-						Emitter: "demo",
-					},
-				},
-			},
-		},
-	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-
-	defer conf.CloseLogger()
-	contextLogger := conf.Log.WithField("rule", "TestPreprocessor_Apply")
-	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	for i, tt := range tests {
-		pp := &TableProcessor{isBatchInput: true, emitterName: "demo", checkSchema: true}
-		pp.streamFields = tt.stmt.StreamFields.ToJsonSchema()
-		pp.output = &xsql.WindowTuples{
-			Content: make([]xsql.Row, 0),
-		}
-
-		var dm []map[string]interface{}
-		if e := json.Unmarshal(tt.data, &dm); e != nil {
-			t.Log(e)
-			t.Fail()
-		} else {
-			fv, afv := xsql.NewFunctionValuersForOp(nil)
-			for _, m := range dm {
-				pp.Apply(ctx, &xsql.Tuple{
-					Emitter: "demo",
-					Message: m,
-				}, fv, afv)
-			}
-
-			result := pp.Apply(ctx, &xsql.Tuple{}, fv, afv)
-			if !reflect.DeepEqual(tt.result, result) {
-				t.Errorf("%d. result mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.result, result)
-			}
-		}
-
-	}
-}
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index a9b163c6..c7aff229 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -289,6 +289,7 @@ func Test_createLogicalPlan(t *testing.T) {
 											},
 										},
 										Emitters: []string{"tableInPlanner"},
+										Sizes:    []int{MaxRetainSize},
 									}.Init(),
 								},
 							},
@@ -1259,6 +1260,7 @@ func Test_createLogicalPlan(t *testing.T) {
 											},
 										},
 										Emitters: []string{"tableInPlanner"},
+										Sizes:    []int{MaxRetainSize},
 									}.Init(),
 								},
 							},
@@ -1353,6 +1355,7 @@ func Test_createLogicalPlan(t *testing.T) {
 											},
 										},
 										Emitters: []string{"tableInPlanner"},
+										Sizes:    []int{MaxRetainSize},
 									}.Init(),
 								},
 							},
@@ -1528,6 +1531,7 @@ func Test_createLogicalPlan(t *testing.T) {
 											},
 										},
 										Emitters: []string{"tableInPlanner"},
+										Sizes:    []int{MaxRetainSize},
 									}.Init(),
 								},
 							},
@@ -2742,30 +2746,28 @@ func Test_createLogicalPlan(t *testing.T) {
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 
-	for i, tt := range tests {
-		stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
-		if err != nil {
-			t.Errorf("%d. %q: error compile sql: %s\n", i, tt.sql, err)
-			continue
-		}
-		p, err := createLogicalPlan(stmt, &def.RuleOption{
-			IsEventTime:        false,
-			LateTol:            0,
-			Concurrency:        0,
-			BufferLength:       0,
-			SendMetaToSink:     false,
-			Qos:                0,
-			CheckpointInterval: 0,
-			SendError:          true,
-		}, kv)
-		if !reflect.DeepEqual(tt.err, testx.Errstring(err)) {
-			t.Errorf("%d. %v: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.sql, tt.err, err)
-		} else {
-			ok := assert.Equal(t, tt.p, p, "%d plan mismatch %s", i)
-			if !ok {
-				t.Errorf("%d. %q\n\nstmt mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, render.AsCode(tt.p), render.AsCode(p))
+	for _, tt := range tests {
+		t.Run(tt.sql, func(t *testing.T) {
+			stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
+			if !assert.NoError(t, err) {
+				return
 			}
-		}
+			p, err := createLogicalPlan(stmt, &def.RuleOption{
+				IsEventTime:        false,
+				LateTol:            0,
+				Concurrency:        0,
+				BufferLength:       0,
+				SendMetaToSink:     false,
+				Qos:                0,
+				CheckpointInterval: 0,
+				SendError:          true,
+			}, kv)
+			if tt.err != "" {
+				assert.EqualError(t, err, tt.err)
+			} else {
+				assert.Equal(t, tt.p, p)
+			}
+		})
 	}
 }
 
@@ -3383,6 +3385,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 											},
 										},
 										Emitters: []string{"tableInPlanner"},
+										Sizes:    []int{MaxRetainSize},
 									}.Init(),
 								},
 							},
@@ -3474,6 +3477,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 											},
 										},
 										Emitters: []string{"tableInPlanner"},
+										Sizes:    []int{MaxRetainSize},
 									}.Init(),
 								},
 							},
@@ -3643,6 +3647,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 											},
 										},
 										Emitters: []string{"tableInPlanner"},
+										Sizes:    []int{MaxRetainSize},
 									}.Init(),
 								},
 							},
diff --git a/internal/topo/topotest/mock_topo.go b/internal/topo/topotest/mock_topo.go
index 565e01e1..03b8c22e 100644
--- a/internal/topo/topotest/mock_topo.go
+++ b/internal/topo/topotest/mock_topo.go
@@ -124,13 +124,14 @@ func doRuleTestWithResultFunc(t *testing.T, tests []RuleTest, opt *def.RuleOptio
 			limit := len(tt.R)
 			consumer := pubsub.CreateSub(id, nil, id, limit)
 			conf.Log.Debugf("test create memory sub %s", id)
-			ticker := time.After(1000 * time.Second)
+			ticker := time.After(10 * time.Second)
 			sinkResult := make([]any, 0, limit)
 		outerloop:
 			for {
 				select {
-				case <-errCh:
-					conf.Log.Debugf("test %s receive error signal", id)
+				// the done signal is sent when all sources are EOF. The mock source is bounded so this will be triggered.
+				case err := <-errCh:
+					conf.Log.Debugf("test %s receive error signal: %v", id, err)
 					tp.Cancel()
 					break outerloop
 				case tuple := <-consumer:
@@ -379,7 +380,7 @@ func HandleStream(createOrDrop bool, names []string, t *testing.T) {
 					name STRING,
 					size BIGINT,
 					id BIGINT
-				) WITH (DATASOURCE="lookup.json", FORMAT="json", CONF_KEY="test");`
+				) WITH (DATASOURCE="lookup.json", FORMAT="json", CONF_KEY="test", STRICT_VALIDATION="true");`
 			case "helloStr":
 				sql = `CREATE STREAM helloStr (name string) WITH (DATASOURCE="helloStr", TYPE="mock", FORMAT="JSON")`
 			case "commands":
diff --git a/internal/topo/topotest/rule_test.go b/internal/topo/topotest/rule_test.go
index 2d438d6a..189c56ee 100644
--- a/internal/topo/topotest/rule_test.go
+++ b/internal/topo/topotest/rule_test.go
@@ -966,6 +966,44 @@ func TestSingleSQL(t *testing.T) {
 				},
 			},
 		},
+		{ // Need to move test/lookup.json to data/lookup.json
+			Name: `TestSingleSQLRule10`,
+			Sql:  "SELECT * FROM demo INNER JOIN table1 on demo.ts = table1.id",
+			R: [][]map[string]interface{}{
+				{{
+					"id":    int64(1541152486013),
+					"name":  "name1",
+					"color": "red",
+					"size":  3,
+					"ts":    1541152486013,
+				}},
+				{{
+					"id":    int64(1541152487632),
+					"name":  "name2",
+					"color": "blue",
+					"size":  2,
+					"ts":    1541152487632,
+				}},
+				{{
+					"id":    int64(1541152489252),
+					"name":  "name3",
+					"color": "red",
+					"size":  1,
+					"ts":    1541152489252,
+				}},
+			},
+			W: 15,
+			M: map[string]interface{}{
+				"sink_memory_0_0_records_in_total":  int64(3),
+				"sink_memory_0_0_records_out_total": int64(3),
+
+				"source_demo_0_records_in_total":  int64(5),
+				"source_demo_0_records_out_total": int64(5),
+
+				"source_table1_0_records_in_total":  int64(1),
+				"source_table1_0_records_out_total": int64(1),
+			},
+		},
 		{
 			Name: `TestSingleSQLRule11`,
 			Sql:  "SELECT device FROM demo INNER JOIN demoTable on demo.ts = demoTable.ts",
@@ -994,6 +1032,41 @@ func TestSingleSQL(t *testing.T) {
 				"source_demoTable_0_records_out_total": int64(5),
 			},
 		},
+		{
+			Name: `TestSingleSQLRule12`,
+			Sql:  "SELECT demo.ts as demoTs, table1.id as table1Id FROM demo INNER JOIN table1 on demoTs = table1Id",
+			R: [][]map[string]interface{}{
+				{{
+					"table1Id": int64(1541152486013),
+					"demoTs":   1541152486013,
+				}},
+				{{
+					"table1Id": int64(1541152487632),
+					"demoTs":   1541152487632,
+				}},
+				{{
+					"table1Id": int64(1541152489252),
+					"demoTs":   1541152489252,
+				}},
+			},
+			W: 15,
+			M: map[string]interface{}{
+				"op_5_join_aligner_0_records_in_total":  int64(8),
+				"op_5_join_aligner_0_records_out_total": int64(5),
+
+				"op_6_join_0_records_in_total":  int64(5),
+				"op_6_join_0_records_out_total": int64(3),
+
+				"sink_memory_0_0_records_in_total":  int64(3),
+				"sink_memory_0_0_records_out_total": int64(3),
+
+				"source_demo_0_records_in_total":  int64(5),
+				"source_demo_0_records_out_total": int64(5),
+
+				"source_table1_0_records_in_total":  int64(1),
+				"source_table1_0_records_out_total": int64(1),
+			},
+		},
 		{
 			Name: `TestChanged13`,
 			Sql:  "SELECT changed_cols(\"tt_\", true, color, size) FROM demo",
diff --git a/internal/topo/topotest/window_rule_test.go b/internal/topo/topotest/window_rule_test.go
index 344723e5..4c7631ae 100644
--- a/internal/topo/topotest/window_rule_test.go
+++ b/internal/topo/topotest/window_rule_test.go
@@ -586,6 +586,42 @@ func TestWindow(t *testing.T) {
 				"op_2_window_0_records_out_total":  int64(5),
 			},
 		},
+		{
+			Name: `TestWindowRule11`,
+			Sql:  `SELECT color, name, window_start(), window_end() FROM demo INNER JOIN table1 on demo.ts = table1.id where demo.size > 2 and table1.size > 1 GROUP BY tumblingwindow(ss, 1)`,
+			R: [][]map[string]interface{}{
+				{{
+					"color":        "red",
+					"name":         "name1",
+					"window_start": int64(1541152486000),
+					"window_end":   int64(1541152487000),
+				}},
+			},
+			M: map[string]interface{}{
+				"op_2_window_0_exceptions_total":  int64(0),
+				"op_2_window_0_records_in_total":  int64(5),
+				"op_2_window_0_records_out_total": int64(3),
+
+				"op_6_join_aligner_0_records_in_total":  int64(6),
+				"op_6_join_aligner_0_records_out_total": int64(3),
+
+				"op_7_join_0_exceptions_total":  int64(0),
+				"op_7_join_0_records_in_total":  int64(3),
+				"op_7_join_0_records_out_total": int64(1),
+
+				"sink_memory_0_0_exceptions_total":  int64(0),
+				"sink_memory_0_0_records_in_total":  int64(1),
+				"sink_memory_0_0_records_out_total": int64(1),
+
+				"source_demo_0_exceptions_total":  int64(0),
+				"source_demo_0_records_in_total":  int64(5),
+				"source_demo_0_records_out_total": int64(5),
+
+				"source_table1_0_exceptions_total":  int64(0),
+				"source_table1_0_records_in_total":  int64(1),
+				"source_table1_0_records_out_total": int64(1),
+			},
+		},
 		{
 			Name: `TestWindowRule12`,
 			Sql:  `SELECT collect(size) as allSize FROM demo GROUP BY HOPPINGWINDOW(ss, 2, 1), color ORDER BY color`,