Repository: lf-edge/ekuiper
Pull request #1420 (2022-09-29 09:02:39)

Make memory sink updatable


Patch:
diff --git a/etc/sources/memory.yaml b/etc/sources/memory.yaml
index 3a248136..e69de29b 100644
--- a/etc/sources/memory.yaml
+++ b/etc/sources/memory.yaml
@@ -1,2 +0,0 @@
-default:
-  index: ["id"]
\ No newline at end of file
diff --git a/internal/topo/memory/lookupsource.go b/internal/topo/memory/lookupsource.go
index c5bf4f05..02521a6e 100644
--- a/internal/topo/memory/lookupsource.go
+++ b/internal/topo/memory/lookupsource.go
@@ -15,9 +15,9 @@
 package memory
 
 import (
+	"fmt"
 	"github.com/lf-edge/ekuiper/internal/topo/memory/store"
 	"github.com/lf-edge/ekuiper/pkg/api"
-	"github.com/lf-edge/ekuiper/pkg/cast"
 	"regexp"
 	"strings"
 )
@@ -27,14 +27,14 @@ import (
 type lookupsource struct {
 	topic      string
 	topicRegex *regexp.Regexp
-	keys       []string
 	table      *store.Table
+	key        string
 }
 
 func (s *lookupsource) Open(ctx api.StreamContext) error {
-	ctx.GetLogger().Infof("lookup source %s is opened with keys %v", s.topic, s.keys)
+	ctx.GetLogger().Infof("lookup source %s is opened with key %v", s.topic, s.key)
 	var err error
-	s.table, err = store.Reg(s.topic, s.topicRegex, s.keys)
+	s.table, err = store.Reg(s.topic, s.topicRegex, s.key)
 	return err
 }
 
@@ -47,11 +47,14 @@ func (s *lookupsource) Configure(datasource string, props map[string]interface{}
 		}
 		s.topicRegex = r
 	}
-	if c, ok := props["index"]; ok {
-		if bl, err := cast.ToStringSlice(c, cast.CONVERT_SAMEKIND); err != nil {
-			s.keys = bl
+	if k, ok := props["key"]; ok {
+		if kk, ok := k.(string); ok {
+			s.key = kk
 		}
 	}
+	if s.key == "" {
+		return fmt.Errorf("key is required for lookup source")
+	}
 	return nil
 }
 
@@ -62,5 +65,5 @@ func (s *lookupsource) Lookup(ctx api.StreamContext, _ []string, keys []string,
 
 func (s *lookupsource) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("lookup source %s is closing", s.topic)
-	return store.Unreg(s.topic)
+	return store.Unreg(s.topic, s.key)
 }
diff --git a/internal/topo/memory/pubsub/manager.go b/internal/topo/memory/pubsub/manager.go
index 8bc2232b..7467726f 100644
--- a/internal/topo/memory/pubsub/manager.go
+++ b/internal/topo/memory/pubsub/manager.go
@@ -110,6 +110,18 @@ func RemovePub(topic string) {
 }
 
 func Produce(ctx api.StreamContext, topic string, data map[string]interface{}) {
+	doProduce(ctx, topic, api.NewDefaultSourceTuple(data, map[string]interface{}{"topic": topic}))
+}
+
+func ProduceUpdatable(ctx api.StreamContext, topic string, data map[string]interface{}, rowkind string, keyval interface{}) {
+	doProduce(ctx, topic, &UpdatableTuple{
+		DefaultSourceTuple: api.NewDefaultSourceTuple(data, map[string]interface{}{"topic": topic}),
+		Rowkind:            rowkind,
+		Keyval:             keyval,
+	})
+}
+
+func doProduce(ctx api.StreamContext, topic string, data api.SourceTuple) {
 	c, exists := pubTopics[topic]
 	if !exists {
 		return
@@ -120,7 +132,7 @@ func Produce(ctx api.StreamContext, topic string, data map[string]interface{}) {
 	// broadcast to all consumers
 	for name, out := range c.consumers {
 		select {
-		case out <- api.NewDefaultSourceTuple(data, map[string]interface{}{"topic": topic}):
+		case out <- data:
 			logger.Debugf("memory source broadcast from topic %s to %s done", topic, name)
 		case <-ctx.Done():
 			// rule stop so stop waiting
@@ -128,7 +140,6 @@ func Produce(ctx api.StreamContext, topic string, data map[string]interface{}) {
 			logger.Errorf("memory source topic %s drop message to %s", topic, name)
 		}
 	}
-
 }
 
 func ProduceError(ctx api.StreamContext, topic string, err error) {
diff --git a/internal/topo/memory/pubsub/tuple.go b/internal/topo/memory/pubsub/tuple.go
new file mode 100644
index 00000000..9384d8de
--- /dev/null
+++ b/internal/topo/memory/pubsub/tuple.go
@@ -0,0 +1,23 @@
+// Copyright 2022 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 pubsub
+
+import "github.com/lf-edge/ekuiper/pkg/api"
+
+type UpdatableTuple struct {
+	*api.DefaultSourceTuple
+	Rowkind string
+	Keyval  interface{}
+}
diff --git a/internal/topo/memory/sink.go b/internal/topo/memory/sink.go
index da3356d6..c97ec421 100644
--- a/internal/topo/memory/sink.go
+++ b/internal/topo/memory/sink.go
@@ -19,12 +19,23 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"github.com/lf-edge/ekuiper/pkg/cast"
 	"strings"
 )
 
+type config struct {
+	Topic        string `json:"topic"`
+	DataTemplate string `json:"dataTemplate"`
+	RowkindField string `json:"rowkindField"`
+	KeyField     string `json:"keyField"`
+}
+
 type sink struct {
 	topic        string
 	hasTransform bool
+	keyField     string
+	rowkindField string
 }
 
 func (s *sink) Open(ctx api.StreamContext) error {
@@ -34,19 +45,23 @@ func (s *sink) Open(ctx api.StreamContext) error {
 }
 
 func (s *sink) Configure(props map[string]interface{}) error {
-	if t, ok := props[pubsub.IdProperty]; ok {
-		if id, casted := t.(string); casted {
-			if strings.ContainsAny(id, "#+") {
-				return fmt.Errorf("invalid memory topic %s: wildcard found", id)
-			}
-			s.topic = id
-		} else {
-			return fmt.Errorf("can't cast value %s to string", t)
-		}
+	cfg := &config{}
+	err := cast.MapToStruct(props, cfg)
+	if err != nil {
+		return err
+	}
+	if strings.ContainsAny(cfg.Topic, "#+") {
+		return fmt.Errorf("invalid memory topic %s: wildcard found", cfg.Topic)
 	}
-	if _, ok := props["dataTemplate"]; ok {
+	s.topic = cfg.Topic
+	if cfg.DataTemplate != "" {
 		s.hasTransform = true
 	}
+	s.rowkindField = cfg.RowkindField
+	s.keyField = cfg.KeyField
+	if s.rowkindField != "" && s.keyField == "" {
+		return fmt.Errorf("keyField is required when rowkindField is set")
+	}
 	return nil
 }
 
@@ -68,14 +83,19 @@ func (s *sink) Collect(ctx api.StreamContext, data interface{}) error {
 		}
 		data = m
 	}
-
 	switch d := data.(type) {
 	case []map[string]interface{}:
 		for _, el := range d {
-			pubsub.Produce(ctx, topic, el)
+			err := s.publish(ctx, topic, el)
+			if err != nil {
+				return fmt.Errorf("fail to publish data %v for error %v", d, err)
+			}
 		}
 	case map[string]interface{}:
-		pubsub.Produce(ctx, topic, d)
+		err := s.publish(ctx, topic, d)
+		if err != nil {
+			return fmt.Errorf("fail to publish data %v for error %v", d, err)
+		}
 	default:
 		return fmt.Errorf("unrecognized format of %s", data)
 	}
@@ -87,3 +107,29 @@ func (s *sink) Close(ctx api.StreamContext) error {
 	pubsub.RemovePub(s.topic)
 	return nil
 }
+
+func (s *sink) publish(ctx api.StreamContext, topic string, el map[string]interface{}) error {
+	if s.rowkindField != "" {
+		c, ok := el[s.rowkindField]
+		var rowkind string
+		if !ok {
+			rowkind = ast.RowkindUpsert
+		} else {
+			rowkind, ok = c.(string)
+			if !ok {
+				return fmt.Errorf("rowkind field %s is not a string in data %v", s.rowkindField, el)
+			}
+			if rowkind != ast.RowkindInsert && rowkind != ast.RowkindUpdate && rowkind != ast.RowkindDelete && rowkind != ast.RowkindUpsert {
+				return fmt.Errorf("invalid rowkind %s", rowkind)
+			}
+		}
+		key, ok := el[s.keyField]
+		if !ok {
+			return fmt.Errorf("key field %s not found in data %v", s.keyField, el)
+		}
+		pubsub.ProduceUpdatable(ctx, topic, el, rowkind, key)
+	} else {
+		pubsub.Produce(ctx, topic, el)
+	}
+	return nil
+}
diff --git a/internal/topo/memory/store/db.go b/internal/topo/memory/store/db.go
index 821c8719..03d570c9 100644
--- a/internal/topo/memory/store/db.go
+++ b/internal/topo/memory/store/db.go
@@ -17,6 +17,7 @@ package store
 import (
 	"context"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"sync"
 )
@@ -39,21 +40,22 @@ func (tc *tableCount) Decrease() int {
 	defer tc.Unlock()
 	tc.count--
 	if tc.count < 0 {
-		fmt.Errorf("Table count is less than 0: %d", tc.count)
+		conf.Log.Errorf("Table count is less than 0: %d", tc.count)
 	}
 	return tc.count
 }
 
 type database struct {
 	sync.RWMutex
-	tables map[string]*tableCount // topic: table
+	tables map[string]*tableCount // topic_key: table
 }
 
 // getTable return the table of the topic.
-func (db *database) getTable(topic string) (*Table, bool) {
+func (db *database) getTable(topic string, key string) (*Table, bool) {
 	db.RLock()
 	defer db.RUnlock()
-	tc, ok := db.tables[topic]
+	tableId := fmt.Sprintf("%s_%s", topic, key)
+	tc, ok := db.tables[tableId]
 	if ok {
 		return tc.t, true
 	} else {
@@ -65,19 +67,20 @@ func (db *database) getTable(topic string) (*Table, bool) {
 // If the table already exists, return the existing table;
 // otherwise, create a new table and return it.
 // The second argument is to indicate if the table is newly created
-func (db *database) addTable(topic string, keys []string) (*Table, bool) {
+func (db *database) addTable(topic string, key string) (*Table, bool) {
 	db.Lock()
 	defer db.Unlock()
-	tc, ok := db.tables[topic]
+	tableId := fmt.Sprintf("%s_%s", topic, key)
+	tc, ok := db.tables[tableId]
 	if ok {
 		tc.Increase()
 	} else {
-		t := createTable(keys)
+		t := createTable(topic, key)
 		tc = &tableCount{
 			count: 1,
 			t:     t,
 		}
-		db.tables[topic] = tc
+		db.tables[tableId] = tc
 	}
 	return tc.t, !ok
 }
@@ -85,79 +88,81 @@ func (db *database) addTable(topic string, keys []string) (*Table, bool) {
 // dropTable drop the table of the topic/values
 // stops to accumulate job
 // deletes the cache data
-func (db *database) dropTable(topic string) error {
+func (db *database) dropTable(topic string, key string) error {
+	tableId := fmt.Sprintf("%s_%s", topic, key)
 	db.Lock()
 	defer db.Unlock()
-	if tc, ok := db.tables[topic]; ok {
+	if tc, ok := db.tables[tableId]; ok {
 		if tc.Decrease() == 0 {
 			if tc.t != nil && tc.t.cancel != nil {
 				tc.t.cancel()
 			}
-			delete(db.tables, topic)
+			delete(db.tables, tableId)
 		}
 		return nil
 	}
-	return fmt.Errorf("Table %s not found", topic)
+	return fmt.Errorf("Table %s not found", tableId)
 }
 
 // Table has one writer and multiple reader
 type Table struct {
 	sync.RWMutex
-	// datamap is the overall data
-	datamap  []api.SourceTuple
-	hasIndex bool
-	// indexes is the indexed data
-	indexes map[string]map[interface{}][]api.SourceTuple
+	topic string
+	key   string
+	// datamap is the overall data indexed by primary key
+	datamap map[interface{}]api.SourceTuple
 	cancel  context.CancelFunc
 }
 
-func createTable(keys []string) *Table {
-	t := &Table{}
-	if len(keys) > 0 {
-		t.indexes = make(map[string]map[interface{}][]api.SourceTuple, len(keys))
-		for _, k := range keys {
-			t.indexes[k] = make(map[interface{}][]api.SourceTuple)
-		}
-		t.hasIndex = true
-	}
+func createTable(topic string, key string) *Table {
+	t := &Table{topic: topic, key: key, datamap: make(map[interface{}]api.SourceTuple)}
 	return t
 }
 
 func (t *Table) add(value api.SourceTuple) {
 	t.Lock()
 	defer t.Unlock()
-	t.datamap = append(t.datamap, value)
-	for k, v := range t.indexes {
-		if val, ok := value.Message()[k]; ok {
-			if _, ok := v[val]; !ok {
-				v[val] = make([]api.SourceTuple, 0)
-			}
-			v[val] = append(v[val], value)
-		}
+	keyval, ok := value.Message()[t.key]
+	if !ok {
+		conf.Log.Errorf("add to table %s omitted, value not found for key %s", t.topic, t.key)
 	}
+	t.datamap[keyval] = value
+}
+
+func (t *Table) delete(key interface{}) {
+	t.Lock()
+	defer t.Unlock()
+	delete(t.datamap, key)
 }
 
 func (t *Table) Read(keys []string, values []interface{}) ([]api.SourceTuple, error) {
 	t.RLock()
 	defer t.RUnlock()
-	data := t.datamap
-	excludeKey := -1
-	if t.hasIndex {
-		// Find the first indexed key
+	// Find the primary key
+	var matched api.SourceTuple
+	for i, k := range keys {
+		if k == t.key {
+			matched = t.datamap[values[i]]
+		}
+	}
+	if matched != nil {
+		match := true
 		for i, k := range keys {
-			if d, ok := t.indexes[k]; ok {
-				data = d[values[i]]
-				excludeKey = i
+			if val, ok := matched.Message()[k]; !ok || val != values[i] {
+				match = false
+				break
 			}
 		}
+		if match {
+			return []api.SourceTuple{matched}, nil
+		} else {
+			return nil, nil
+		}
 	}
 	var result []api.SourceTuple
-	for _, v := range data {
+	for _, v := range t.datamap {
 		match := true
 		for i, k := range keys {
-			if i == excludeKey {
-				continue
-			}
 			if val, ok := v.Message()[k]; !ok || val != values[i] {
 				match = false
 				break
diff --git a/internal/topo/memory/store/store.go b/internal/topo/memory/store/store.go
index aee41187..6317db0b 100644
--- a/internal/topo/memory/store/store.go
+++ b/internal/topo/memory/store/store.go
@@ -19,13 +19,14 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
+	"github.com/lf-edge/ekuiper/pkg/ast"
 	"regexp"
 )
 
 // Reg registers a topic to save it to memory store
 // Create a new go routine to listen to the topic and save the data to memory
-func Reg(topic string, topicRegex *regexp.Regexp, keys []string) (*Table, error) {
-	t, isNew := db.addTable(topic, keys)
+func Reg(topic string, topicRegex *regexp.Regexp, key string) (*Table, error) {
+	t, isNew := db.addTable(topic, key)
 	if isNew {
 		go runTable(topic, topicRegex, t)
 	}
@@ -46,7 +47,17 @@ func runTable(topic string, topicRegex *regexp.Regexp, t *Table) {
 			if !opened { // exit go routine is not sync with drop table
 				return
 			}
-			t.add(v)
+			switch vv := v.(type) {
+			case *pubsub.UpdatableTuple:
+				switch vv.Rowkind {
+				case ast.RowkindInsert, ast.RowkindUpdate, ast.RowkindUpsert:
+					t.add(vv.DefaultSourceTuple)
+				case ast.RowkindDelete:
+					t.delete(vv.Keyval)
+				}
+			default:
+				t.add(v)
+			}
 			conf.Log.Debugf("receive data %v for %s", v, topic)
 		case <-ctx.Done():
 			return
@@ -55,7 +66,7 @@ func runTable(topic string, topicRegex *regexp.Regexp, t *Table) {
 }
 
 // Unreg unregisters a topic to remove it from memory store
-func Unreg(topic string) error {
+func Unreg(topic string, key string) error {
 	// Must be an atomic operation
-	return db.dropTable(topic)
+	return db.dropTable(topic, key)
 }
diff --git a/internal/topo/node/conf/source.go b/internal/topo/node/conf/source.go
index 689288e6..7d59a9be 100644
--- a/internal/topo/node/conf/source.go
+++ b/internal/topo/node/conf/source.go
@@ -54,6 +54,7 @@ func GetSourceConf(sourceType string, options *ast.Options) map[string]interface
 		f = "json"
 	}
 	props["format"] = strings.ToLower(f)
+	props["key"] = strings.ToLower(options.KEY)
 	conf.Log.Debugf("get conf for %s with conf key %s: %v", sourceType, confkey, printable(props))
 	return props
 }
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index f4d0babe..e9592fee 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -1461,6 +1461,9 @@ func (p *Parser) parseStreamOptions() (*ast.Options, error) {
 	} else {
 		return nil, fmt.Errorf("found %q, expect stream options.", lit)
 	}
+	if opts.KIND == ast.StreamKindLookup && opts.TYPE == "memory" && opts.KEY == "" {
+		return nil, fmt.Errorf("Option \"key\" is required for memory lookup table.")
+	}
 	return opts, nil
 }
 
diff --git a/pkg/ast/statement.go b/pkg/ast/statement.go
index 3d4c0ed1..394f140f 100644
--- a/pkg/ast/statement.go
+++ b/pkg/ast/statement.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 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.
@@ -189,3 +189,10 @@ type SortField struct {
 type SortFields []SortField
 
 func (d SortFields) node() {}
+
+const (
+	RowkindInsert = "insert"
+	RowkindUpdate = "update"
+	RowkindUpsert = "upsert"
+	RowkindDelete = "delete"
+)


Test patch:
diff --git a/internal/processor/stream_test.go b/internal/processor/stream_test.go
index 059e9e68..08584b3f 100644
--- a/internal/processor/stream_test.go
+++ b/internal/processor/stream_test.go
@@ -205,8 +205,8 @@ func TestTableProcessor(t *testing.T) {
 func TestTableList(t *testing.T) {
 	p := NewStreamProcessor()
 	p.ExecStmt(`CREATE TABLE tt1 () WITH (DATASOURCE="users", FORMAT="JSON", KIND="scan")`)
-	p.ExecStmt(`CREATE TABLE tt2 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KIND="lookup")`)
-	p.ExecStmt(`CREATE TABLE tt3 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KIND="lookup")`)
+	p.ExecStmt(`CREATE TABLE tt2 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KEY="id", KIND="lookup")`)
+	p.ExecStmt(`CREATE TABLE tt3 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KEY="id", KIND="lookup")`)
 	p.ExecStmt(`CREATE TABLE tt4 () WITH (DATASOURCE="users", FORMAT="JSON")`)
 	defer func() {
 		p.ExecStmt(`DROP TABLE tt1`)
diff --git a/internal/topo/lookup/table_test.go b/internal/topo/lookup/table_test.go
index 7422ef15..99af264f 100644
--- a/internal/topo/lookup/table_test.go
+++ b/internal/topo/lookup/table_test.go
@@ -27,6 +27,7 @@ func TestTable(t *testing.T) {
 		DATASOURCE: "test",
 		TYPE:       "memory",
 		KIND:       "lookup",
+		KEY:        "id",
 	})
 	if err != nil {
 		t.Error(err)
@@ -36,6 +37,7 @@ func TestTable(t *testing.T) {
 		DATASOURCE: "test2",
 		TYPE:       "memory",
 		KIND:       "lookup",
+		KEY:        "id",
 	})
 	if err != nil {
 		t.Error(err)
diff --git a/internal/topo/memory/lookupsource_test.go b/internal/topo/memory/lookupsource_test.go
index 74a38b98..e28a9afd 100644
--- a/internal/topo/memory/lookupsource_test.go
+++ b/internal/topo/memory/lookupsource_test.go
@@ -25,11 +25,11 @@ import (
 	"time"
 )
 
-func TestNoIndexLookup(t *testing.T) {
+func TestUpdateLookup(t *testing.T) {
 	contextLogger := conf.Log.WithField("rule", "test")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
 	ls := GetLookupSource()
-	err := ls.Configure("test", map[string]interface{}{"option": "value"})
+	err := ls.Configure("test", map[string]interface{}{"key": "ff"})
 	if err != nil {
 		t.Error(err)
 		return
@@ -42,8 +42,12 @@ func TestNoIndexLookup(t *testing.T) {
 	// wait for the source to be ready
 	time.Sleep(100 * time.Millisecond)
 	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value1", "gg": "value2"})
-	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value2", "gg": "value2"})
+	pubsub.ProduceUpdatable(ctx, "test", map[string]interface{}{"ff": "value1", "gg": "value2"}, "delete", "value1")
+	pubsub.ProduceUpdatable(ctx, "test", map[string]interface{}{"ff": "value2", "gg": "value2"}, "insert", "value2")
 	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value1", "gg": "value4"})
+	pubsub.ProduceUpdatable(ctx, "test", map[string]interface{}{"ff": "value2", "gg": "value2"}, "delete", "value2")
+	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value1", "gg": "value2"})
+	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value2", "gg": "value2"})
 	// wait for table accumulation
 	time.Sleep(100 * time.Millisecond)
 	canctx, cancel := gocontext.WithCancel(gocontext.Background())
@@ -60,7 +64,6 @@ func TestNoIndexLookup(t *testing.T) {
 	}()
 	expected := []api.SourceTuple{
 		api.NewDefaultSourceTuple(map[string]interface{}{"ff": "value1", "gg": "value2"}, map[string]interface{}{"topic": "test"}),
-		api.NewDefaultSourceTuple(map[string]interface{}{"ff": "value1", "gg": "value4"}, map[string]interface{}{"topic": "test"}),
 	}
 	result, err := ls.Lookup(ctx, []string{}, []string{"ff"}, []interface{}{"value1"})
 	if !reflect.DeepEqual(result, expected) {
@@ -73,11 +76,11 @@ func TestNoIndexLookup(t *testing.T) {
 	}
 }
 
-func TestSingleIndexLookup(t *testing.T) {
+func TestLookup(t *testing.T) {
 	contextLogger := conf.Log.WithField("rule", "test2")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
 	ls := GetLookupSource()
-	err := ls.Configure("test2", map[string]interface{}{"index": []string{"ff"}})
+	err := ls.Configure("test2", map[string]interface{}{"key": "gg"})
 	if err != nil {
 		t.Error(err)
 		return
@@ -90,7 +93,7 @@ func TestSingleIndexLookup(t *testing.T) {
 	// wait for the source to be ready
 	time.Sleep(100 * time.Millisecond)
 	pubsub.Produce(ctx, "test2", map[string]interface{}{"ff": "value1", "gg": "value2"})
-	pubsub.Produce(ctx, "test2", map[string]interface{}{"ff": "value2", "gg": "value2"})
+	pubsub.Produce(ctx, "test2", map[string]interface{}{"ff": "value2", "gg": "value3"})
 	pubsub.Produce(ctx, "test2", map[string]interface{}{"ff": "value1", "gg": "value4"})
 	// wait for table accumulation
 	time.Sleep(100 * time.Millisecond)
@@ -102,7 +105,7 @@ func TestSingleIndexLookup(t *testing.T) {
 			case <-canctx.Done():
 				return
 			case <-time.After(10 * time.Millisecond):
-				pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value4", "gg": "value2"})
+				pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value4", "gg": "value5"})
 			}
 		}
 	}()
@@ -111,6 +114,13 @@ func TestSingleIndexLookup(t *testing.T) {
 		api.NewDefaultSourceTuple(map[string]interface{}{"ff": "value1", "gg": "value4"}, map[string]interface{}{"topic": "test2"}),
 	}
 	result, err := ls.Lookup(ctx, []string{}, []string{"ff"}, []interface{}{"value1"})
+	if len(result) != 2 {
+		t.Errorf("expect %v but got %v", expected, result)
+	} else {
+		if result[0].Message()["gg"] != "value2" {
+			result[0], result[1] = result[1], result[0]
+		}
+	}
 	if !reflect.DeepEqual(result, expected) {
 		t.Errorf("expect %v but got %v", expected, result)
 	}
diff --git a/internal/topo/memory/sink_test.go b/internal/topo/memory/sink_test.go
new file mode 100644
index 00000000..bbabef50
--- /dev/null
+++ b/internal/topo/memory/sink_test.go
@@ -0,0 +1,84 @@
+// Copyright 2022 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 memory
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"reflect"
+	"testing"
+)
+
+func TestUpdate(t *testing.T) {
+	contextLogger := conf.Log.WithField("rule", "test2")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	ms := GetSink()
+	err := ms.Configure(map[string]interface{}{"topic": "testupdate", "rowkindField": "verb", "keyField": "id"})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = ms.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var data = []map[string]interface{}{
+		{"id": "1", "verb": "insert", "name": "test1"},
+		{"id": "2", "verb": "insert", "name": "test2"},
+		{"id": "1", "verb": "update", "name": "test1"},
+		{"id": "2", "verb": "delete", "name": "test2"},
+	}
+	c := pubsub.CreateSub("testupdate", nil, "testSource", 100)
+	go func() {
+		for _, d := range data {
+			ms.Collect(ctx, d)
+		}
+	}()
+	var actual []api.SourceTuple
+	for i := 0; i < 4; i++ {
+		d := <-c
+		fmt.Println(d)
+		actual = append(actual, d)
+	}
+	expects := []api.SourceTuple{
+		&pubsub.UpdatableTuple{
+			DefaultSourceTuple: api.NewDefaultSourceTuple(map[string]interface{}{"id": "1", "verb": "insert", "name": "test1"}, map[string]interface{}{"topic": "testupdate"}),
+			Rowkind:            "insert",
+			Keyval:             "1",
+		},
+		&pubsub.UpdatableTuple{
+			DefaultSourceTuple: api.NewDefaultSourceTuple(map[string]interface{}{"id": "2", "verb": "insert", "name": "test2"}, map[string]interface{}{"topic": "testupdate"}),
+			Rowkind:            "insert",
+			Keyval:             "2",
+		},
+		&pubsub.UpdatableTuple{
+			DefaultSourceTuple: api.NewDefaultSourceTuple(map[string]interface{}{"id": "1", "verb": "update", "name": "test1"}, map[string]interface{}{"topic": "testupdate"}),
+			Rowkind:            "update",
+			Keyval:             "1",
+		},
+		&pubsub.UpdatableTuple{
+			DefaultSourceTuple: api.NewDefaultSourceTuple(map[string]interface{}{"id": "2", "verb": "delete", "name": "test2"}, map[string]interface{}{"topic": "testupdate"}),
+			Rowkind:            "delete",
+			Keyval:             "2",
+		},
+	}
+	if !reflect.DeepEqual(actual, expects) {
+		t.Errorf("expect %v but got %v", expects, actual)
+	}
+}
diff --git a/internal/topo/memory/store/db_test.go b/internal/topo/memory/store/db_test.go
index bd00e4c3..f5a4d4cb 100644
--- a/internal/topo/memory/store/db_test.go
+++ b/internal/topo/memory/store/db_test.go
@@ -21,35 +21,34 @@ import (
 )
 
 func TestTable(t *testing.T) {
-	tb := createTable([]string{"a"})
+	tb := createTable("topicT", "a")
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "0"}, nil))
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 2, "b": "0"}, nil))
-	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 3, "b": "0"}, nil))
+	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 3, "b": "4"}, nil))
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil))
 	v, _ := tb.Read([]string{"a"}, []interface{}{1})
 	exp := []api.SourceTuple{
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "0"}, nil),
 		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
 	}
 	if !reflect.DeepEqual(v, exp) {
-		t.Errorf("read 1 expect %v, but got %v", exp, v)
+		t.Errorf("read a 1 expect %v, but got %v", exp, v)
 		return
 	}
-	v, _ = tb.Read([]string{"a"}, []interface{}{3})
+	v, _ = tb.Read([]string{"b"}, []interface{}{"0"})
 	exp = []api.SourceTuple{
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 3, "b": "0"}, nil),
+		api.NewDefaultSourceTuple(map[string]interface{}{"a": 2, "b": "0"}, nil),
 	}
 	if !reflect.DeepEqual(v, exp) {
-		t.Errorf("read 3 expect %v, but got %v", exp, v)
+		t.Errorf("read b 0 expect %v, but got %v", exp, v)
 		return
 	}
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 5, "b": "0"}, nil))
+	tb.delete(3)
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil))
-	v, _ = tb.Read([]string{"a"}, []interface{}{1})
+	v, _ = tb.Read([]string{"b"}, []interface{}{"0"})
 	exp = []api.SourceTuple{
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "0"}, nil),
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
+		api.NewDefaultSourceTuple(map[string]interface{}{"a": 2, "b": "0"}, nil),
+		api.NewDefaultSourceTuple(map[string]interface{}{"a": 5, "b": "0"}, nil),
 	}
 	if !reflect.DeepEqual(v, exp) {
 		t.Errorf("read 1 again expect %v, but got %v", exp, v)
@@ -58,57 +57,66 @@ func TestTable(t *testing.T) {
 	v, _ = tb.Read([]string{"a", "b"}, []interface{}{1, "1"})
 	exp = []api.SourceTuple{
 		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
 	}
 	if !reflect.DeepEqual(v, exp) {
 		t.Errorf("read a,b expect %v, but got %v", exp, v)
 		return
 	}
+	v, _ = tb.Read([]string{"a"}, []interface{}{3})
+	if v != nil {
+		t.Errorf("read a 3 expect nil, but got %v", v)
+		return
+	}
+	tb.delete(1)
+	v, _ = tb.Read([]string{"a"}, []interface{}{1})
+	if v != nil {
+		t.Errorf("read a 1 expect nil, but got %v", v)
+	}
 }
 
 func TestDb(t *testing.T) {
 	db = &database{
 		tables: make(map[string]*tableCount),
 	}
-	db.addTable("t1", []string{"a"})
-	db.addTable("t1", []string{"a", "b"})
-	db.addTable("t2", []string{"a"})
-	db.addTable("t1", []string{"a"})
-	_, ok := db.getTable("t1")
+	db.addTable("t1", "a")
+	db.addTable("t1", "b")
+	db.addTable("t2", "a")
+	db.addTable("t1", "a")
+	_, ok := db.getTable("t1", "a")
 	if !ok {
 		t.Errorf("table t1 a should exist")
 		return
 	}
-	_, ok = db.getTable("t1")
+	_, ok = db.getTable("t1", "b")
 	if !ok {
 		t.Errorf("table t1 b should exist")
 		return
 	}
-	_, ok = db.getTable("t3")
+	_, ok = db.getTable("t3", "a")
 	if ok {
 		t.Errorf("table t1 c should not exist")
 		return
 	}
-	tc := db.tables["t1"]
-	if tc.count != 3 {
+	tc := db.tables["t1_a"]
+	if tc.count != 2 {
 		t.Errorf("table t1 a should have 2 instances but got %d", tc.count)
 		return
 	}
-	tc = db.tables["t2"]
+	tc = db.tables["t2_a"]
 	if tc.count != 1 {
-		t.Errorf("table t1 a should have 1 instances")
+		t.Errorf("table t2 a should have 1 instances")
 		return
 	}
-	db.dropTable("t1")
-	db.dropTable("t2")
-	_, ok = db.getTable("t2")
+	db.dropTable("t1", "a")
+	db.dropTable("t2", "a")
+	_, ok = db.getTable("t2", "a")
 	if ok {
-		t.Errorf("table ta a should not exist")
+		t.Errorf("table t2 a should not exist")
 		return
 	}
-	tc = db.tables["t1"]
-	if tc.count != 2 {
-		t.Errorf("table t1 a should have 2 instances but got %d", tc.count)
+	tc = db.tables["t1_a"]
+	if tc.count != 1 {
+		t.Errorf("table t1 a should have 1 instances but got %d", tc.count)
 		return
 	}
 }
diff --git a/internal/topo/memory/store/store_test.go b/internal/topo/memory/store/store_test.go
index a63a6ec9..647966c3 100644
--- a/internal/topo/memory/store/store_test.go
+++ b/internal/topo/memory/store/store_test.go
@@ -23,18 +23,18 @@ func TestReg(t *testing.T) {
 	db = &database{
 		tables: make(map[string]*tableCount),
 	}
-	reg1, err := Reg("test", nil, []string{"a"})
+	reg1, err := Reg("test", nil, "a")
 	if err != nil {
 		t.Errorf("register test error: %v", err)
 		return
 	}
-	_, err2 := Reg("test", nil, []string{"a", "b"})
+	_, err2 := Reg("test", nil, "a")
 	if err2 != nil {
 		t.Errorf("register test error: %v", err2)
 		return
 	}
 	exp := map[string]*tableCount{
-		"test": {
+		"test_a": {
 			count: 2,
 			t:     reg1,
 		},
@@ -43,9 +43,9 @@ func TestReg(t *testing.T) {
 		t.Errorf("register expect %v, but got %v", exp, db.tables)
 		return
 	}
-	Unreg("test")
+	Unreg("test", "a")
 	exp = map[string]*tableCount{
-		"test": {
+		"test_a": {
 			count: 1,
 			t:     reg1,
 		},
diff --git a/internal/topo/node/source_node_test.go b/internal/topo/node/source_node_test.go
index f50afabf..7b5f813f 100644
--- a/internal/topo/node/source_node_test.go
+++ b/internal/topo/node/source_node_test.go
@@ -31,6 +31,7 @@ func TestGetConf_Apply(t *testing.T) {
 		"incremental":        false,
 		"body":               "{}",
 		"bodyType":           "json",
+		"key":                "",
 		"format":             "json",
 		"insecureSkipVerify": true,
 		"headers": map[string]interface{}{
@@ -56,6 +57,7 @@ func TestGetConfAndConvert_Apply(t *testing.T) {
 		"incremental":        true,
 		"body":               "{}",
 		"bodyType":           "json",
+		"key":                "",
 		"format":             "json",
 		"insecureSkipVerify": true,
 		"headers": map[string]interface{}{
diff --git a/internal/xsql/parser_tree_test.go b/internal/xsql/parser_tree_test.go
index 937f2e28..7567974e 100644
--- a/internal/xsql/parser_tree_test.go
+++ b/internal/xsql/parser_tree_test.go
@@ -112,7 +112,7 @@ func TestParser_ParseTree(t *testing.T) {
 					name STRING,
 					size BIGINT,
 					id BIGINT
-				) WITH (DATASOURCE="devices", KIND="LOOKUP", TYPE="sql");`,
+				) WITH (DATASOURCE="devices", KIND="LOOKUP", TYPE="sql", KEY="id");`,
 			stmt: &ast.StreamStmt{
 				Name: ast.StreamName("table1"),
 				StreamFields: []ast.StreamField{
@@ -125,10 +125,19 @@ func TestParser_ParseTree(t *testing.T) {
 					STRICT_VALIDATION: true,
 					KIND:              ast.StreamKindLookup,
 					TYPE:              "sql",
+					KEY:               "id",
 				},
 				StreamType: ast.TypeTable,
 			},
 		},
+		{
+			s: `CREATE TABLE table1 (
+					name STRING,
+					size BIGINT,
+					id BIGINT
+				) WITH (DATASOURCE="devices", KIND="LOOKUP", TYPE="memory");`,
+			err: `Option "key" is required for memory lookup table.`,
+		},
 		{
 			s:    `SHOW STREAMS`,
 			stmt: &ast.ShowStreamsStatement{},