Repository: lf-edge/ekuiper
Pull request #2948 (2024-06-25 05:57:10)

fix(test): checkpoint tests

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2948?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 `74.46809%` with `24 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@2baffea`). [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/2948?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/topotest/mock\_topo.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2948?src=pr&el=tree&filepath=internal%2Ftopo%2Ftopotest%2Fmock_topo.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby90b3BvdGVzdC9tb2NrX3RvcG8uZ28=) | 74.47% | [18 Missing and 6 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2948?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    #2948   +/-   ##
============================================
  Coverage             ?   69.90%           
============================================
  Files                ?      311           
  Lines                ?    29839           
  Branches             ?        0           
============================================
  Hits                 ?    20857           
  Misses               ?     7178           
  Partials             ?     1804           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2948?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/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 {


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/topotest/checkpoint_test.go b/internal/topo/topotest/checkpoint_test.go
new file mode 100644
index 00000000..cbdb34ea
--- /dev/null
+++ b/internal/topo/topotest/checkpoint_test.go
@@ -0,0 +1,181 @@
+// 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 topotest
+
+import (
+	"testing"
+	"time"
+
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+// Full lifecycle test: Run window rule; trigger checkpoints by mock timer; restart rule; make sure the result is right;
+func TestCheckpoint(t *testing.T) {
+	streamList := []string{"demo"}
+	HandleStream(false, streamList, t)
+	tests := []RuleCheckpointTest{
+		{
+			RuleTest: RuleTest{
+				Name: `TestCheckpointRule1`,
+				Sql:  `SELECT * FROM demo GROUP BY HOPPINGWINDOW(ss, 2, 1)`,
+				R: [][]map[string]interface{}{
+					//{{ // 7000
+					//	"color": "red",
+					//	"size":  3,
+					//	"ts":    1541152486013,
+					//}, {
+					//	"color": "blue",
+					//	"size":  6,
+					//	"ts":    1541152486822,
+					//}},
+					{{ // restore in 8000
+						"color": "red",
+						"size":  3,
+						"ts":    1541152486013,
+					}, {
+						"color": "blue",
+						"size":  6,
+						"ts":    1541152486822,
+					}},
+					{{
+						"color": "blue",
+						"size":  2,
+						"ts":    1541152487632,
+					}},
+					{
+						{
+							"color": "blue",
+							"size":  2,
+							"ts":    1541152487632,
+						},
+						{ // 9000
+							"color": "yellow",
+							"size":  4,
+							"ts":    1541152488442,
+						},
+					},
+				},
+				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(3),
+					"source_demo_0_records_out_total": int64(3),
+
+					"op_2_window_0_records_in_total":  int64(3),
+					"op_2_window_0_records_out_total": int64(3),
+				},
+			},
+			PauseSize: 3,
+			Cc:        2,
+		},
+	}
+	HandleStream(true, streamList, t)
+	options := []*def.RuleOption{
+		{ // Need to clean up checkpoint cache before running twice
+			//	BufferLength:       100,
+			//	Qos:                def.AtLeastOnce,
+			//	CheckpointInterval: cast.DurationConf(600 * time.Millisecond),
+			//	SendError:          true,
+			//}, {
+			BufferLength:       100,
+			Qos:                def.ExactlyOnce,
+			CheckpointInterval: cast.DurationConf(600 * time.Millisecond),
+			SendError:          true,
+		},
+	}
+	for j, opt := range options {
+		DoCheckpointRuleTest(t, tests, opt, j)
+	}
+}
+
+func TestTableJoinCheckpoint(t *testing.T) {
+	streamList := []string{"demo", "table1"}
+	HandleStream(false, streamList, t)
+	tests := []RuleCheckpointTest{
+		{
+			RuleTest: RuleTest{
+				Name: `TestCheckpointRule2`,
+				Sql:  `SELECT * FROM demo INNER JOIN table1 on demo.ts = table1.id`,
+				R: [][]map[string]interface{}{
+					//{{
+					//	"id":    float64(1541152486013),
+					//	"name":  "name1",
+					//	"color": "red",
+					//	"size":  float64(3),
+					//	"ts":    float64(1541152486013),
+					//}},
+					{
+						{
+							"id":    int64(1541152487632),
+							"name":  "name2",
+							"color": "blue",
+							"size":  2,
+							"ts":    1541152487632,
+						},
+						{
+							"id":    int64(1541152487632),
+							"name":  "name2",
+							"color": "blue",
+							"size":  2,
+							"ts":    1541152487632,
+						},
+					},
+					{ // One from cache, one from newly sent
+						{
+							"id":    int64(1541152489252),
+							"name":  "name3",
+							"color": "red",
+							"size":  1,
+							"ts":    1541152489252,
+						},
+						{
+							"id":    int64(1541152489252),
+							"name":  "name3",
+							"color": "red",
+							"size":  1,
+							"ts":    1541152489252,
+						},
+					},
+				},
+				M: map[string]interface{}{
+					"sink_memory_0_0_records_in_total":  int64(2),
+					"sink_memory_0_0_records_out_total": int64(2),
+
+					"source_demo_0_records_in_total":  int64(3),
+					"source_demo_0_records_out_total": int64(3),
+
+					"source_table1_0_records_in_total":  int64(1),
+					"source_table1_0_records_out_total": int64(1),
+				},
+			},
+			PauseSize: 3,
+			Cc:        2,
+		},
+	}
+	HandleStream(true, streamList, t)
+	options := []*def.RuleOption{
+		{
+			BufferLength:       100,
+			Qos:                def.AtLeastOnce,
+			CheckpointInterval: cast.DurationConf(600 * time.Millisecond),
+			SendError:          true,
+		},
+	}
+	for j, opt := range options {
+		DoCheckpointRuleTest(t, tests, opt, j)
+	}
+}
diff --git a/internal/topo/topotest/mock_topo.go b/internal/topo/topotest/mock_topo.go
index 03b8c22e..67e98435 100644
--- a/internal/topo/topotest/mock_topo.go
+++ b/internal/topo/topotest/mock_topo.go
@@ -29,7 +29,6 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/topo"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/planner"
-	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mocknode"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
@@ -204,12 +203,8 @@ func CompareMetrics(tp *topo.Topo, m map[string]interface{}) (err error) {
 }
 
 func sendData(dataLength int, datas [][]*xsql.Tuple, tp *topo.Topo, postleap int, wait int) {
-	// Send data and move time
-	mockClock := mockclock.GetMockClock()
-	// Set the current time
-	mockClock.Add(0)
 	// TODO assume multiple data source send the data in order and has the same length
-	mockClock.Set(datas[0][0].Timestamp.Add(-time.Second))
+	conf.Log.Infof("Send clock init to %d", timex.GetNowInMilli())
 	for i := 0; i < dataLength; i++ {
 		// wait for table to load
 		time.Sleep(100 * time.Millisecond)
@@ -221,8 +216,8 @@ func sendData(dataLength int, datas [][]*xsql.Tuple, tp *topo.Topo, postleap int
 				if d[i].Timestamp.Before(n) {
 					n = d[i].Timestamp
 				}
-				mockClock.Set(n)
-				conf.Log.Debugf("Clock set to %d", timex.GetNowInMilli())
+				timex.Set(n.UnixMilli())
+				conf.Log.Infof("Clock set to %d", timex.GetNowInMilli())
 				time.Sleep(1 * time.Millisecond)
 			}
 			select {
@@ -232,13 +227,12 @@ func sendData(dataLength int, datas [][]*xsql.Tuple, tp *topo.Topo, postleap int
 			}
 		}
 	}
-	mockClock.Add(time.Duration(postleap) * time.Millisecond)
-	conf.Log.Debugf("Clock add to %d", timex.GetNowInMilli())
+	timex.Add(time.Duration(postleap) * time.Millisecond)
+	conf.Log.Infof("Clock add to %d", timex.GetNowInMilli())
 }
 
 // create a test rule with memory sink
 func createTestRule(t *testing.T, id string, tt RuleTest, opt *def.RuleOption) ([][]*xsql.Tuple, int, *topo.Topo, <-chan error) {
-	mockclock.ResetClock(1541152486000)
 	// Create stream
 	var (
 		datas      [][]*xsql.Tuple
@@ -263,6 +257,11 @@ func createTestRule(t *testing.T, id string, tt RuleTest, opt *def.RuleOption) (
 			}
 		}
 	}
+	startTs := datas[0][0].Timestamp.UnixMilli() / 1000 * 1000
+	for startTs != timex.GetNowInMilli() {
+		timex.Set(startTs)
+		conf.Log.Infof("Init Clock to %d, and now is %d", startTs, timex.GetNowInMilli())
+	}
 	rule := &def.Rule{
 		Id:  id,
 		Sql: tt.Sql,
@@ -416,3 +415,122 @@ func HandleStream(createOrDrop bool, names []string, t *testing.T) {
 		}
 	}
 }
+
+type RuleCheckpointTest struct {
+	RuleTest
+	PauseSize int // Stop stream after sending pauseSize source to test checkpoint resume
+	Cc        int // checkpoint count when paused
+	// PauseMetric map[string]interface{} // The metric to check when paused
+}
+
+func DoCheckpointRuleTest(t *testing.T, tests []RuleCheckpointTest, opt *def.RuleOption, w int) {
+	fmt.Printf("The test bucket for option %d size is %d.\n\n", w, len(tests))
+	for i, tt := range tests {
+		t.Run(tt.Name, func(t *testing.T) {
+			id := strings.ReplaceAll(strings.ReplaceAll(t.Name(), "#", "_"), "/", "_")
+			conf.Log.Debugf("run test %s", id)
+			// Create the rule which sink to memory topic
+			datas, dataLength, tp, _ := createTestRule(t, id, tt.RuleTest, opt)
+			if tp == nil {
+				t.Errorf("topo is not created successfully")
+				return
+			}
+			// Send data with leaps
+			wait := tt.W
+			if wait == 0 {
+				if w > 0 {
+					wait = w
+				} else {
+					wait = 5
+				}
+			}
+			switch opt.Qos {
+			case def.ExactlyOnce:
+				wait *= 10
+			case def.AtLeastOnce:
+				wait *= 3
+			default:
+				// do nothing
+			}
+			var retry int
+			if opt.Qos > def.AtMostOnce {
+				for retry = 3; retry > 0; retry-- {
+					if tp.GetCoordinator() == nil || !tp.GetCoordinator().IsActivated() {
+						conf.Log.Debugf("waiting for coordinator ready %d\n", retry)
+						time.Sleep(10 * time.Millisecond)
+					} else {
+						break
+					}
+				}
+				if retry < 0 {
+					t.Error("coordinator timeout")
+					t.FailNow()
+				}
+			}
+			// Send async
+			go sendData(tt.PauseSize, datas, tp, 100, wait)
+			conf.Log.Debugf("Send first phase data done at %d", timex.GetNowInMilli())
+			// compare checkpoint count
+			time.Sleep(1 * time.Second)
+			for retry = 10; retry > 0; retry-- {
+				actual := tp.GetCoordinator().GetCompleteCount()
+				if tt.Cc == actual {
+					break
+				}
+				conf.Log.Debugf("check checkpointCount error at %d: %d\n", retry, actual)
+				time.Sleep(500 * time.Millisecond)
+			}
+			cc := tp.GetCoordinator().GetCompleteCount()
+			tp.Cancel()
+			if retry == 0 {
+				t.Errorf("%d-%d. checkpoint count\n\nresult mismatch:\n\nexp=%#v\n\ngot=%d\n\n", i, w, tt.Cc, cc)
+				return
+			} else if retry < 3 {
+				conf.Log.Debugf("try %d for checkpoint count\n", 4-retry)
+			}
+			time.Sleep(10 * time.Millisecond)
+			// resume stream
+			conf.Log.Debugf("Resume stream at %d", timex.GetNowInMilli())
+			errCh := tp.Open()
+			conf.Log.Debugf("After open stream at %d", timex.GetNowInMilli())
+			go sendData(dataLength-tt.PauseSize, [][]*xsql.Tuple{datas[0][tt.PauseSize:]}, tp, POSTLEAP, 10)
+			// Receive data
+			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)
+			sinkResult := make([]any, 0, limit)
+		outerloop:
+			for {
+				select {
+				case <-errCh:
+					conf.Log.Debugf("test %s receive error signal", id)
+					tp.Cancel()
+					break outerloop
+				case tuple := <-consumer:
+					sinkResult = append(sinkResult, tuple)
+					conf.Log.Debugf("test %s append result %v", id, tuple)
+				case <-ticker:
+					tp.Cancel()
+					assert.Fail(t, "timeout")
+					break outerloop
+				}
+			}
+		outloop:
+			for {
+				// Receive the last will if any
+				select {
+				case tuple := <-consumer:
+					sinkResult = append(sinkResult, tuple)
+					conf.Log.Debugf("test %s append result %v", id, tuple)
+				default:
+					break outloop
+				}
+			}
+			conf.Log.Debugf("test %s receive %d result", id, len(sinkResult))
+			assert.Equal(t, tt.R, CommonResultFunc(sinkResult))
+			err := CompareMetrics(tp, tt.M)
+			assert.NoError(t, err)
+		})
+	}
+}
diff --git a/internal/topo/topotest/mocknode/mock_source.go b/internal/topo/topotest/mocknode/mock_source.go
index e247bd0d..02dbc374 100644
--- a/internal/topo/topotest/mocknode/mock_source.go
+++ b/internal/topo/topotest/mocknode/mock_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.