Repository: lf-edge/ekuiper
Pull request #1405 (2022-09-20 01:31:50)

feat(lookup): lookup table runtime
Signed-off-by: Jiyong Huang <huangjy@emqx.io>


Patch:
diff --git a/etc/sources/memory.yaml b/etc/sources/memory.yaml
index e69de29b..3a248136 100644
--- a/etc/sources/memory.yaml
+++ b/etc/sources/memory.yaml
@@ -0,0 +1,2 @@
+default:
+  index: ["id"]
\ No newline at end of file
diff --git a/extensions/sources/sql/convert.go b/extensions/sources/sql/convert.go
new file mode 100644
index 00000000..bf15cc4b
--- /dev/null
+++ b/extensions/sources/sql/convert.go
@@ -0,0 +1,52 @@
+// 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 main
+
+import (
+	"database/sql"
+	"database/sql/driver"
+	"reflect"
+)
+
+func scanIntoMap(mapValue map[string]interface{}, values []interface{}, columns []string) {
+	for idx, column := range columns {
+		if reflectValue := reflect.Indirect(reflect.Indirect(reflect.ValueOf(values[idx]))); reflectValue.IsValid() {
+			mapValue[column] = reflectValue.Interface()
+			if valuer, ok := mapValue[column].(driver.Valuer); ok {
+				mapValue[column], _ = valuer.Value()
+			} else if b, ok := mapValue[column].(sql.RawBytes); ok {
+				mapValue[column] = string(b)
+			}
+		} else {
+			mapValue[column] = nil
+		}
+	}
+}
+
+func prepareValues(values []interface{}, columnTypes []*sql.ColumnType, columns []string) {
+	if len(columnTypes) > 0 {
+		for idx, columnType := range columnTypes {
+			if columnType.ScanType() != nil {
+				values[idx] = reflect.New(reflect.PtrTo(columnType.ScanType())).Interface()
+			} else {
+				values[idx] = new(interface{})
+			}
+		}
+	} else {
+		for idx := range columns {
+			values[idx] = new(interface{})
+		}
+	}
+}
diff --git a/extensions/sources/sql/sql.go b/extensions/sources/sql/sql.go
index 1c4b8364..55cdf843 100644
--- a/extensions/sources/sql/sql.go
+++ b/extensions/sources/sql/sql.go
@@ -16,14 +16,12 @@ package main
 
 import (
 	"database/sql"
-	"database/sql/driver"
 	"fmt"
 	driver2 "github.com/lf-edge/ekuiper/extensions/sqldatabase/driver"
 	"github.com/lf-edge/ekuiper/extensions/sqldatabase/sqlgen"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/xo/dburl"
-	"reflect"
 	"time"
 )
 
@@ -113,7 +111,7 @@ func (m *sqlsource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple,
 			for rows.Next() {
 				data := make(map[string]interface{})
 				columns := make([]interface{}, len(cols))
-				m.prepareValues(columns, types, cols)
+				prepareValues(columns, types, cols)
 
 				err := rows.Scan(columns...)
 				if err != nil {
@@ -122,7 +120,7 @@ func (m *sqlsource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple,
 					return
 				}
 
-				m.scanIntoMap(data, columns, cols)
+				scanIntoMap(data, columns, cols)
 				m.Query.UpdateMaxIndexValue(data)
 				consumer <- api.NewDefaultSourceTuple(data, nil)
 			}
@@ -132,37 +130,6 @@ func (m *sqlsource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple,
 	}
 }
 
-func (m *sqlsource) prepareValues(values []interface{}, columnTypes []*sql.ColumnType, columns []string) {
-	if len(columnTypes) > 0 {
-		for idx, columnType := range columnTypes {
-			if columnType.ScanType() != nil {
-				values[idx] = reflect.New(reflect.PtrTo(columnType.ScanType())).Interface()
-			} else {
-				values[idx] = new(interface{})
-			}
-		}
-	} else {
-		for idx := range columns {
-			values[idx] = new(interface{})
-		}
-	}
-}
-
-func (m *sqlsource) scanIntoMap(mapValue map[string]interface{}, values []interface{}, columns []string) {
-	for idx, column := range columns {
-		if reflectValue := reflect.Indirect(reflect.Indirect(reflect.ValueOf(values[idx]))); reflectValue.IsValid() {
-			mapValue[column] = reflectValue.Interface()
-			if valuer, ok := mapValue[column].(driver.Valuer); ok {
-				mapValue[column], _ = valuer.Value()
-			} else if b, ok := mapValue[column].(sql.RawBytes); ok {
-				mapValue[column] = string(b)
-			}
-		} else {
-			mapValue[column] = nil
-		}
-	}
-}
-
 func (m *sqlsource) GetOffset() (interface{}, error) {
 	return m.Query.GetIndexValue(), nil
 }
diff --git a/extensions/sources/sql/sqlLookup.go b/extensions/sources/sql/sqlLookup.go
new file mode 100644
index 00000000..211816d4
--- /dev/null
+++ b/extensions/sources/sql/sqlLookup.go
@@ -0,0 +1,116 @@
+// 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 main
+
+import (
+	"database/sql"
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/xo/dburl"
+)
+
+type sqlLookupConfig struct {
+	Url string `json:"url"`
+}
+
+type sqlLookupSource struct {
+	url      string
+	template string
+	db       *sql.DB
+}
+
+// Open establish a connection to the database
+func (s *sqlLookupSource) Open(ctx api.StreamContext) error {
+	ctx.GetLogger().Debugf("Opening sql lookup source")
+	db, err := dburl.Open(s.url)
+	if err != nil {
+		return fmt.Errorf("connection to %s Open with error %v", s.url, err)
+	}
+	s.db = db
+	return nil
+}
+
+func (s *sqlLookupSource) Configure(datasource string, props map[string]interface{}) error {
+	cfg := &sqlLookupConfig{}
+	err := cast.MapToStruct(props, cfg)
+	if err != nil {
+		return fmt.Errorf("read properties %v fail with error: %v", props, err)
+	}
+	if cfg.Url == "" {
+		return fmt.Errorf("property Url is required")
+	}
+	_, err = dburl.Parse(cfg.Url)
+	if err != nil {
+		return fmt.Errorf("dburl.Parse %s fail with error: %v", cfg.Url, err)
+	}
+	s.url = cfg.Url
+	s.template = fmt.Sprintf("SELECT * FROM `%s` WHERE ", datasource)
+	return nil
+}
+
+func (s *sqlLookupSource) Lookup(ctx api.StreamContext, keys []string, values []interface{}) ([]api.SourceTuple, error) {
+	ctx.GetLogger().Debug("Start to lookup tuple")
+	query := s.template
+	for i, k := range keys {
+		if i > 0 {
+			query += " AND "
+		}
+		switch v := values[i].(type) {
+		case string:
+			query += fmt.Sprintf("`%s` = \"%s\"", k, v)
+		default:
+			query += fmt.Sprintf("`%s` = %v", k, v)
+		}
+	}
+	ctx.GetLogger().Debugf("Query is %s", query)
+	// TODO extract common functions
+	rows, err := s.db.Query(query)
+	if err != nil {
+		return nil, err
+	}
+	cols, _ := rows.Columns()
+
+	types, err := rows.ColumnTypes()
+	if err != nil {
+		return nil, err
+	}
+	var result []api.SourceTuple
+	for rows.Next() {
+		data := make(map[string]interface{})
+		columns := make([]interface{}, len(cols))
+		prepareValues(columns, types, cols)
+
+		err := rows.Scan(columns...)
+		if err != nil {
+			return nil, err
+		}
+		scanIntoMap(data, columns, cols)
+		result = append(result, api.NewDefaultSourceTuple(data, nil))
+	}
+	return result, nil
+}
+
+func (s *sqlLookupSource) Close(ctx api.StreamContext) error {
+	defer func() { s.db = nil }()
+	if s.db != nil {
+		return s.db.Close()
+	}
+	return nil
+}
+
+func SqlLookup() api.LookupSource {
+	return &sqlLookupSource{}
+}
diff --git a/go.sum b/go.sum
index 86389084..047ae574 100644
--- a/go.sum
+++ b/go.sum
@@ -52,8 +52,6 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/eclipse/paho.mqtt.golang v1.3.5 h1:sWtmgNxYM9P2sP+xEItMozsR3w0cqZFlqnNN1bdl41Y=
-github.com/eclipse/paho.mqtt.golang v1.3.5/go.mod h1:eTzb4gxwwyWpqBUHGQZ4ABAV7+Jgm1PklsYT/eo8Hcc=
 github.com/eclipse/paho.mqtt.golang v1.4.2-0.20220810043731-079a117b4614 h1:36h5MuaQ0gVNV1clhTE4lF3Vram4HNGX3JcDlD3u6UA=
 github.com/eclipse/paho.mqtt.golang v1.4.2-0.20220810043731-079a117b4614/go.mod h1:JGt0RsEwEX+Xa/agj90YJ9d9DH2b7upDZMK9HRbFvCA=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.2.0 h1:Sfi9jAIgRXZaJQw8Ji6+8//47D+iOyGiXQSNZXhy3HE=
diff --git a/internal/binder/factory.go b/internal/binder/factory.go
index 6b3168e0..a79eed7b 100644
--- a/internal/binder/factory.go
+++ b/internal/binder/factory.go
@@ -18,6 +18,7 @@ import "github.com/lf-edge/ekuiper/pkg/api"
 
 type SourceFactory interface {
 	Source(name string) (api.Source, error)
+	LookupSource(name string) (api.LookupSource, error)
 }
 
 type SinkFactory interface {
diff --git a/internal/binder/io/binder.go b/internal/binder/io/binder.go
index ae768e8b..7495a861 100644
--- a/internal/binder/io/binder.go
+++ b/internal/binder/io/binder.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.
@@ -80,3 +80,17 @@ func Sink(name string) (api.Sink, error) {
 	}
 	return nil, e.GetError()
 }
+
+func LookupSource(name string) (api.LookupSource, error) {
+	e := make(errorx.MultiError)
+	for i, sf := range sourceFactories {
+		r, err := sf.LookupSource(name)
+		if err != nil {
+			e[sourceFactoriesNames[i]] = err
+		}
+		if r != nil {
+			return r, e.GetError()
+		}
+	}
+	return nil, e.GetError()
+}
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index ce3fee27..58114c36 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -23,6 +23,7 @@ import (
 )
 
 type NewSourceFunc func() api.Source
+type NewLookupSourceFunc func() api.LookupSource
 type NewSinkFunc func() api.Sink
 
 var (
@@ -43,6 +44,9 @@ var (
 		"memory":      func() api.Sink { return memory.GetSink() },
 		"neuron":      func() api.Sink { return neuron.GetSink() },
 	}
+	lookupSources = map[string]NewLookupSourceFunc{
+		"memory": func() api.LookupSource { return memory.GetLookupSource() },
+	}
 )
 
 type Manager struct{}
@@ -54,6 +58,13 @@ func (m *Manager) Source(name string) (api.Source, error) {
 	return nil, nil
 }
 
+func (m *Manager) LookupSource(name string) (api.LookupSource, error) {
+	if s, ok := lookupSources[name]; ok {
+		return s(), nil
+	}
+	return nil, nil
+}
+
 func (m *Manager) Sink(name string) (api.Sink, error) {
 	if s, ok := sinks[name]; ok {
 		return s(), nil
diff --git a/internal/binder/mock/mock_factory.go b/internal/binder/mock/mock_factory.go
index c6918c03..e08fa033 100644
--- a/internal/binder/mock/mock_factory.go
+++ b/internal/binder/mock/mock_factory.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.
@@ -35,6 +35,10 @@ func (f *MockFactory) Source(name string) (api.Source, error) {
 	}
 }
 
+func (f *MockFactory) LookupSource(name string) (api.LookupSource, error) {
+	return nil, errorx.NotFoundErr
+}
+
 func (f *MockFactory) Sink(name string) (api.Sink, error) {
 	if strings.HasPrefix(name, "mock") {
 		return &mockSink{}, nil
diff --git a/internal/plugin/native/manager.go b/internal/plugin/native/manager.go
index f8969ce5..b4f7c0b7 100644
--- a/internal/plugin/native/manager.go
+++ b/internal/plugin/native/manager.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.
@@ -55,7 +55,7 @@ type Manager struct {
 	// A map from function name to its plugin file name. It is constructed during initialization by reading kv info. All functions must have at least an entry, even the function resizes in a one function plugin.
 	symbols map[string]string
 	// loaded symbols in current runtime
-	runtime map[string]plugin.Symbol
+	runtime map[string]*plugin.Plugin
 	// dirs
 	pluginDir string
 	etcDir    string
@@ -77,7 +77,7 @@ func InitManager() (*Manager, error) {
 	if err != nil {
 		return nil, fmt.Errorf("error when opening db: %v", err)
 	}
-	registry := &Manager{symbols: make(map[string]string), db: db, pluginDir: pluginDir, etcDir: etcDir, runtime: make(map[string]plugin.Symbol)}
+	registry := &Manager{symbols: make(map[string]string), db: db, pluginDir: pluginDir, etcDir: etcDir, runtime: make(map[string]*plugin.Plugin)}
 	manager = registry
 	plugins := make([]map[string]string, 3)
 	for i := range plugin2.PluginTypes {
@@ -116,7 +116,7 @@ func findAll(t plugin2.PluginType, pluginDir string) (result map[string]string,
 			n, v := parseName(baseName)
 			//load the plugins when ekuiper set up
 			if !conf.IsTesting {
-				if _, err := manager.loadRuntime(t, n, path.Join(dir, baseName)); err != nil {
+				if _, err := manager.loadRuntime(t, n, path.Join(dir, baseName), ""); err != nil {
 					continue
 				}
 			}
@@ -513,7 +513,7 @@ func (rr *Manager) install(t plugin2.PluginType, name, src string, shellParas []
 
 	if !conf.IsTesting {
 		// load the runtime first
-		_, err = manager.loadRuntime(t, soName, soPath)
+		_, err = manager.loadRuntime(t, soName, soPath, "")
 		if err != nil {
 			return version, err
 		}
@@ -526,7 +526,7 @@ func (rr *Manager) install(t plugin2.PluginType, name, src string, shellParas []
 // binder factory implementations
 
 func (rr *Manager) Source(name string) (api.Source, error) {
-	nf, err := rr.loadRuntime(plugin2.SOURCE, name, "")
+	nf, err := rr.loadRuntime(plugin2.SOURCE, name, "", "")
 	if err != nil {
 		return nil, err
 	}
@@ -543,8 +543,26 @@ func (rr *Manager) Source(name string) (api.Source, error) {
 	}
 }
 
+func (rr *Manager) LookupSource(name string) (api.LookupSource, error) {
+	nf, err := rr.loadRuntime(plugin2.SOURCE, name, "", ucFirst(name)+"Lookup")
+	if err != nil {
+		return nil, err
+	}
+	if nf == nil {
+		return nil, nil
+	}
+	switch t := nf.(type) {
+	case api.LookupSource:
+		return t, nil
+	case func() api.LookupSource:
+		return t(), nil
+	default:
+		return nil, fmt.Errorf("exported symbol %s is not type of api.LookupSource or function that return api.LookupSource", t)
+	}
+}
+
 func (rr *Manager) Sink(name string) (api.Sink, error) {
-	nf, err := rr.loadRuntime(plugin2.SINK, name, "")
+	nf, err := rr.loadRuntime(plugin2.SINK, name, "", "")
 	if err != nil {
 		return nil, err
 	}
@@ -564,7 +582,7 @@ func (rr *Manager) Sink(name string) (api.Sink, error) {
 }
 
 func (rr *Manager) Function(name string) (api.Function, error) {
-	nf, err := rr.loadRuntime(plugin2.FUNCTION, name, "")
+	nf, err := rr.loadRuntime(plugin2.FUNCTION, name, "", "")
 	if err != nil {
 		return nil, err
 	}
@@ -597,43 +615,50 @@ func (rr *Manager) ConvName(name string) (string, bool) {
 }
 
 // If not found, return nil,nil; Other errors return nil, err
-func (rr *Manager) loadRuntime(t plugin2.PluginType, name, soFilepath string) (plugin.Symbol, error) {
-	ut := ucFirst(name)
+func (rr *Manager) loadRuntime(t plugin2.PluginType, soName, soFilepath, symbolName string) (plugin.Symbol, error) {
 	ptype := plugin2.PluginTypes[t]
-	key := ptype + "/" + name
-	var nf plugin.Symbol
+	key := ptype + "/" + soName
+	var (
+		plug *plugin.Plugin
+		ok   bool
+		err  error
+	)
 	rr.RLock()
-	nf, ok := rr.runtime[key]
+	plug, ok = rr.runtime[key]
 	rr.RUnlock()
 	if !ok {
 		var soPath string
 		if soFilepath != "" {
 			soPath = soFilepath
 		} else {
-			mod, err := rr.getSoFilePath(t, name, false)
+			mod, err := rr.getSoFilePath(t, soName, false)
 			if err != nil {
-				conf.Log.Warnf(fmt.Sprintf("cannot find the native plugin %s in path: %v", name, err))
+				conf.Log.Warnf(fmt.Sprintf("cannot find the native plugin %s in path: %v", soName, err))
 				return nil, nil
 			}
 			soPath = mod
 		}
 		conf.Log.Debugf("Opening plugin %s", soPath)
-		plug, err := plugin.Open(soPath)
+		plug, err = plugin.Open(soPath)
 		if err != nil {
-			conf.Log.Errorf(fmt.Sprintf("plugin %s open error: %v", name, err))
+			conf.Log.Errorf(fmt.Sprintf("plugin %s open error: %v", soName, err))
 			return nil, fmt.Errorf("cannot open %s: %v", soPath, err)
 		}
-		conf.Log.Debugf("Successfully open plugin %s", soPath)
-		nf, err = plug.Lookup(ut)
-		if err != nil {
-			conf.Log.Warnf(fmt.Sprintf("cannot find symbol %s, please check if it is exported: %v", ut, err))
-			return nil, nil
-		}
-		conf.Log.Debugf("Successfully look-up plugin %s", soPath)
 		rr.Lock()
-		rr.runtime[key] = nf
+		rr.runtime[key] = plug
 		rr.Unlock()
+		conf.Log.Debugf("Successfully open plugin %s", soPath)
+	}
+	if symbolName == "" {
+		symbolName = ucFirst(soName)
+	}
+	conf.Log.Debugf("Loading symbol %s", symbolName)
+	nf, err := plug.Lookup(symbolName)
+	if err != nil {
+		conf.Log.Warnf(fmt.Sprintf("cannot find symbol %s, please check if it is exported: %v", symbolName, err))
+		return nil, nil
 	}
+	conf.Log.Debugf("Successfully look-up plugin %s", symbolName)
 	return nf, nil
 }
 
diff --git a/internal/plugin/portable/factory.go b/internal/plugin/portable/factory.go
index 0255ba26..16116cd1 100644
--- a/internal/plugin/portable/factory.go
+++ b/internal/plugin/portable/factory.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.
@@ -30,6 +30,11 @@ func (m *Manager) Source(name string) (api.Source, error) {
 	return runtime.NewPortableSource(name, meta), nil
 }
 
+func (m *Manager) LookupSource(_ string) (api.LookupSource, error) {
+	// TODO add support
+	return nil, nil
+}
+
 func (m *Manager) Sink(name string) (api.Sink, error) {
 	meta, ok := m.GetPluginMeta(plugin.SINK, name)
 	if !ok {
diff --git a/internal/processor/stream.go b/internal/processor/stream.go
index bcbdf742..5bfa1e69 100644
--- a/internal/processor/stream.go
+++ b/internal/processor/stream.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/internal/topo/lookup"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
@@ -99,7 +100,49 @@ func (p *StreamProcessor) ExecStmt(statement string) (result []string, err error
 	return
 }
 
+func (p *StreamProcessor) RecoverLookupTable() error {
+	keys, err := p.db.Keys()
+	if err != nil {
+		return fmt.Errorf("error loading data from db: %v.", err)
+	}
+	var (
+		v  string
+		vs = &xsql.StreamInfo{}
+	)
+	for _, k := range keys {
+		if ok, _ := p.db.Get(k, &v); ok {
+			if err := json.Unmarshal([]byte(v), vs); err == nil && vs.StreamType == ast.TypeTable {
+				parser := xsql.NewParser(strings.NewReader(vs.Statement))
+				stmt, e := xsql.Language.Parse(parser)
+				if e != nil {
+					log.Error(err)
+				}
+				switch s := stmt.(type) {
+				case *ast.StreamStmt:
+					log.Infof("Starting lookup table %s", s.Name)
+					e = lookup.CreateInstance(string(s.Name), s.Options.TYPE, s.Options)
+					if err != nil {
+						log.Errorf("%s", err.Error())
+						return err
+					}
+				default:
+					log.Errorf("Invalid lookup table statement: %s", vs.Statement)
+				}
+
+			}
+		}
+	}
+	return nil
+}
+
 func (p *StreamProcessor) execSave(stmt *ast.StreamStmt, statement string, replace bool) error {
+	if stmt.StreamType == ast.TypeTable && stmt.Options.KIND == ast.StreamKindLookup {
+		log.Infof("Creating lookup table %s", stmt.Name)
+		err := lookup.CreateInstance(string(stmt.Name), stmt.Options.TYPE, stmt.Options)
+		if err != nil {
+			return err
+		}
+	}
 	s, err := json.Marshal(xsql.StreamInfo{
 		StreamType: stmt.StreamType,
 		Statement:  statement,
@@ -281,6 +324,12 @@ func (p *StreamProcessor) execDrop(stmt ast.NameNode, st ast.StreamType) (string
 }
 
 func (p *StreamProcessor) DropStream(name string, st ast.StreamType) (string, error) {
+	if st == ast.TypeTable {
+		err := lookup.DropInstance(name)
+		if err != nil {
+			return "", err
+		}
+	}
 	_, err := p.getStream(name, st)
 	if err != nil {
 		return "", err
diff --git a/internal/server/server.go b/internal/server/server.go
index 58870ca3..78f42263 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -74,6 +74,8 @@ func StartUp(Version, LoadFileType string) {
 	meta.Bind()
 
 	registry = &RuleRegistry{internal: make(map[string]*RuleState)}
+	//Start lookup tables
+	streamProcessor.RecoverLookupTable()
 	//Start rules
 	if rules, err := ruleProcessor.GetAllRules(); err != nil {
 		logger.Infof("Start rules error: %s", err)
diff --git a/internal/topo/lookup/table.go b/internal/topo/lookup/table.go
new file mode 100644
index 00000000..55fdc924
--- /dev/null
+++ b/internal/topo/lookup/table.go
@@ -0,0 +1,106 @@
+// 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 lookup
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/binder/io"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	kctx "github.com/lf-edge/ekuiper/internal/topo/context"
+	nodeConf "github.com/lf-edge/ekuiper/internal/topo/node/conf"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"sync"
+	"sync/atomic"
+)
+
+// Table is a lookup table runtime instance. It will run once the table is created.
+// It will only stop once the table is dropped.
+
+type info struct {
+	ls    api.LookupSource
+	count int32
+}
+
+var (
+	instances = make(map[string]*info)
+	lock      = &sync.Mutex{}
+)
+
+// Attach called by lookup nodes. Add a count to the info
+func Attach(name string) (api.LookupSource, error) {
+	lock.Lock()
+	defer lock.Unlock()
+	if i, ok := instances[name]; ok {
+		atomic.AddInt32(&i.count, 1)
+		return i.ls, nil
+	}
+	return nil, fmt.Errorf("lookup table %s is not found", name)
+}
+
+// Detach called by lookup nodes when it is closed
+func Detach(name string) error {
+	lock.Lock()
+	defer lock.Unlock()
+	if i, ok := instances[name]; ok {
+		atomic.AddInt32(&i.count, -1)
+		return nil
+	}
+	return fmt.Errorf("lookup table %s is not found", name)
+}
+
+// CreateInstance called when create a lookup table
+func CreateInstance(name string, sourceType string, options *ast.Options) error {
+	lock.Lock()
+	defer lock.Unlock()
+	contextLogger := conf.Log.WithField("table", name)
+	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+	props := nodeConf.GetSourceConf(ctx, sourceType, options)
+	ctx.GetLogger().Infof("open lookup table with props %v", conf.Printable(props))
+	// Create the lookup source according to the source options
+	ns, err := io.LookupSource(sourceType)
+	if err != nil {
+		ctx.GetLogger().Error(err)
+		return err
+	}
+	ctx.GetLogger().Debugf("lookup source %s is created", sourceType)
+	err = ns.Configure(options.DATASOURCE, props)
+	if err != nil {
+		return err
+	}
+	ctx.GetLogger().Debugf("lookup source %s is configured", sourceType)
+	err = ns.Open(ctx)
+	if err != nil {
+		return err
+	}
+	ctx.GetLogger().Debugf("lookup source %s is opened", sourceType)
+	instances[name] = &info{ls: ns, count: 0}
+	return nil
+}
+
+// DropInstance called when drop a lookup table
+func DropInstance(name string) error {
+	lock.Lock()
+	defer lock.Unlock()
+	if i, ok := instances[name]; ok {
+		if atomic.LoadInt32(&i.count) > 0 {
+			return fmt.Errorf("lookup table %s is still in use, stop all using rules before dropping it", name)
+		}
+		delete(instances, name)
+		return nil
+	} else {
+		return nil
+	}
+}
diff --git a/internal/topo/memory/export.go b/internal/topo/memory/export.go
new file mode 100644
index 00000000..d8a3cf28
--- /dev/null
+++ b/internal/topo/memory/export.go
@@ -0,0 +1,27 @@
+// 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
+
+func GetSink() *sink {
+	return &sink{}
+}
+
+func GetSource() *source {
+	return &source{}
+}
+
+func GetLookupSource() *lookupsource {
+	return &lookupsource{}
+}
diff --git a/internal/topo/memory/lookupsource.go b/internal/topo/memory/lookupsource.go
new file mode 100644
index 00000000..b6773a19
--- /dev/null
+++ b/internal/topo/memory/lookupsource.go
@@ -0,0 +1,66 @@
+// 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 (
+	"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"
+)
+
+// lookupsource is a lookup source that reads data from memory
+// The memory lookup table reads a global memory store for data
+type lookupsource struct {
+	topic      string
+	topicRegex *regexp.Regexp
+	keys       []string
+	table      *store.Table
+}
+
+func (s *lookupsource) Open(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("lookup source %s is opened with keys %v", s.topic, s.keys)
+	var err error
+	s.table, err = store.Reg(s.topic, s.topicRegex, s.keys)
+	return err
+}
+
+func (s *lookupsource) Configure(datasource string, props map[string]interface{}) error {
+	s.topic = datasource
+	if strings.ContainsAny(datasource, "+#") {
+		r, err := getRegexp(datasource)
+		if err != nil {
+			return err
+		}
+		s.topicRegex = r
+	}
+	if c, ok := props["index"]; ok {
+		if bl, err := cast.ToStringSlice(c, cast.CONVERT_SAMEKIND); err != nil {
+			s.keys = bl
+		}
+	}
+	return nil
+}
+
+func (s *lookupsource) Lookup(ctx api.StreamContext, keys []string, values []interface{}) ([]api.SourceTuple, error) {
+	ctx.GetLogger().Debugf("lookup source %s is looking up keys %v with values %v", s.topic, keys, values)
+	return s.table.Read(keys, values)
+}
+
+func (s *lookupsource) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("lookup source %s is closing", s.topic)
+	return store.Unreg(s.topic)
+}
diff --git a/internal/topo/memory/manager.go b/internal/topo/memory/pubsub/manager.go
similarity index 97%
rename from internal/topo/memory/manager.go
rename to internal/topo/memory/pubsub/manager.go
index 28e4ac58..8bc2232b 100644
--- a/internal/topo/memory/manager.go
+++ b/internal/topo/memory/pubsub/manager.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package memory
+package pubsub
 
 import (
 	"github.com/lf-edge/ekuiper/internal/conf"
@@ -40,14 +40,6 @@ var (
 	mu        = sync.RWMutex{}
 )
 
-func GetSink() *sink {
-	return &sink{}
-}
-
-func GetSource() *source {
-	return &source{}
-}
-
 func CreatePub(topic string) {
 	mu.Lock()
 	defer mu.Unlock()
@@ -186,3 +178,9 @@ func removePubConsumer(topic string, sourceId string, c *pubConsumers) {
 		delete(pubTopics, topic)
 	}
 }
+
+// Reset For testing only
+func Reset() {
+	pubTopics = make(map[string]*pubConsumers)
+	subExps = make(map[string]*subChan)
+}
diff --git a/internal/topo/memory/sink.go b/internal/topo/memory/sink.go
index b3abaa65..da3356d6 100644
--- a/internal/topo/memory/sink.go
+++ b/internal/topo/memory/sink.go
@@ -17,6 +17,7 @@ package memory
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"strings"
 )
@@ -28,12 +29,12 @@ type sink struct {
 
 func (s *sink) Open(ctx api.StreamContext) error {
 	ctx.GetLogger().Debugf("Opening memory sink: %v", s.topic)
-	CreatePub(s.topic)
+	pubsub.CreatePub(s.topic)
 	return nil
 }
 
 func (s *sink) Configure(props map[string]interface{}) error {
-	if t, ok := props[IdProperty]; ok {
+	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)
@@ -71,10 +72,10 @@ func (s *sink) Collect(ctx api.StreamContext, data interface{}) error {
 	switch d := data.(type) {
 	case []map[string]interface{}:
 		for _, el := range d {
-			Produce(ctx, topic, el)
+			pubsub.Produce(ctx, topic, el)
 		}
 	case map[string]interface{}:
-		Produce(ctx, topic, d)
+		pubsub.Produce(ctx, topic, d)
 	default:
 		return fmt.Errorf("unrecognized format of %s", data)
 	}
@@ -83,6 +84,6 @@ func (s *sink) Collect(ctx api.StreamContext, data interface{}) error {
 
 func (s *sink) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Debugf("closing memory sink")
-	RemovePub(s.topic)
+	pubsub.RemovePub(s.topic)
 	return nil
 }
diff --git a/internal/topo/memory/source.go b/internal/topo/memory/source.go
index 234bc5e7..89ee462b 100644
--- a/internal/topo/memory/source.go
+++ b/internal/topo/memory/source.go
@@ -16,6 +16,7 @@ package memory
 
 import (
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"regexp"
@@ -29,7 +30,7 @@ type source struct {
 }
 
 func (s *source) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, _ chan<- error) {
-	ch := CreateSub(s.topic, s.topicRegex, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), s.bufferLength)
+	ch := pubsub.CreateSub(s.topic, s.topicRegex, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), s.bufferLength)
 	for {
 		select {
 		case v, opened := <-ch:
@@ -78,6 +79,6 @@ func getRegexp(topic string) (*regexp.Regexp, error) {
 
 func (s *source) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Debugf("closing memory source")
-	CloseSourceConsumerChannel(s.topic, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()))
+	pubsub.CloseSourceConsumerChannel(s.topic, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()))
 	return nil
 }
diff --git a/internal/topo/memory/store/db.go b/internal/topo/memory/store/db.go
new file mode 100644
index 00000000..821c8719
--- /dev/null
+++ b/internal/topo/memory/store/db.go
@@ -0,0 +1,177 @@
+// 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 store
+
+import (
+	"context"
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"sync"
+)
+
+type tableCount struct {
+	sync.RWMutex
+	count int
+	t     *Table
+}
+
+func (tc *tableCount) Increase() int {
+	tc.Lock()
+	defer tc.Unlock()
+	tc.count++
+	return tc.count
+}
+
+func (tc *tableCount) Decrease() int {
+	tc.Lock()
+	defer tc.Unlock()
+	tc.count--
+	if tc.count < 0 {
+		fmt.Errorf("Table count is less than 0: %d", tc.count)
+	}
+	return tc.count
+}
+
+type database struct {
+	sync.RWMutex
+	tables map[string]*tableCount // topic: table
+}
+
+// getTable return the table of the topic.
+func (db *database) getTable(topic string) (*Table, bool) {
+	db.RLock()
+	defer db.RUnlock()
+	tc, ok := db.tables[topic]
+	if ok {
+		return tc.t, true
+	} else {
+		return nil, false
+	}
+}
+
+// addTable add a table to the database
+// 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) {
+	db.Lock()
+	defer db.Unlock()
+	tc, ok := db.tables[topic]
+	if ok {
+		tc.Increase()
+	} else {
+		t := createTable(keys)
+		tc = &tableCount{
+			count: 1,
+			t:     t,
+		}
+		db.tables[topic] = tc
+	}
+	return tc.t, !ok
+}
+
+// dropTable drop the table of the topic/values
+// stops to accumulate job
+// deletes the cache data
+func (db *database) dropTable(topic string) error {
+	db.Lock()
+	defer db.Unlock()
+	if tc, ok := db.tables[topic]; ok {
+		if tc.Decrease() == 0 {
+			if tc.t != nil && tc.t.cancel != nil {
+				tc.t.cancel()
+			}
+			delete(db.tables, topic)
+		}
+		return nil
+	}
+	return fmt.Errorf("Table %s not found", topic)
+}
+
+// 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
+	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
+	}
+	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)
+		}
+	}
+}
+
+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
+		for i, k := range keys {
+			if d, ok := t.indexes[k]; ok {
+				data = d[values[i]]
+				excludeKey = i
+			}
+		}
+	}
+	var result []api.SourceTuple
+	for _, v := range data {
+		match := true
+		for i, k := range keys {
+			if i == excludeKey {
+				continue
+			}
+			if val, ok := v.Message()[k]; !ok || val != values[i] {
+				match = false
+				break
+			}
+		}
+		if match {
+			result = append(result, v)
+		}
+	}
+	return result, nil
+}
+
+var (
+	db = &database{
+		tables: make(map[string]*tableCount),
+	}
+)
diff --git a/internal/topo/memory/store/store.go b/internal/topo/memory/store/store.go
new file mode 100644
index 00000000..aee41187
--- /dev/null
+++ b/internal/topo/memory/store/store.go
@@ -0,0 +1,61 @@
+// 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 store
+
+import (
+	"context"
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
+	"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)
+	if isNew {
+		go runTable(topic, topicRegex, t)
+	}
+	return t, nil
+}
+
+// runTable should only run in a single instance.
+// This go routine is used to accumulate data in memory
+// If the go routine close, the go routine exits but the data will be kept until table dropped
+func runTable(topic string, topicRegex *regexp.Regexp, t *Table) {
+	conf.Log.Infof("runTable %s", topic)
+	ch := pubsub.CreateSub(topic, topicRegex, fmt.Sprintf("store_%s", topic), 1024)
+	ctx, cancel := context.WithCancel(context.Background())
+	t.cancel = cancel
+	for {
+		select {
+		case v, opened := <-ch:
+			if !opened { // exit go routine is not sync with drop table
+				return
+			}
+			t.add(v)
+			conf.Log.Debugf("receive data %v for %s", v, topic)
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// Unreg unregisters a topic to remove it from memory store
+func Unreg(topic string) error {
+	// Must be an atomic operation
+	return db.dropTable(topic)
+}
diff --git a/internal/topo/neuron/connection.go b/internal/topo/neuron/connection.go
index 61987c58..10b33051 100644
--- a/internal/topo/neuron/connection.go
+++ b/internal/topo/neuron/connection.go
@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	kctx "github.com/lf-edge/ekuiper/internal/topo/context"
-	"github.com/lf-edge/ekuiper/internal/topo/memory"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
@@ -68,7 +68,7 @@ func createOrGetConnection(sc api.StreamContext, url string) error {
 			return err
 		}
 		sc.GetLogger().Infof("Neuron connected")
-		memory.CreatePub(NeuronTopic)
+		pubsub.CreatePub(NeuronTopic)
 		go run(sctx)
 	}
 	connectionCount++
@@ -79,7 +79,7 @@ func closeConnection(ctx api.StreamContext, url string) error {
 	m.Lock()
 	defer m.Unlock()
 	ctx.GetLogger().Infof("closeConnection count: %d", connectionCount)
-	memory.RemovePub(NeuronTopic)
+	pubsub.RemovePub(NeuronTopic)
 	if connectionCount == 1 {
 		err := disconnect(url)
 		if err != nil {
@@ -114,7 +114,7 @@ func connect(ctx api.StreamContext, url string) error {
 		case mangos.PipeEventDetached:
 			atomic.StoreInt32(&opened, 0)
 			conf.Log.Warnf("neuron connection detached")
-			memory.ProduceError(ctx, NeuronTopic, fmt.Errorf("neuron connection detached"))
+			pubsub.ProduceError(ctx, NeuronTopic, fmt.Errorf("neuron connection detached"))
 		}
 	})
 	//sock.SetOption(mangos.OptionWriteQLen, 100)
@@ -145,7 +145,7 @@ func run(ctx api.StreamContext) {
 				ctx.GetLogger().Errorf("neuron decode message error %v", err)
 				continue
 			}
-			memory.Produce(ctx, NeuronTopic, result)
+			pubsub.Produce(ctx, NeuronTopic, result)
 		} else if err == mangos.ErrClosed {
 			ctx.GetLogger().Infof("neuron connection closed, exit receiving loop")
 			return
diff --git a/internal/topo/neuron/source.go b/internal/topo/neuron/source.go
index ae6bcef0..bd8e2064 100644
--- a/internal/topo/neuron/source.go
+++ b/internal/topo/neuron/source.go
@@ -16,7 +16,7 @@ package neuron
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/memory"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/infra"
@@ -45,8 +45,8 @@ func (s *source) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, er
 		return
 	}
 	defer closeConnection(ctx, s.url)
-	ch := memory.CreateSub(NeuronTopic, nil, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), s.bufferLength)
-	defer memory.CloseSourceConsumerChannel(NeuronTopic, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()))
+	ch := pubsub.CreateSub(NeuronTopic, nil, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), s.bufferLength)
+	defer pubsub.CloseSourceConsumerChannel(NeuronTopic, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()))
 	for {
 		select {
 		case v, opened := <-ch:
diff --git a/internal/topo/node/conf/source.go b/internal/topo/node/conf/source.go
new file mode 100644
index 00000000..e79a8882
--- /dev/null
+++ b/internal/topo/node/conf/source.go
@@ -0,0 +1,77 @@
+// 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 conf
+
+import (
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"strings"
+)
+
+func GetSourceConf(ctx api.StreamContext, sourceType string, options *ast.Options) map[string]interface{} {
+	confkey := options.CONF_KEY
+	logger := ctx.GetLogger()
+	confPath := "sources/" + sourceType + ".yaml"
+	if sourceType == "mqtt" {
+		confPath = "mqtt_source.yaml"
+	}
+	props := make(map[string]interface{})
+	cfg := make(map[string]interface{})
+	err := conf.LoadConfigByName(confPath, &cfg)
+	if err != nil {
+		logger.Warnf("fail to parse yaml for source %s. Return an empty configuration", sourceType)
+	} else {
+		def, ok := cfg["default"]
+		if !ok {
+			logger.Warnf("default conf is not found", confkey)
+		} else {
+			if def1, ok1 := def.(map[string]interface{}); ok1 {
+				props = def1
+			}
+			if c, ok := cfg[strings.ToLower(confkey)]; ok {
+				if c1, ok := c.(map[string]interface{}); ok {
+					c2 := c1
+					for k, v := range c2 {
+						props[k] = v
+					}
+				}
+			}
+		}
+	}
+	f := options.FORMAT
+	if f == "" {
+		f = "json"
+	}
+	props["format"] = strings.ToLower(f)
+	logger.Debugf("get conf for %s with conf key %s: %v", sourceType, confkey, printable(props))
+	return props
+}
+
+func printable(m map[string]interface{}) map[string]interface{} {
+	printableMap := make(map[string]interface{})
+	for k, v := range m {
+		if strings.EqualFold(k, "password") {
+			printableMap[k] = "*"
+		} else {
+			if vm, ok := v.(map[string]interface{}); ok {
+				printableMap[k] = printable(vm)
+			} else {
+				printableMap[k] = v
+			}
+		}
+	}
+	return printableMap
+}
diff --git a/internal/topo/node/lookup_node.go b/internal/topo/node/lookup_node.go
new file mode 100644
index 00000000..8c46d695
--- /dev/null
+++ b/internal/topo/node/lookup_node.go
@@ -0,0 +1,247 @@
+// 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.
+// 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 (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/lookup"
+	"github.com/lf-edge/ekuiper/internal/topo/node/metric"
+	"github.com/lf-edge/ekuiper/internal/xsql"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"github.com/lf-edge/ekuiper/pkg/infra"
+)
+
+// LookupNode will look up the data from the external source when receiving an event
+type LookupNode struct {
+	*defaultSinkNode
+	statManager metric.StatManager
+	sourceType  string
+	joinType    ast.JoinType
+	vals        []ast.Expr
+
+	srcOptions *ast.Options
+	keys       []string
+}
+
+func NewLookupNode(name string, keys []string, joinType ast.JoinType, vals []ast.Expr, srcOptions *ast.Options, options *api.RuleOption) (*LookupNode, error) {
+	t := srcOptions.TYPE
+	if t == "" {
+		return nil, fmt.Errorf("source type is not specified")
+	}
+	n := &LookupNode{
+		keys:       keys,
+		srcOptions: srcOptions,
+		sourceType: t,
+		joinType:   joinType,
+		vals:       vals,
+	}
+	n.defaultSinkNode = &defaultSinkNode{
+		input: make(chan interface{}, options.BufferLength),
+		defaultNode: &defaultNode{
+			outputs:   make(map[string]chan<- interface{}),
+			name:      name,
+			sendError: options.SendError,
+		},
+	}
+	return n, nil
+}
+
+func (n *LookupNode) Exec(ctx api.StreamContext, errCh chan<- error) {
+	n.ctx = ctx
+	log := ctx.GetLogger()
+	log.Debugf("LookupNode %s is started", n.name)
+
+	if len(n.outputs) <= 0 {
+		infra.DrainError(ctx, fmt.Errorf("no output channel found"), errCh)
+		return
+	}
+	stats, err := metric.NewStatManager(ctx, "op")
+	if err != nil {
+		infra.DrainError(ctx, fmt.Errorf("no output channel found"), errCh)
+		return
+	}
+	n.statManager = stats
+	go func() {
+		err := infra.SafeRun(func() error {
+			ns, err := lookup.Attach(n.name)
+			if err != nil {
+				return err
+			}
+			defer lookup.Detach(n.name)
+			fv, _ := xsql.NewFunctionValuersForOp(ctx)
+			// Start the lookup source loop
+			for {
+				log.Debugf("LookupNode %s is looping", n.name)
+				select {
+				// process incoming item from both streams(transformed) and tables
+				case item, opened := <-n.input:
+					processed := false
+					if item, processed = n.preprocess(item); processed {
+						break
+					}
+					n.statManager.IncTotalRecordsIn()
+					n.statManager.ProcessTimeStart()
+					if !opened {
+						n.statManager.IncTotalExceptions("input channel closed")
+						break
+					}
+					switch d := item.(type) {
+					case error:
+						n.Broadcast(d)
+						n.statManager.IncTotalExceptions(d.Error())
+					case xsql.TupleRow:
+						log.Debugf("Lookup Node receive tuple input %s", d)
+						n.statManager.ProcessTimeStart()
+						sets := &xsql.JoinTuples{Content: make([]*xsql.JoinTuple, 0)}
+						err := n.lookup(ctx, d, fv, ns, sets)
+						if err != nil {
+							n.Broadcast(err)
+							n.statManager.IncTotalExceptions(err.Error())
+						} else {
+							n.Broadcast(sets)
+							n.statManager.IncTotalRecordsOut()
+						}
+						n.statManager.ProcessTimeEnd()
+						n.statManager.SetBufferLength(int64(len(n.input)))
+					case *xsql.WindowTuples:
+						log.Debugf("Lookup Node receive window input %s", d)
+						n.statManager.ProcessTimeStart()
+						sets := &xsql.JoinTuples{Content: make([]*xsql.JoinTuple, 0)}
+						err := d.Range(func(i int, r xsql.ReadonlyRow) (bool, error) {
+							tr, ok := r.(xsql.TupleRow)
+							if !ok {
+								return false, fmt.Errorf("Invalid window element, must be a tuple row but got %v", r)
+							}
+							err := n.lookup(ctx, tr, fv, ns, sets)
+							if err != nil {
+								return false, err
+							}
+							return true, nil
+						})
+						if err != nil {
+							n.Broadcast(err)
+							n.statManager.IncTotalExceptions(err.Error())
+						} else {
+							n.Broadcast(sets)
+							n.statManager.IncTotalRecordsOut()
+						}
+						n.statManager.ProcessTimeEnd()
+						n.statManager.SetBufferLength(int64(len(n.input)))
+					default:
+						e := fmt.Errorf("run lookup node error: invalid input type but got %[1]T(%[1]v)", d)
+						n.Broadcast(e)
+						n.statManager.IncTotalExceptions(e.Error())
+					}
+				case <-ctx.Done():
+					log.Infoln("Cancelling lookup node....")
+					return nil
+				}
+			}
+		})
+		if err != nil {
+			infra.DrainError(ctx, err, errCh)
+		}
+	}()
+}
+
+func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.TupleRow, fv *xsql.FunctionValuer, ns api.LookupSource, tuples *xsql.JoinTuples) error {
+	ve := &xsql.ValuerEval{Valuer: xsql.MultiValuer(d, fv)}
+	cvs := make([]interface{}, len(n.vals))
+	hasNil := false
+	for i, val := range n.vals {
+		cvs[i] = ve.Eval(val)
+		if cvs[i] == nil {
+			hasNil = true
+		}
+	}
+	var (
+		r []api.SourceTuple
+		e error
+	)
+	if !hasNil { // if any of the value is nil, the lookup will always return empty result
+		r, e = ns.Lookup(ctx, n.keys, cvs)
+	}
+	if e != nil {
+		return e
+	} else {
+		if len(r) == 0 {
+			if n.joinType == ast.LEFT_JOIN {
+				merged := &xsql.JoinTuple{}
+				merged.AddTuple(d)
+				tuples.Content = append(tuples.Content, merged)
+			} else {
+				ctx.GetLogger().Debugf("Lookup Node %s no result found for tuple %s", n.name, d)
+				return nil
+			}
+		}
+		for _, v := range r {
+			merged := &xsql.JoinTuple{}
+			merged.AddTuple(d)
+			t := &xsql.Tuple{
+				Emitter:   n.name,
+				Message:   v.Message(),
+				Metadata:  v.Meta(),
+				Timestamp: conf.GetNowInMilli(),
+			}
+			merged.AddTuple(t)
+			tuples.Content = append(tuples.Content, merged)
+		}
+		return nil
+	}
+}
+
+func (n *LookupNode) GetMetrics() [][]interface{} {
+	if n.statManager != nil {
+		return [][]interface{}{
+			n.statManager.GetMetrics(),
+		}
+	} else {
+		return nil
+	}
+}
+
+func (n *LookupNode) merge(ctx api.StreamContext, d xsql.TupleRow, r []map[string]interface{}) {
+	n.statManager.ProcessTimeStart()
+	sets := &xsql.JoinTuples{Content: make([]*xsql.JoinTuple, 0)}
+
+	if len(r) == 0 {
+		if n.joinType == ast.LEFT_JOIN {
+			merged := &xsql.JoinTuple{}
+			merged.AddTuple(d)
+			sets.Content = append(sets.Content, merged)
+		} else {
+			ctx.GetLogger().Debugf("Lookup Node %s no result found for tuple %s", n.name, d)
+			return
+		}
+	}
+	for _, v := range r {
+		merged := &xsql.JoinTuple{}
+		merged.AddTuple(d)
+		t := &xsql.Tuple{
+			Emitter:   n.name,
+			Message:   v,
+			Timestamp: conf.GetNowInMilli(),
+		}
+		merged.AddTuple(t)
+		sets.Content = append(sets.Content, merged)
+	}
+
+	n.Broadcast(sets)
+	n.statManager.ProcessTimeEnd()
+	n.statManager.IncTotalRecordsOut()
+	n.statManager.SetBufferLength(int64(len(n.input)))
+}
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
index 1419d464..b33c22d3 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -16,13 +16,10 @@ package node
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/checkpoint"
 	"github.com/lf-edge/ekuiper/internal/topo/node/metric"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
-	"github.com/lf-edge/ekuiper/pkg/ast"
-	"strings"
 )
 
 type OperatorNode interface {
@@ -169,58 +166,3 @@ func (o *defaultSinkNode) preprocess(data interface{}) (interface{}, bool) {
 	}
 	return data, false
 }
-
-func printable(m map[string]interface{}) map[string]interface{} {
-	printableMap := make(map[string]interface{})
-	for k, v := range m {
-		if strings.EqualFold(k, "password") {
-			printableMap[k] = "*"
-		} else {
-			if vm, ok := v.(map[string]interface{}); ok {
-				printableMap[k] = printable(vm)
-			} else {
-				printableMap[k] = v
-			}
-		}
-	}
-	return printableMap
-}
-
-func getSourceConf(ctx api.StreamContext, sourceType string, options *ast.Options) map[string]interface{} {
-	confkey := options.CONF_KEY
-	logger := ctx.GetLogger()
-	confPath := "sources/" + sourceType + ".yaml"
-	if sourceType == "mqtt" {
-		confPath = "mqtt_source.yaml"
-	}
-	props := make(map[string]interface{})
-	cfg := make(map[string]interface{})
-	err := conf.LoadConfigByName(confPath, &cfg)
-	if err != nil {
-		logger.Warnf("fail to parse yaml for source %s. Return an empty configuration", sourceType)
-	} else {
-		def, ok := cfg["default"]
-		if !ok {
-			logger.Warnf("default conf is not found", confkey)
-		} else {
-			if def1, ok1 := def.(map[string]interface{}); ok1 {
-				props = def1
-			}
-			if c, ok := cfg[strings.ToLower(confkey)]; ok {
-				if c1, ok := c.(map[string]interface{}); ok {
-					c2 := c1
-					for k, v := range c2 {
-						props[k] = v
-					}
-				}
-			}
-		}
-	}
-	f := options.FORMAT
-	if f == "" {
-		f = "json"
-	}
-	props["format"] = strings.ToLower(f)
-	logger.Debugf("get conf for %s with conf key %s: %v", sourceType, confkey, printable(props))
-	return props
-}
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index 10b307f5..b3fb4cb6 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -19,6 +19,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/converter"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
+	nodeConf "github.com/lf-edge/ekuiper/internal/topo/node/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/node/metric"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
@@ -72,7 +73,7 @@ func (m *SourceNode) Open(ctx api.StreamContext, errCh chan<- error) {
 	logger.Infof("open source node %s with option %v", m.name, m.options)
 	go func() {
 		panicOrError := infra.SafeRun(func() error {
-			props := getSourceConf(ctx, m.sourceType, m.options)
+			props := nodeConf.GetSourceConf(ctx, m.sourceType, m.options)
 			m.props = props
 			if c, ok := props["concurrency"]; ok {
 				if t, err := cast.ToInt(c, cast.STRICT); err != nil || t <= 0 {
diff --git a/internal/topo/planner/logicalPlan.go b/internal/topo/planner/logicalPlan.go
index a5f69761..1f9f9f75 100644
--- a/internal/topo/planner/logicalPlan.go
+++ b/internal/topo/planner/logicalPlan.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.
@@ -41,7 +41,7 @@ func (p *baseLogicalPlan) SetChildren(children []LogicalPlan) {
 	p.children = children
 }
 
-// By default, push down the predicate to the first child instead of the children
+// PushDownPredicate By default, push down the predicate to the first child instead of the children
 // as most plan cannot have multiple children
 func (p *baseLogicalPlan) PushDownPredicate(condition ast.Expr) (ast.Expr, LogicalPlan) {
 	if len(p.children) == 0 {
diff --git a/internal/topo/planner/lookupPlan.go b/internal/topo/planner/lookupPlan.go
new file mode 100644
index 00000000..d6920b2c
--- /dev/null
+++ b/internal/topo/planner/lookupPlan.go
@@ -0,0 +1,145 @@
+// 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package planner
+
+import (
+	"github.com/lf-edge/ekuiper/pkg/ast"
+)
+
+// LookupPlan is the plan for table lookup and then merged/joined
+type LookupPlan struct {
+	baseLogicalPlan
+
+	joinExpr   ast.Join
+	keys       []string
+	valvars    []ast.Expr
+	options    *ast.Options
+	conditions ast.Expr
+}
+
+// Init must run validateAndExtractCondition before this func
+func (p LookupPlan) Init() *LookupPlan {
+	p.baseLogicalPlan.self = &p
+	return &p
+}
+
+// PushDownPredicate do not deal with conditions, push down or return up
+func (p *LookupPlan) PushDownPredicate(condition ast.Expr) (ast.Expr, LogicalPlan) {
+	a := combine(condition, p.conditions)
+	if len(p.children) == 0 {
+		return a, p.self
+	}
+	rest, _ := p.baseLogicalPlan.PushDownPredicate(a)
+	// Swallow all filter conditions. If there are other filter plans, there may have multiple filters
+	if rest != nil {
+		// Add a filter plan for children
+		f := FilterPlan{
+			condition: rest,
+		}.Init()
+		f.SetChildren([]LogicalPlan{p})
+		return nil, f
+	}
+	return nil, p.self
+}
+
+// validateAndExtractCondition Make sure the join condition is equi-join and extreact other conditions
+func (p *LookupPlan) validateAndExtractCondition() bool {
+	equi, conditions := flatConditions(p.joinExpr.Expr)
+	// No equal predict condition found
+	if len(equi) == 0 {
+		return false
+	}
+	if len(conditions) > 0 {
+		p.conditions = conditions[0]
+		for _, c := range conditions[1:] {
+			p.conditions = &ast.BinaryExpr{OP: ast.AND, LHS: p.conditions, RHS: c}
+		}
+	}
+
+	strName := p.joinExpr.Name
+	kset := make(map[string]struct{})
+	// Extract equi-join condition
+	for _, c := range equi {
+		lref, lok := c.LHS.(*ast.FieldRef)
+		rref, rok := c.RHS.(*ast.FieldRef)
+		if lok && rok {
+			if lref.StreamName == rref.StreamName {
+				continue
+			}
+			if string(lref.StreamName) == strName {
+				if _, ok := kset[lref.Name]; ok {
+					return false
+				}
+				kset[lref.Name] = struct{}{}
+				p.valvars = append(p.valvars, rref)
+			} else if string(rref.StreamName) == strName {
+				if _, ok := kset[rref.Name]; ok {
+					return false
+				}
+				kset[rref.Name] = struct{}{}
+				p.valvars = append(p.valvars, lref)
+			} else {
+				continue
+			}
+		} else if lok {
+			if string(lref.StreamName) == strName {
+				if _, ok := kset[lref.Name]; ok {
+					return false
+				}
+				kset[lref.Name] = struct{}{}
+				p.valvars = append(p.valvars, c.RHS)
+			} else {
+				continue
+			}
+		} else if rok {
+			if string(rref.StreamName) == strName {
+				if _, ok := kset[rref.Name]; ok {
+					return false
+				}
+				kset[rref.Name] = struct{}{}
+				p.valvars = append(p.valvars, c.LHS)
+			} else {
+				continue
+			}
+		} else {
+			continue
+		}
+	}
+	if len(kset) > 0 {
+		p.keys = make([]string, 0, len(kset))
+		for k := range kset {
+			p.keys = append(p.keys, k)
+		}
+		return true
+	}
+	return false
+}
+
+// flatConditions flat the join condition. Only binary condition of EQ and AND are allowed
+func flatConditions(condition ast.Expr) ([]*ast.BinaryExpr, []ast.Expr) {
+	if be, ok := condition.(*ast.BinaryExpr); ok {
+		switch be.OP {
+		case ast.EQ:
+			return []*ast.BinaryExpr{be}, []ast.Expr{}
+		case ast.AND:
+			e1, e2 := flatConditions(be.LHS)
+			e3, e4 := flatConditions(be.RHS)
+			return append(e1, e3...), append(e2, e4...)
+		default:
+			return []*ast.BinaryExpr{}, []ast.Expr{condition}
+		}
+	}
+	return []*ast.BinaryExpr{}, []ast.Expr{condition}
+}
diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
index 12eebd4d..6baa3a5e 100644
--- a/internal/topo/planner/planner.go
+++ b/internal/topo/planner/planner.go
@@ -171,6 +171,8 @@ func buildOps(lp LogicalPlan, tp *topo.Topo, options *api.RuleOption, sources []
 		if err != nil {
 			return nil, 0, err
 		}
+	case *LookupPlan:
+		op, err = node.NewLookupNode(t.joinExpr.Name, t.keys, t.joinExpr.JoinType, t.valvars, t.options, options)
 	case *JoinAlignPlan:
 		op, err = node.NewJoinAlignNode(fmt.Sprintf("%d_join_aligner", newIndex), t.Emitters, options)
 	case *JoinPlan:
@@ -213,10 +215,11 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *api.RuleOption, store kv.
 		p        LogicalPlan
 		children []LogicalPlan
 		// If there are tables, the plan graph will be different for join/window
-		tableChildren []LogicalPlan
-		tableEmitters []string
-		w             *ast.Window
-		ds            ast.Dimensions
+		lookupTableChildren map[string]*ast.Options
+		scanTableChildren   []LogicalPlan
+		scanTableEmitters   []string
+		w                   *ast.Window
+		ds                  ast.Dimensions
 	)
 
 	streamStmts, err := decorateStmt(stmt, store)
@@ -225,17 +228,24 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *api.RuleOption, store kv.
 	}
 
 	for _, streamStmt := range streamStmts {
-		p = DataSourcePlan{
-			name:       streamStmt.Name,
-			streamStmt: streamStmt,
-			iet:        opt.IsEventTime,
-			allMeta:    opt.SendMetaToSink,
-		}.Init()
-		if streamStmt.StreamType == ast.TypeStream {
-			children = append(children, p)
+		if streamStmt.StreamType == ast.TypeTable && streamStmt.Options.KIND == ast.StreamKindLookup {
+			if lookupTableChildren == nil {
+				lookupTableChildren = make(map[string]*ast.Options)
+			}
+			lookupTableChildren[string(streamStmt.Name)] = streamStmt.Options
 		} else {
-			tableChildren = append(tableChildren, p)
-			tableEmitters = append(tableEmitters, string(streamStmt.Name))
+			p = DataSourcePlan{
+				name:       streamStmt.Name,
+				streamStmt: streamStmt,
+				iet:        opt.IsEventTime,
+				allMeta:    opt.SendMetaToSink,
+			}.Init()
+			if streamStmt.StreamType == ast.TypeStream {
+				children = append(children, p)
+			} else {
+				scanTableChildren = append(scanTableChildren, p)
+				scanTableEmitters = append(scanTableEmitters, string(streamStmt.Name))
+			}
 		}
 	}
 	if dimensions != nil {
@@ -252,7 +262,7 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *api.RuleOption, store kv.
 			if w.Interval != nil {
 				wp.interval = w.Interval.Val
 			} else if w.WindowType == ast.COUNT_WINDOW {
-				//if no interval value is set and it's count window, then set interval to length value.
+				//if no interval value is set, and it's count window, then set interval to length value.
 				wp.interval = w.Length.Val
 			}
 			if w.Filter != nil {
@@ -266,22 +276,50 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *api.RuleOption, store kv.
 		}
 	}
 	if stmt.Joins != nil {
-		if len(tableChildren) > 0 {
-			p = JoinAlignPlan{
-				Emitters: tableEmitters,
+		if len(lookupTableChildren) == 0 && len(scanTableChildren) == 0 && w == nil {
+			return nil, errors.New("a time window or count window is required to join multiple streams")
+		}
+		if len(lookupTableChildren) > 0 {
+			var joins []ast.Join
+			for _, join := range stmt.Joins {
+				if streamOpt, ok := lookupTableChildren[join.Name]; ok {
+					lookupPlan := LookupPlan{
+						joinExpr: join,
+						options:  streamOpt,
+					}
+					if !lookupPlan.validateAndExtractCondition() {
+						return nil, fmt.Errorf("join condition %s is invalid, at least one equi-join predicate is required", join.Expr)
+					}
+					p = lookupPlan.Init()
+					p.SetChildren(children)
+					children = []LogicalPlan{p}
+					delete(lookupTableChildren, join.Name)
+				} else {
+					joins = append(joins, join)
+				}
+			}
+			if len(lookupTableChildren) > 0 {
+				return nil, fmt.Errorf("cannot find lookup table %v in any join", lookupTableChildren)
+			}
+			stmt.Joins = joins
+		}
+		// Not all joins are lookup joins, so we need to create a join plan for the remaining joins
+		if len(stmt.Joins) > 0 {
+			if len(scanTableChildren) > 0 {
+				p = JoinAlignPlan{
+					Emitters: scanTableEmitters,
+				}.Init()
+				p.SetChildren(append(children, scanTableChildren...))
+				children = []LogicalPlan{p}
+			}
+			// TODO extract on filter
+			p = JoinPlan{
+				from:  stmt.Sources[0].(*ast.Table),
+				joins: stmt.Joins,
 			}.Init()
-			p.SetChildren(append(children, tableChildren...))
+			p.SetChildren(children)
 			children = []LogicalPlan{p}
-		} else if w == nil {
-			return nil, errors.New("a time window or count window is required to join multiple streams")
 		}
-		// TODO extract on filter
-		p = JoinPlan{
-			from:  stmt.Sources[0].(*ast.Table),
-			joins: stmt.Joins,
-		}.Init()
-		p.SetChildren(children)
-		children = []LogicalPlan{p}
 	}
 	if stmt.Condition != nil {
 		p = FilterPlan{
diff --git a/internal/topo/source/httppush_source.go b/internal/topo/source/httppush_source.go
index d6fc7e30..5705b166 100644
--- a/internal/topo/source/httppush_source.go
+++ b/internal/topo/source/httppush_source.go
@@ -17,7 +17,7 @@ package source
 import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/topo/memory"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/source/httpserver"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
@@ -70,8 +70,8 @@ func (hps *HTTPPushSource) Open(ctx api.StreamContext, consumer chan<- api.Sourc
 		return
 	}
 	defer httpserver.UnregisterEndpoint(hps.conf.Endpoint)
-	ch := memory.CreateSub(t, nil, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), hps.conf.BufferLength)
-	defer memory.CloseSourceConsumerChannel(t, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()))
+	ch := pubsub.CreateSub(t, nil, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), hps.conf.BufferLength)
+	defer pubsub.CloseSourceConsumerChannel(t, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()))
 	for {
 		select {
 		case <-done: // http data server error
diff --git a/internal/topo/source/httpserver/data_server.go b/internal/topo/source/httpserver/data_server.go
index 94a94b99..6e86b860 100644
--- a/internal/topo/source/httpserver/data_server.go
+++ b/internal/topo/source/httpserver/data_server.go
@@ -21,7 +21,7 @@ import (
 	"github.com/gorilla/mux"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	kctx "github.com/lf-edge/ekuiper/internal/topo/context"
-	"github.com/lf-edge/ekuiper/internal/topo/memory"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"net/http"
@@ -75,7 +75,7 @@ func RegisterEndpoint(endpoint string, method string, _ string) (string, chan st
 		return "", nil, err
 	}
 	topic := TopicPrefix + endpoint
-	memory.CreatePub(topic)
+	pubsub.CreatePub(topic)
 	router.HandleFunc(endpoint, func(w http.ResponseWriter, r *http.Request) {
 		sctx.GetLogger().Debugf("receive http request: %s", r.URL.String())
 		defer r.Body.Close()
@@ -83,11 +83,11 @@ func RegisterEndpoint(endpoint string, method string, _ string) (string, chan st
 		err := json.NewDecoder(r.Body).Decode(&m)
 		if err != nil {
 			handleError(w, err, "Fail to decode data")
-			memory.ProduceError(sctx, topic, fmt.Errorf("fail to decode data %s: %v", r.Body, err))
+			pubsub.ProduceError(sctx, topic, fmt.Errorf("fail to decode data %s: %v", r.Body, err))
 			return
 		}
 		sctx.GetLogger().Debugf("httppush received message %s", m)
-		memory.Produce(sctx, topic, m)
+		pubsub.Produce(sctx, topic, m)
 		w.WriteHeader(http.StatusOK)
 		_, _ = w.Write([]byte("ok"))
 	}).Methods(method)
@@ -97,7 +97,7 @@ func RegisterEndpoint(endpoint string, method string, _ string) (string, chan st
 func UnregisterEndpoint(endpoint string) {
 	lock.Lock()
 	defer lock.Unlock()
-	memory.RemovePub(TopicPrefix + endpoint)
+	pubsub.RemovePub(TopicPrefix + endpoint)
 	refCount--
 	// TODO async close server
 	if refCount == 0 {
diff --git a/internal/xsql/lexical.go b/internal/xsql/lexical.go
index a27fa3e0..81c9957c 100644
--- a/internal/xsql/lexical.go
+++ b/internal/xsql/lexical.go
@@ -283,6 +283,8 @@ func (s *Scanner) ScanIdent() (tok ast.Token, lit string) {
 		return ast.SHARED, lit
 	case "SCHEMAID":
 		return ast.SCHEMAID, lit
+	case "KIND":
+		return ast.KIND, lit
 	case "DD":
 		return ast.DD, lit
 	case "HH":
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index 0c560404..f4d0babe 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -1406,7 +1406,7 @@ func (p *Parser) parseStreamOptions() (*ast.Options, error) {
 	if tok, lit := p.scanIgnoreWhitespace(); tok == ast.LPAREN {
 		lStack.Push(ast.LPAREN)
 		for {
-			if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.DATASOURCE || tok1 == ast.FORMAT || tok1 == ast.KEY || tok1 == ast.CONF_KEY || tok1 == ast.STRICT_VALIDATION || tok1 == ast.TYPE || tok1 == ast.TIMESTAMP || tok1 == ast.TIMESTAMP_FORMAT || tok1 == ast.RETAIN_SIZE || tok1 == ast.SHARED || tok1 == ast.SCHEMAID {
+			if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.DATASOURCE || tok1 == ast.FORMAT || tok1 == ast.KEY || tok1 == ast.CONF_KEY || tok1 == ast.STRICT_VALIDATION || tok1 == ast.TYPE || tok1 == ast.TIMESTAMP || tok1 == ast.TIMESTAMP_FORMAT || tok1 == ast.RETAIN_SIZE || tok1 == ast.SHARED || tok1 == ast.SCHEMAID || tok1 == ast.KIND {
 				if tok2, lit2 := p.scanIgnoreWhitespace(); tok2 == ast.EQ {
 					if tok3, lit3 := p.scanIgnoreWhitespace(); tok3 == ast.STRING {
 						switch tok1 {
@@ -1428,6 +1428,9 @@ func (p *Parser) parseStreamOptions() (*ast.Options, error) {
 							} else {
 								opts.SHARED = val == "TRUE"
 							}
+						case ast.KIND:
+							val := strings.ToLower(lit3)
+							opts.KIND = val
 						default:
 							f := v.Elem().FieldByName(lit1)
 							if f.IsValid() {
diff --git a/internal/xsql/row.go b/internal/xsql/row.go
index 4dda1f07..a4498531 100644
--- a/internal/xsql/row.go
+++ b/internal/xsql/row.go
@@ -18,7 +18,6 @@ import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"strings"
-	"sync"
 )
 
 // The original message map may be big. Make sure it is immutable so that never make a copy of it.
@@ -192,7 +191,6 @@ type Tuple struct {
 	Metadata  Metadata // immutable
 
 	AffiliateRow
-	lock sync.Mutex
 
 	cachedMap map[string]interface{} // clone of the row and cached for performance
 }
@@ -330,8 +328,6 @@ func (t *Tuple) Clone() TupleRow {
 
 // ToMap should only use in sink.
 func (t *Tuple) ToMap() map[string]interface{} {
-	t.lock.Lock()
-	defer t.lock.Unlock()
 	if t.AffiliateRow.IsEmpty() {
 		return t.Message
 	}
@@ -341,7 +337,6 @@ func (t *Tuple) ToMap() map[string]interface{} {
 			m[k] = v
 		}
 		t.cachedMap = m
-		t.Message = t.cachedMap
 	}
 	t.AffiliateRow.MergeMap(t.cachedMap)
 	return t.cachedMap
@@ -377,18 +372,18 @@ func (t *Tuple) Pick(allWildcard bool, cols [][]string, wildcardEmitters map[str
 	}
 	if !allWildcard {
 		if len(cols) > 0 {
-			t.cachedMap = make(map[string]interface{})
+			pickedMap := make(map[string]interface{})
 			for _, colTab := range cols {
 				if colTab[1] == "" || colTab[1] == string(ast.DefaultStream) || colTab[1] == t.Emitter {
 					if v, ok := t.Message.Value(colTab[0], colTab[1]); ok {
-						t.cachedMap[colTab[0]] = v
+						pickedMap[colTab[0]] = v
 					}
 				}
 			}
-			t.Message = t.cachedMap
+			t.Message = pickedMap
 		} else {
 			t.Message = make(map[string]interface{})
-			t.cachedMap = t.Message
+			t.cachedMap = make(map[string]interface{})
 		}
 	}
 }
diff --git a/internal/xsql/stmtx.go b/internal/xsql/stmtx.go
index 7dbb69cd..46d83409 100644
--- a/internal/xsql/stmtx.go
+++ b/internal/xsql/stmtx.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.
@@ -27,6 +27,7 @@ func GetStreams(stmt *ast.SelectStatement) (result []string) {
 	if stmt == nil {
 		return nil
 	}
+	// TODO sources must be a stream
 	for _, source := range stmt.Sources {
 		if s, ok := source.(*ast.Table); ok {
 			result = append(result, s.Name)
diff --git a/pkg/api/stream.go b/pkg/api/stream.go
index 9a5b3fe1..7bcf2791 100644
--- a/pkg/api/stream.go
+++ b/pkg/api/stream.go
@@ -78,12 +78,15 @@ type Source interface {
 	Closable
 }
 
-type TableSource interface {
-	// Load the data at batch
-	Load(ctx StreamContext) ([]SourceTuple, error)
+type LookupSource interface {
+	// Open creates the connection to the external data source
+	Open(ctx StreamContext) error
 	// Configure Called during initialization. Configure the source with the data source(e.g. topic for mqtt) and the properties
 	//read from the yaml
 	Configure(datasource string, props map[string]interface{}) error
+	// Lookup receive lookup values to construct the query and return query results
+	Lookup(ctx StreamContext, keys []string, values []interface{}) ([]SourceTuple, error)
+	Closable
 }
 
 type Sink interface {
diff --git a/pkg/ast/sourceStmt.go b/pkg/ast/sourceStmt.go
index 6ba206ea..4d43ad90 100644
--- a/pkg/ast/sourceStmt.go
+++ b/pkg/ast/sourceStmt.go
@@ -29,6 +29,11 @@ var StreamTypeMap = map[StreamType]string{
 	TypeTable:  "table",
 }
 
+const (
+	StreamKindLookup = "lookup"
+	StreamKindScan   = "scan"
+)
+
 type StreamType int
 
 type StreamStmt struct {
@@ -86,9 +91,12 @@ type Options struct {
 	STRICT_VALIDATION bool   `json:"strictValidation,omitempty"`
 	TIMESTAMP         string `json:"timestamp,omitempty"`
 	TIMESTAMP_FORMAT  string `json:"timestampFormat,omitempty"`
-	RETAIN_SIZE       int    `json:"retainSize,omitempty"`
 	SHARED            bool   `json:"shared,omitempty"`
 	SCHEMAID          string `json:"schemaid,omitempty"`
+	// for scan table only
+	RETAIN_SIZE int `json:"retainSize,omitempty"`
+	// for table only, to distinguish lookup & scan
+	KIND string `json:"kind,omitempty"`
 }
 
 func (o Options) node() {}
diff --git a/pkg/ast/token.go b/pkg/ast/token.go
index 835485e1..e5c2b43a 100644
--- a/pkg/ast/token.go
+++ b/pkg/ast/token.go
@@ -136,6 +136,7 @@ const (
 	RETAIN_SIZE
 	SHARED
 	SCHEMAID
+	KIND
 
 	DD
 	HH


Test patch:
diff --git a/internal/plugin/portable/test/portable_rule_test.go b/internal/plugin/portable/test/portable_rule_test.go
index 5b3a2c31..e239809c 100644
--- a/internal/plugin/portable/test/portable_rule_test.go
+++ b/internal/plugin/portable/test/portable_rule_test.go
@@ -15,9 +15,7 @@
 package test
 
 import (
-	"bufio"
 	"context"
-	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/binder"
 	"github.com/lf-edge/ekuiper/internal/binder/function"
@@ -26,11 +24,11 @@ import (
 	"github.com/lf-edge/ekuiper/internal/plugin/portable/runtime"
 	"github.com/lf-edge/ekuiper/internal/processor"
 	"github.com/lf-edge/ekuiper/internal/topo"
+	"github.com/lf-edge/ekuiper/internal/topo/memory"
 	"github.com/lf-edge/ekuiper/internal/topo/planner"
 	"github.com/lf-edge/ekuiper/internal/topo/topotest"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"log"
-	"os"
 	"reflect"
 	"testing"
 	"time"
@@ -63,27 +61,24 @@ func TestSourceAndFunc(t *testing.T) {
 	}{
 		{
 			Name: `TestPortableRule1`,
-			Rule: `{"sql":"SELECT echo(count) as ee FROM ext","actions":[{"file":{"path":"cache1"}}]}`,
+			Rule: `{"sql":"SELECT count as ee FROM ext","actions":[{"memory":{"topic":"cache"}}]}`,
 			R: [][]map[string]interface{}{
 				{{
-					"ee": float64(50),
+					"ee": 50,
 				}},
 				{{
-					"ee": float64(50),
+					"ee": 50,
 				}},
 				{{
-					"ee": float64(50),
+					"ee": 50,
 				}},
 			},
 			M: map[string]interface{}{
-				"source_ext_0_exceptions_total":   int64(0),
-				"source_ext_0_records_in_total":   int64(3),
-				"source_ext_0_records_out_total":  int64(3),
-				"sink_file_0_0_records_out_total": int64(3),
+				"sink_memory_0_0_records_out_total": int64(3),
 			},
 		}, {
 			Name: `TestPythonRule`,
-			Rule: `{"sql":"SELECT revert(name) as ee FROM extpy","actions":[{"file":{"path":"cache2"}},{"print":{}}]}`,
+			Rule: `{"sql":"SELECT revert(name) as ee FROM extpy","actions":[{"memory":{"topic":"cache"}},{"print":{}}]}`,
 			R: [][]map[string]interface{}{
 				{{
 					"ee": "nosjyp",
@@ -96,11 +91,7 @@ func TestSourceAndFunc(t *testing.T) {
 				}},
 			},
 			M: map[string]interface{}{
-				"source_extpy_0_exceptions_total":  int64(0),
-				"source_extpy_0_records_in_total":  int64(3),
-				"source_extpy_0_records_out_total": int64(3),
-				"sink_file_0_0_records_out_total":  int64(3),
-				"sink_print_1_0_records_out_total": int64(3),
+				"sink_memory_0_0_records_out_total": int64(3),
 			},
 		},
 	}
@@ -108,7 +99,6 @@ func TestSourceAndFunc(t *testing.T) {
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	defer runtime.GetPluginInsManager().KillAll()
 	for i, tt := range tests {
-		_ = os.Remove(fmt.Sprintf("cache%d", i+1))
 		topotest.HandleStream(true, streamList[i:i+1], t)
 		rs, err := CreateRule(tt.Name, tt.Rule)
 		if err != nil {
@@ -120,7 +110,9 @@ func TestSourceAndFunc(t *testing.T) {
 			t.Errorf("fail to init rule: %v", err)
 			continue
 		}
-		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		var mm [][]map[string]interface{}
+		ch := memory.CreateSub("cache", nil, fmt.Sprintf("cache%d", i+1), 10)
+		ctx, cancel := context.WithTimeout(context.Background(), 10000*time.Second)
 		go func(ctx context.Context) {
 			select {
 			case err := <-tp.Open():
@@ -132,6 +124,18 @@ func TestSourceAndFunc(t *testing.T) {
 			}
 			fmt.Println("all exit")
 		}(ctx)
+		go func(ctx context.Context) {
+			for {
+				select {
+				case s := <-ch:
+					log.Printf("get %v", s)
+					mm = append(mm, []map[string]interface{}{s.Message()})
+				case <-ctx.Done():
+					log.Printf("ctx done %v\n", ctx.Err())
+					return
+				}
+			}
+		}(ctx)
 		topotest.HandleStream(false, streamList[i:i+1], t)
 		for {
 			if ctx.Err() != nil {
@@ -141,24 +145,6 @@ func TestSourceAndFunc(t *testing.T) {
 			time.Sleep(10 * time.Millisecond)
 			if compareMetrics(tp, tt.M) {
 				cancel()
-				// need to wait for file results
-				time.Sleep(1 * time.Second)
-				results := getResults(fmt.Sprintf("cache%d", i+1))
-				fmt.Printf("get results %v\n", results)
-				time.Sleep(10 * time.Millisecond)
-				var mm [][]map[string]interface{}
-				for i, v := range results {
-					if i >= 3 {
-						break
-					}
-					var mapRes []map[string]interface{}
-					err := json.Unmarshal([]byte(v), &mapRes)
-					if err != nil {
-						t.Errorf("Failed to parse the input into map")
-						continue
-					}
-					mm = append(mm, mapRes)
-				}
 				if !reflect.DeepEqual(tt.R, mm) {
 					t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.Rule, tt.R, mm)
 				}
@@ -166,8 +152,6 @@ func TestSourceAndFunc(t *testing.T) {
 			}
 		}
 	}
-	// wait for rule clean up
-	time.Sleep(1 * time.Second)
 }
 
 func compareMetrics(tp *topo.Topo, m map[string]interface{}) bool {
@@ -198,23 +182,6 @@ func compareMetrics(tp *topo.Topo, m map[string]interface{}) bool {
 	return true
 }
 
-func getResults(fileName string) []string {
-	f, err := os.Open(fileName)
-	if err != nil {
-		panic(err)
-	}
-	result := make([]string, 0)
-	scanner := bufio.NewScanner(f)
-	for scanner.Scan() {
-		result = append(result, scanner.Text())
-	}
-	if err := scanner.Err(); err != nil {
-		panic(err)
-	}
-	f.Close()
-	return result
-}
-
 func CreateRule(name, sql string) (*api.Rule, error) {
 	p := processor.NewRuleProcessor()
 	p.ExecDrop(name)
diff --git a/internal/topo/lookup/table_test.go b/internal/topo/lookup/table_test.go
new file mode 100644
index 00000000..7422ef15
--- /dev/null
+++ b/internal/topo/lookup/table_test.go
@@ -0,0 +1,98 @@
+// 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 lookup
+
+import (
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"sync"
+	"testing"
+)
+
+func TestTable(t *testing.T) {
+	var wg sync.WaitGroup
+	wg.Add(2)
+	err := CreateInstance("test1", "memory", &ast.Options{
+		DATASOURCE: "test",
+		TYPE:       "memory",
+		KIND:       "lookup",
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = CreateInstance("test2", "memory", &ast.Options{
+		DATASOURCE: "test2",
+		TYPE:       "memory",
+		KIND:       "lookup",
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	go func() {
+		for i := 0; i < 3; i++ {
+			_, err := Attach("test1")
+			if err != nil {
+				t.Error(err)
+			}
+			_, err = Attach("test2")
+			if err != nil {
+				t.Error(err)
+			}
+		}
+		wg.Done()
+	}()
+	go func() {
+		for i := 0; i < 3; i++ {
+			_, err := Attach("test2")
+			if err != nil {
+				t.Error(err)
+			}
+			_, err = Attach("test1")
+			if err != nil {
+				t.Error(err)
+			}
+		}
+		wg.Done()
+	}()
+	wg.Wait()
+	if len(instances) != 2 {
+		t.Errorf("expect 2 instances, but got %d", len(instances))
+		return
+	}
+	for _, ins := range instances {
+		if ins.count != 6 {
+			t.Errorf("expect 6 count, but got %d", ins.count)
+			return
+		}
+	}
+	err = DropInstance("test1")
+	if err == nil {
+		t.Error("should have error to drop instance")
+		return
+	}
+	for i := 0; i < 6; i++ {
+		Detach("test1")
+	}
+	err = DropInstance("test1")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if len(instances) != 1 {
+		t.Errorf("expect 2 instances, but got %d", len(instances))
+		return
+	}
+}
diff --git a/internal/topo/memory/lookupsource_test.go b/internal/topo/memory/lookupsource_test.go
new file mode 100644
index 00000000..a2fc398a
--- /dev/null
+++ b/internal/topo/memory/lookupsource_test.go
@@ -0,0 +1,122 @@
+// 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 (
+	gocontext "context"
+	"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"
+	"time"
+)
+
+func TestNoIndexLookup(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"})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = ls.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	// 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.Produce(ctx, "test", map[string]interface{}{"ff": "value1", "gg": "value4"})
+	// wait for table accumulation
+	time.Sleep(100 * time.Millisecond)
+	canctx, cancel := gocontext.WithCancel(gocontext.Background())
+	defer cancel()
+	go func() {
+		for {
+			select {
+			case <-canctx.Done():
+				return
+			case <-time.After(10 * time.Millisecond):
+				pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value4", "gg": "value2"})
+			}
+		}
+	}()
+	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{"ff"}, []interface{}{"value1"})
+	if !reflect.DeepEqual(result, expected) {
+		t.Errorf("expect %v but got %v", expected, result)
+	}
+	err = ls.Close(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+}
+
+func TestSingleIndexLookup(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"}})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = ls.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	// 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": "value1", "gg": "value4"})
+	// wait for table accumulation
+	time.Sleep(100 * time.Millisecond)
+	canctx, cancel := gocontext.WithCancel(gocontext.Background())
+	defer cancel()
+	go func() {
+		for {
+			select {
+			case <-canctx.Done():
+				return
+			case <-time.After(10 * time.Millisecond):
+				pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value4", "gg": "value2"})
+			}
+		}
+	}()
+	expected := []api.SourceTuple{
+		api.NewDefaultSourceTuple(map[string]interface{}{"ff": "value1", "gg": "value2"}, map[string]interface{}{"topic": "test2"}),
+		api.NewDefaultSourceTuple(map[string]interface{}{"ff": "value1", "gg": "value4"}, map[string]interface{}{"topic": "test2"}),
+	}
+	result, err := ls.Lookup(ctx, []string{"ff"}, []interface{}{"value1"})
+	if !reflect.DeepEqual(result, expected) {
+		t.Errorf("expect %v but got %v", expected, result)
+	}
+	err = ls.Close(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+}
diff --git a/internal/topo/memory/manager_test.go b/internal/topo/memory/memory_test.go
similarity index 78%
rename from internal/topo/memory/manager_test.go
rename to internal/topo/memory/memory_test.go
index c087677b..cf9ef456 100644
--- a/internal/topo/memory/manager_test.go
+++ b/internal/topo/memory/memory_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2022 EMQ Technologies Co., Ltd.
+// 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.
@@ -19,25 +19,19 @@ import (
 	"github.com/gdexlab/go-render/render"
 	"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/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"reflect"
-	"regexp"
-	"strings"
 	"testing"
 	"time"
 )
 
-func reset() {
-	pubTopics = make(map[string]*pubConsumers)
-	subExps = make(map[string]*subChan)
-}
-
 func TestSharedInmemoryNode(t *testing.T) {
-	reset()
+	pubsub.Reset()
 	id := "test_id"
 	sinkProps := make(map[string]interface{})
-	sinkProps[IdProperty] = id
+	sinkProps[pubsub.IdProperty] = id
 	src := GetSource()
 	snk := GetSink()
 	contextLogger := conf.Log.WithField("rule", "test")
@@ -56,7 +50,7 @@ func TestSharedInmemoryNode(t *testing.T) {
 		t.Error(err)
 		return
 	}
-	srcProps[IdProperty] = id
+	srcProps[pubsub.IdProperty] = id
 	err = src.Configure(id, srcProps)
 	if err != nil {
 		t.Error(err)
@@ -65,9 +59,9 @@ func TestSharedInmemoryNode(t *testing.T) {
 		src.Open(ctx, consumer, errorChannel)
 	}()
 
-	if _, contains := pubTopics[id]; !contains {
-		t.Errorf("there should be memory node for topic")
-	}
+	//if _, contains := pubTopics[id]; !contains {
+	//	t.Errorf("there should be memory node for topic")
+	//}
 
 	data := make(map[string]interface{})
 	data["temperature"] = 33.0
@@ -92,94 +86,8 @@ func TestSharedInmemoryNode(t *testing.T) {
 	}
 }
 
-func TestCreateAndClose(t *testing.T) {
-	reset()
-	var (
-		sourceTopics = []string{"h/d1/c1/s2", "h/+/+/s1", "h/d3/#", "h/d1/c1/s2", "h/+/c1/s1"}
-		sinkTopics   = []string{"h/d1/c1/s1", "h/d1/c1/s2", "h/d2/c2/s1", "h/d3/c3/s1", "h/d1/c1/s1"}
-		chans        []chan api.SourceTuple
-	)
-	for i, topic := range sinkTopics {
-		CreatePub(topic)
-		var (
-			r   *regexp.Regexp
-			err error
-		)
-		if strings.ContainsAny(sourceTopics[i], "+#") {
-			r, err = getRegexp(sourceTopics[i])
-			if err != nil {
-				t.Error(err)
-				return
-			}
-		}
-		c := CreateSub(sourceTopics[i], r, fmt.Sprintf("%d", i), 100)
-		chans = append(chans, c)
-	}
-
-	expPub := map[string]*pubConsumers{
-		"h/d1/c1/s1": {
-			count: 2,
-			consumers: map[string]chan api.SourceTuple{
-				"1": chans[1],
-				"4": chans[4],
-			},
-		},
-		"h/d1/c1/s2": {
-			count: 1,
-			consumers: map[string]chan api.SourceTuple{
-				"0": chans[0],
-				"3": chans[3],
-			},
-		},
-		"h/d2/c2/s1": {
-			count: 1,
-			consumers: map[string]chan api.SourceTuple{
-				"1": chans[1],
-			},
-		},
-		"h/d3/c3/s1": {
-			count: 1,
-			consumers: map[string]chan api.SourceTuple{
-				"1": chans[1],
-				"2": chans[2],
-			},
-		},
-	}
-	if !reflect.DeepEqual(expPub, pubTopics) {
-		t.Errorf("Error adding: Expect\n\t%v\nbut got\n\t%v", render.AsCode(expPub), render.AsCode(pubTopics))
-		return
-	}
-	i := 0
-	for i < 3 {
-		CloseSourceConsumerChannel(sourceTopics[i], fmt.Sprintf("%d", i))
-		RemovePub(sinkTopics[i])
-		i++
-	}
-	expPub = map[string]*pubConsumers{
-		"h/d1/c1/s1": {
-			count: 1,
-			consumers: map[string]chan api.SourceTuple{
-				"4": chans[4],
-			},
-		},
-		"h/d1/c1/s2": {
-			count: 0,
-			consumers: map[string]chan api.SourceTuple{
-				"3": chans[3],
-			},
-		},
-		"h/d3/c3/s1": {
-			count:     1,
-			consumers: map[string]chan api.SourceTuple{},
-		},
-	}
-	if !reflect.DeepEqual(expPub, pubTopics) {
-		t.Errorf("Error closing: Expect\n\t%v\nbut got\n\t %v", render.AsCode(expPub), render.AsCode(pubTopics))
-	}
-}
-
 func TestMultipleTopics(t *testing.T) {
-	reset()
+	pubsub.Reset()
 	var (
 		sourceTopics = []string{"h/d1/c1/s2", "h/+/+/s1", "h/d3/#", "h/d1/c1/s2", "h/+/c1/s1"}
 		sinkTopics   = []string{"h/d1/c1/s1", "h/d1/c1/s2", "h/d2/c2/s1", "h/d3/c3/s1"}
@@ -480,7 +388,7 @@ func TestMultipleTopics(t *testing.T) {
 		for c < 3 {
 			for i, v := range sinkData {
 				time.Sleep(10 * time.Millisecond)
-				Produce(ctx, sinkTopics[i], v[c])
+				pubsub.Produce(ctx, sinkTopics[i], v[c])
 			}
 			c++
 		}
diff --git a/internal/topo/memory/pubsub/manager_test.go b/internal/topo/memory/pubsub/manager_test.go
new file mode 100644
index 00000000..14e6a536
--- /dev/null
+++ b/internal/topo/memory/pubsub/manager_test.go
@@ -0,0 +1,126 @@
+// 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.
+// 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 (
+	"fmt"
+	"github.com/gdexlab/go-render/render"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"reflect"
+	"regexp"
+	"strings"
+	"testing"
+)
+
+func TestCreateAndClose(t *testing.T) {
+	Reset()
+	var (
+		sourceTopics = []string{"h/d1/c1/s2", "h/+/+/s1", "h/d3/#", "h/d1/c1/s2", "h/+/c1/s1"}
+		sinkTopics   = []string{"h/d1/c1/s1", "h/d1/c1/s2", "h/d2/c2/s1", "h/d3/c3/s1", "h/d1/c1/s1"}
+		chans        []chan api.SourceTuple
+	)
+	for i, topic := range sinkTopics {
+		CreatePub(topic)
+		var (
+			r   *regexp.Regexp
+			err error
+		)
+		if strings.ContainsAny(sourceTopics[i], "+#") {
+			r, err = getRegexp(sourceTopics[i])
+			if err != nil {
+				t.Error(err)
+				return
+			}
+		}
+		c := CreateSub(sourceTopics[i], r, fmt.Sprintf("%d", i), 100)
+		chans = append(chans, c)
+	}
+
+	expPub := map[string]*pubConsumers{
+		"h/d1/c1/s1": {
+			count: 2,
+			consumers: map[string]chan api.SourceTuple{
+				"1": chans[1],
+				"4": chans[4],
+			},
+		},
+		"h/d1/c1/s2": {
+			count: 1,
+			consumers: map[string]chan api.SourceTuple{
+				"0": chans[0],
+				"3": chans[3],
+			},
+		},
+		"h/d2/c2/s1": {
+			count: 1,
+			consumers: map[string]chan api.SourceTuple{
+				"1": chans[1],
+			},
+		},
+		"h/d3/c3/s1": {
+			count: 1,
+			consumers: map[string]chan api.SourceTuple{
+				"1": chans[1],
+				"2": chans[2],
+			},
+		},
+	}
+	if !reflect.DeepEqual(expPub, pubTopics) {
+		t.Errorf("Error adding: Expect\n\t%v\nbut got\n\t%v", render.AsCode(expPub), render.AsCode(pubTopics))
+		return
+	}
+	i := 0
+	for i < 3 {
+		CloseSourceConsumerChannel(sourceTopics[i], fmt.Sprintf("%d", i))
+		RemovePub(sinkTopics[i])
+		i++
+	}
+	expPub = map[string]*pubConsumers{
+		"h/d1/c1/s1": {
+			count: 1,
+			consumers: map[string]chan api.SourceTuple{
+				"4": chans[4],
+			},
+		},
+		"h/d1/c1/s2": {
+			count: 0,
+			consumers: map[string]chan api.SourceTuple{
+				"3": chans[3],
+			},
+		},
+		"h/d3/c3/s1": {
+			count:     1,
+			consumers: map[string]chan api.SourceTuple{},
+		},
+	}
+	if !reflect.DeepEqual(expPub, pubTopics) {
+		t.Errorf("Error closing: Expect\n\t%v\nbut got\n\t %v", render.AsCode(expPub), render.AsCode(pubTopics))
+	}
+}
+
+func getRegexp(topic string) (*regexp.Regexp, error) {
+	if len(topic) == 0 {
+		return nil, fmt.Errorf("invalid empty topic")
+	}
+
+	levels := strings.Split(topic, "/")
+	for i, level := range levels {
+		if level == "#" && i != len(levels)-1 {
+			return nil, fmt.Errorf("invalid topic %s: # must at the last level", topic)
+		}
+	}
+	regstr := strings.Replace(strings.ReplaceAll(topic, "+", "([^/]+)"), "#", ".", 1)
+	return regexp.Compile(regstr)
+}
diff --git a/internal/topo/memory/store/db_test.go b/internal/topo/memory/store/db_test.go
new file mode 100644
index 00000000..bd00e4c3
--- /dev/null
+++ b/internal/topo/memory/store/db_test.go
@@ -0,0 +1,114 @@
+// 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 store
+
+import (
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"reflect"
+	"testing"
+)
+
+func TestTable(t *testing.T) {
+	tb := createTable([]string{"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": 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)
+		return
+	}
+	v, _ = tb.Read([]string{"a"}, []interface{}{3})
+	exp = []api.SourceTuple{
+		api.NewDefaultSourceTuple(map[string]interface{}{"a": 3, "b": "0"}, nil),
+	}
+	if !reflect.DeepEqual(v, exp) {
+		t.Errorf("read 3 expect %v, but got %v", exp, v)
+		return
+	}
+	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 5, "b": "0"}, 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),
+		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
+	}
+	if !reflect.DeepEqual(v, exp) {
+		t.Errorf("read 1 again expect %v, but got %v", exp, v)
+		return
+	}
+	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
+	}
+}
+
+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")
+	if !ok {
+		t.Errorf("table t1 a should exist")
+		return
+	}
+	_, ok = db.getTable("t1")
+	if !ok {
+		t.Errorf("table t1 b should exist")
+		return
+	}
+	_, ok = db.getTable("t3")
+	if ok {
+		t.Errorf("table t1 c should not exist")
+		return
+	}
+	tc := db.tables["t1"]
+	if tc.count != 3 {
+		t.Errorf("table t1 a should have 2 instances but got %d", tc.count)
+		return
+	}
+	tc = db.tables["t2"]
+	if tc.count != 1 {
+		t.Errorf("table t1 a should have 1 instances")
+		return
+	}
+	db.dropTable("t1")
+	db.dropTable("t2")
+	_, ok = db.getTable("t2")
+	if ok {
+		t.Errorf("table ta 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)
+		return
+	}
+}
diff --git a/internal/topo/memory/store/store_test.go b/internal/topo/memory/store/store_test.go
new file mode 100644
index 00000000..a63a6ec9
--- /dev/null
+++ b/internal/topo/memory/store/store_test.go
@@ -0,0 +1,57 @@
+// 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 store
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestReg(t *testing.T) {
+	db = &database{
+		tables: make(map[string]*tableCount),
+	}
+	reg1, err := Reg("test", nil, []string{"a"})
+	if err != nil {
+		t.Errorf("register test error: %v", err)
+		return
+	}
+	_, err2 := Reg("test", nil, []string{"a", "b"})
+	if err2 != nil {
+		t.Errorf("register test error: %v", err2)
+		return
+	}
+	exp := map[string]*tableCount{
+		"test": {
+			count: 2,
+			t:     reg1,
+		},
+	}
+	if !reflect.DeepEqual(exp, db.tables) {
+		t.Errorf("register expect %v, but got %v", exp, db.tables)
+		return
+	}
+	Unreg("test")
+	exp = map[string]*tableCount{
+		"test": {
+			count: 1,
+			t:     reg1,
+		},
+	}
+	if !reflect.DeepEqual(exp, db.tables) {
+		t.Errorf("unregister expect %v, but got %v", exp, db.tables)
+		return
+	}
+}
diff --git a/internal/topo/node/lookup_node_test.go b/internal/topo/node/lookup_node_test.go
new file mode 100644
index 00000000..de7cd6dd
--- /dev/null
+++ b/internal/topo/node/lookup_node_test.go
@@ -0,0 +1,343 @@
+// 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 node
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/binder"
+	"github.com/lf-edge/ekuiper/internal/binder/io"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/internal/topo/lookup"
+	"github.com/lf-edge/ekuiper/internal/xsql"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"reflect"
+	"testing"
+	"time"
+)
+
+type mockLookupSrc struct {
+}
+
+func (m *mockLookupSrc) Open(_ api.StreamContext) error {
+	return nil
+}
+
+func (m *mockLookupSrc) Configure(_ string, _ map[string]interface{}) error {
+	return nil
+}
+
+// Lookup accept int value as the first array value
+func (m *mockLookupSrc) Lookup(_ api.StreamContext, _ []string, values []interface{}) ([]api.SourceTuple, error) {
+	a1, ok := values[0].(int)
+	if ok {
+		var result []api.SourceTuple
+		c := a1 % 2
+		if c != 0 {
+			result = append(result, api.NewDefaultSourceTuple(map[string]interface{}{
+				"newA": c,
+				"newB": c * 2,
+			}, nil))
+		}
+		c = a1 % 3
+		if c != 0 {
+			result = append(result, api.NewDefaultSourceTuple(map[string]interface{}{
+				"newA": c,
+				"newB": c * 2,
+			}, nil))
+		}
+		c = a1 % 5
+		if c != 0 {
+			result = append(result, api.NewDefaultSourceTuple(map[string]interface{}{
+				"newA": c,
+				"newB": c * 2,
+			}, nil))
+		}
+		c = a1 % 7
+		if c != 0 {
+			result = append(result, api.NewDefaultSourceTuple(map[string]interface{}{
+				"newA": c,
+				"newB": c * 2,
+			}, nil))
+		}
+		return result, nil
+	} else {
+		return []api.SourceTuple{
+			api.NewDefaultSourceTuple(map[string]interface{}{
+				"newA": 0,
+				"newB": 0,
+			}, nil),
+		}, nil
+	}
+}
+
+func (m *mockLookupSrc) Close(_ api.StreamContext) error {
+	// do nothing
+	return nil
+}
+
+type mockFac struct{}
+
+func (m *mockFac) Source(_ string) (api.Source, error) {
+	return nil, nil
+}
+
+func (m *mockFac) LookupSource(name string) (api.LookupSource, error) {
+	if name == "mock" {
+		return &mockLookupSrc{}, nil
+	}
+	return nil, nil
+}
+
+// init mock lookup source factory
+func init() {
+	io.Initialize([]binder.FactoryEntry{
+		{Name: "native plugin", Factory: &mockFac{}, Weight: 10},
+	})
+}
+
+func TestLookup(t *testing.T) {
+	var tests = []struct {
+		input  interface{} // a tuple or a window
+		output *xsql.JoinTuples
+	}{
+		{
+			input: &xsql.Tuple{
+				Emitter: "demo",
+				Message: map[string]interface{}{
+					"a": 6,
+					"b": "aaaa",
+				},
+			},
+			output: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.TupleRow{
+							&xsql.Tuple{
+								Emitter: "demo",
+								Message: map[string]interface{}{
+									"a": 6,
+									"b": "aaaa",
+								},
+							},
+							&xsql.Tuple{
+								Emitter: "mock",
+								Message: map[string]interface{}{
+									"newA": 1,
+									"newB": 2,
+								},
+							},
+						},
+					}, {
+						Tuples: []xsql.TupleRow{
+							&xsql.Tuple{
+								Emitter: "demo",
+								Message: map[string]interface{}{
+									"a": 6,
+									"b": "aaaa",
+								},
+							},
+							&xsql.Tuple{
+								Emitter: "mock",
+								Message: map[string]interface{}{
+									"newA": 6,
+									"newB": 12,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			input: &xsql.WindowTuples{
+				Content: []xsql.TupleRow{
+					&xsql.Tuple{
+						Emitter: "demo",
+						Message: map[string]interface{}{
+							"a": 9,
+							"b": "aaaa",
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "demo",
+						Message: map[string]interface{}{
+							"a": 4,
+							"b": "bbaa",
+						},
+					},
+				},
+			},
+			output: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.TupleRow{
+							&xsql.Tuple{
+								Emitter: "demo",
+								Message: map[string]interface{}{
+									"a": 9,
+									"b": "aaaa",
+								},
+							},
+							&xsql.Tuple{
+								Emitter: "mock",
+								Message: map[string]interface{}{
+									"newA": 1,
+									"newB": 2,
+								},
+							},
+						},
+					}, {
+						Tuples: []xsql.TupleRow{
+							&xsql.Tuple{
+								Emitter: "demo",
+								Message: map[string]interface{}{
+									"a": 9,
+									"b": "aaaa",
+								},
+							},
+							&xsql.Tuple{
+								Emitter: "mock",
+								Message: map[string]interface{}{
+									"newA": 4,
+									"newB": 8,
+								},
+							},
+						},
+					}, {
+						Tuples: []xsql.TupleRow{
+							&xsql.Tuple{
+								Emitter: "demo",
+								Message: map[string]interface{}{
+									"a": 9,
+									"b": "aaaa",
+								},
+							},
+							&xsql.Tuple{
+								Emitter: "mock",
+								Message: map[string]interface{}{
+									"newA": 2,
+									"newB": 4,
+								},
+							},
+						},
+					}, {
+						Tuples: []xsql.TupleRow{
+							&xsql.Tuple{
+								Emitter: "demo",
+								Message: map[string]interface{}{
+									"a": 4,
+									"b": "bbaa",
+								},
+							},
+							&xsql.Tuple{
+								Emitter: "mock",
+								Message: map[string]interface{}{
+									"newA": 1,
+									"newB": 2,
+								},
+							},
+						},
+					}, {
+						Tuples: []xsql.TupleRow{
+							&xsql.Tuple{
+								Emitter: "demo",
+								Message: map[string]interface{}{
+									"a": 4,
+									"b": "bbaa",
+								},
+							},
+							&xsql.Tuple{
+								Emitter: "mock",
+								Message: map[string]interface{}{
+									"newA": 4,
+									"newB": 8,
+								},
+							},
+						},
+					}, {
+						Tuples: []xsql.TupleRow{
+							&xsql.Tuple{
+								Emitter: "demo",
+								Message: map[string]interface{}{
+									"a": 4,
+									"b": "bbaa",
+								},
+							},
+							&xsql.Tuple{
+								Emitter: "mock",
+								Message: map[string]interface{}{
+									"newA": 4,
+									"newB": 8,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	options := &ast.Options{
+		DATASOURCE:        "mock",
+		TYPE:              "mock",
+		STRICT_VALIDATION: true,
+		KIND:              "lookup",
+	}
+	lookup.CreateInstance("mock", "mock", options)
+	contextLogger := conf.Log.WithField("rule", "TestLookup")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	l, _ := NewLookupNode("mock", []string{"a"}, ast.INNER_JOIN, []ast.Expr{&ast.FieldRef{
+		StreamName: "",
+		Name:       "a",
+	}}, options, &api.RuleOption{
+		IsEventTime:        false,
+		LateTol:            0,
+		Concurrency:        0,
+		BufferLength:       0,
+		SendMetaToSink:     false,
+		SendError:          false,
+		Qos:                0,
+		CheckpointInterval: 0,
+	})
+	errCh := make(chan error)
+	outputCh := make(chan interface{}, 1)
+	l.outputs["mock"] = outputCh
+	l.Exec(ctx, errCh)
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	for i, tt := range tests {
+		select {
+		case err := <-errCh:
+			t.Error(err)
+			return
+		case l.input <- tt.input:
+		case <-time.After(1 * time.Second):
+			t.Error("send message timeout")
+			return
+		}
+		select {
+		case err := <-errCh:
+			t.Error(err)
+			return
+		case output := <-outputCh:
+			if !reflect.DeepEqual(tt.output, output) {
+				t.Errorf("\ncase %d: expect %v but got %v", i, tt.output, output)
+			}
+		case <-time.After(1 * time.Second):
+			t.Error("send message timeout")
+			return
+		}
+	}
+}
diff --git a/internal/topo/node/source_node_test.go b/internal/topo/node/source_node_test.go
index 2322dda4..c3f09b35 100644
--- a/internal/topo/node/source_node_test.go
+++ b/internal/topo/node/source_node_test.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.
@@ -17,6 +17,7 @@ package node
 import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
+	nodeConf "github.com/lf-edge/ekuiper/internal/topo/node/conf"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"reflect"
@@ -44,7 +45,7 @@ func TestGetConf_Apply(t *testing.T) {
 	}, false)
 	contextLogger := conf.Log.WithField("rule", "test")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	conf := getSourceConf(ctx, n.sourceType, n.options)
+	conf := nodeConf.GetSourceConf(ctx, n.sourceType, n.options)
 	if !reflect.DeepEqual(result, conf) {
 		t.Errorf("result mismatch:\n\nexp=%s\n\ngot=%s\n\n", result, conf)
 	}
@@ -72,7 +73,7 @@ func TestGetConfAndConvert_Apply(t *testing.T) {
 	}, false)
 	contextLogger := conf.Log.WithField("rule", "test")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	conf := getSourceConf(ctx, n.sourceType, n.options)
+	conf := nodeConf.GetSourceConf(ctx, n.sourceType, n.options)
 	if !reflect.DeepEqual(result, conf) {
 		t.Errorf("result mismatch:\n\nexp=%s\n\ngot=%s\n\n", result, conf)
 		return
diff --git a/internal/topo/planner/lookupPlan_test.go b/internal/topo/planner/lookupPlan_test.go
new file mode 100644
index 00000000..c4b3b75b
--- /dev/null
+++ b/internal/topo/planner/lookupPlan_test.go
@@ -0,0 +1,319 @@
+// 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 planner
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"reflect"
+	"testing"
+)
+
+func TestValidate(t *testing.T) {
+	var tests = []struct {
+		p  *LookupPlan
+		v  bool
+		c  ast.Expr
+		k  []string
+		vv []ast.Expr
+	}{
+		{ // 0
+			p: &LookupPlan{
+				joinExpr: ast.Join{
+					Name:     "good",
+					JoinType: 0,
+					Expr: &ast.BinaryExpr{
+						OP: ast.EQ,
+						LHS: &ast.FieldRef{
+							StreamName: "left",
+							Name:       "device_id",
+						},
+						RHS: &ast.FieldRef{
+							StreamName: "good",
+							Name:       "id",
+						},
+					},
+				},
+			},
+			v: true,
+			k: []string{
+				"id",
+			},
+			vv: []ast.Expr{
+				&ast.FieldRef{
+					StreamName: "left",
+					Name:       "device_id",
+				},
+			},
+			c: nil,
+		}, { // 1
+			p: &LookupPlan{
+				joinExpr: ast.Join{
+					Name:     "good",
+					JoinType: 0,
+					Expr: &ast.BinaryExpr{
+						OP: ast.GT,
+						LHS: &ast.FieldRef{
+							StreamName: "left",
+							Name:       "device_id",
+						},
+						RHS: &ast.FieldRef{
+							StreamName: "good",
+							Name:       "id",
+						},
+					},
+				},
+			},
+			v: false,
+			c: nil,
+		}, { // 2
+			p: &LookupPlan{
+				joinExpr: ast.Join{
+					Name:     "good",
+					JoinType: 0,
+					Expr: &ast.BinaryExpr{
+						OP: ast.EQ,
+						LHS: &ast.FieldRef{
+							StreamName: "left",
+							Name:       "device_id",
+						},
+						RHS: &ast.IntegerLiteral{Val: 23},
+					},
+				},
+			},
+			v: false,
+			c: nil,
+		}, { // 3
+			p: &LookupPlan{
+				joinExpr: ast.Join{
+					Name:     "good",
+					JoinType: 0,
+					Expr: &ast.BinaryExpr{
+						OP: ast.OR,
+						LHS: &ast.BinaryExpr{
+							OP: ast.EQ,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.FieldRef{
+								StreamName: "good",
+								Name:       "id",
+							},
+						},
+						RHS: &ast.BinaryExpr{
+							OP: ast.EQ,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.FieldRef{
+								StreamName: "good",
+								Name:       "id1",
+							},
+						},
+					},
+				},
+			},
+			v: false,
+			c: nil,
+		}, { // 4
+			p: &LookupPlan{
+				joinExpr: ast.Join{
+					Name:     "good",
+					JoinType: 0,
+					Expr: &ast.BinaryExpr{
+						OP: ast.AND,
+						LHS: &ast.BinaryExpr{
+							OP: ast.EQ,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.FieldRef{
+								StreamName: "good",
+								Name:       "id",
+							},
+						},
+						RHS: &ast.BinaryExpr{
+							OP: ast.EQ,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.FieldRef{
+								StreamName: "good",
+								Name:       "id1",
+							},
+						},
+					},
+				},
+			},
+			v: true,
+			k: []string{
+				"id", "id1",
+			},
+			vv: []ast.Expr{
+				&ast.FieldRef{
+					StreamName: "left",
+					Name:       "device_id",
+				},
+				&ast.FieldRef{
+					StreamName: "left",
+					Name:       "device_id",
+				},
+			},
+			c: nil,
+		}, { // 5
+			p: &LookupPlan{
+				joinExpr: ast.Join{
+					Name:     "good",
+					JoinType: 0,
+					Expr: &ast.BinaryExpr{
+						OP: ast.AND,
+						LHS: &ast.BinaryExpr{
+							OP: ast.GT,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.IntegerLiteral{
+								Val: 33,
+							},
+						},
+						RHS: &ast.BinaryExpr{
+							OP: ast.EQ,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.FieldRef{
+								StreamName: "good",
+								Name:       "id1",
+							},
+						},
+					},
+				},
+			},
+			v: true,
+			k: []string{
+				"id1",
+			},
+			vv: []ast.Expr{
+				&ast.FieldRef{
+					StreamName: "left",
+					Name:       "device_id",
+				},
+			},
+			c: &ast.BinaryExpr{
+				OP: ast.GT,
+				LHS: &ast.FieldRef{
+					StreamName: "left",
+					Name:       "device_id",
+				},
+				RHS: &ast.IntegerLiteral{
+					Val: 33,
+				},
+			},
+		}, { // 6
+			p: &LookupPlan{
+				joinExpr: ast.Join{
+					Name:     "good",
+					JoinType: 0,
+					Expr: &ast.BinaryExpr{
+						OP: ast.AND,
+						LHS: &ast.BinaryExpr{
+							OP: ast.EQ,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.FieldRef{
+								StreamName: "good",
+								Name:       "id",
+							},
+						},
+						RHS: &ast.BinaryExpr{
+							OP: ast.EQ,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.FieldRef{
+								StreamName: "good",
+								Name:       "id",
+							},
+						},
+					},
+				},
+			},
+			v: false,
+		}, { // 7
+			p: &LookupPlan{
+				joinExpr: ast.Join{
+					Name:     "good",
+					JoinType: 0,
+					Expr: &ast.BinaryExpr{
+						OP: ast.AND,
+						LHS: &ast.BinaryExpr{
+							OP: ast.EQ,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.FieldRef{
+								StreamName: "right",
+								Name:       "id",
+							},
+						},
+						RHS: &ast.BinaryExpr{
+							OP: ast.EQ,
+							LHS: &ast.FieldRef{
+								StreamName: "left",
+								Name:       "device_id",
+							},
+							RHS: &ast.FieldRef{
+								StreamName: "right",
+								Name:       "id2",
+							},
+						},
+					},
+				},
+			},
+			v: false,
+		},
+	}
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	for i, tt := range tests {
+		rv := tt.p.validateAndExtractCondition()
+		if rv != tt.v {
+			t.Errorf("case %d: expect validate %v but got %v", i, tt.v, rv)
+			continue
+		}
+		if rv {
+			if !reflect.DeepEqual(tt.c, tt.p.conditions) {
+				t.Errorf("case %d: expect conditions %v but got %v", i, tt.c, tt.p.conditions)
+				continue
+			}
+			if !reflect.DeepEqual(tt.k, tt.p.keys) {
+				t.Errorf("case %d: expect keys %v but got %v", i, tt.k, tt.p.keys)
+				continue
+			}
+			if !reflect.DeepEqual(tt.vv, tt.p.valvars) {
+				t.Errorf("case %d: expect val vars %v but got %v", i, tt.vv, tt.p.valvars)
+			}
+		}
+	}
+}
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index 6f6387fb..a6616f93 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -2074,3 +2074,491 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 		}
 	}
 }
+
+func Test_createLogicalPlan4Lookup(t *testing.T) {
+	err, store := store.GetKV("stream")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	streamSqls := map[string]string{
+		"src1":   `CREATE STREAM src1 () WITH (DATASOURCE="src1", FORMAT="json", KEY="ts");`,
+		"table1": `CREATE TABLE table1 () WITH (DATASOURCE="table1",TYPE="sql", KIND="lookup");`,
+		"table2": `CREATE TABLE table2 () WITH (DATASOURCE="table2",TYPE="sql", KIND="lookup");`,
+	}
+	types := map[string]ast.StreamType{
+		"src1":   ast.TypeStream,
+		"table1": ast.TypeTable,
+		"table2": ast.TypeTable,
+	}
+	for name, sql := range streamSqls {
+		s, err := json.Marshal(&xsql.StreamInfo{
+			StreamType: types[name],
+			Statement:  sql,
+		})
+		if err != nil {
+			t.Error(err)
+			t.Fail()
+		}
+		err = store.Set(name, string(s))
+		if err != nil {
+			t.Error(err)
+			t.Fail()
+		}
+	}
+	streams := make(map[string]*ast.StreamStmt)
+	for n := range streamSqls {
+		streamStmt, err := xsql.GetDataSource(store, n)
+		if err != nil {
+			t.Errorf("fail to get stream %s, please check if stream is created", n)
+			return
+		}
+		streams[n] = streamStmt
+	}
+	var tests = []struct {
+		sql string
+		p   LogicalPlan
+		err string
+	}{
+		{ // 0
+			sql: `SELECT src1.a, table1.b FROM src1 INNER JOIN table1 ON src1.id = table1.id`,
+			p: ProjectPlan{
+				baseLogicalPlan: baseLogicalPlan{
+					children: []LogicalPlan{
+						LookupPlan{
+							baseLogicalPlan: baseLogicalPlan{
+								children: []LogicalPlan{
+									DataSourcePlan{
+										baseLogicalPlan: baseLogicalPlan{},
+										name:            "src1",
+										streamFields: []interface{}{
+											"a",
+										},
+										streamStmt: streams["src1"],
+										metaFields: []string{},
+									}.Init(),
+								},
+							},
+							joinExpr: ast.Join{
+								Name:     "table1",
+								Alias:    "",
+								JoinType: ast.INNER_JOIN,
+								Expr: &ast.BinaryExpr{
+									OP: ast.EQ,
+									LHS: &ast.FieldRef{
+										StreamName: "src1",
+										Name:       "id",
+									},
+									RHS: &ast.FieldRef{
+										StreamName: "table1",
+										Name:       "id",
+									},
+								},
+							},
+							keys: []string{"id"},
+							valvars: []ast.Expr{
+								&ast.FieldRef{
+									StreamName: "src1",
+									Name:       "id",
+								},
+							},
+							options: &ast.Options{
+								DATASOURCE:        "table1",
+								TYPE:              "sql",
+								STRICT_VALIDATION: true,
+								KIND:              "lookup",
+							},
+							conditions: nil,
+						}.Init(),
+					},
+				},
+				fields: []ast.Field{
+					{
+						Expr: &ast.FieldRef{
+							StreamName: "src1",
+							Name:       "a",
+						},
+						Name:  "a",
+						AName: "",
+					},
+					{
+						Expr: &ast.FieldRef{
+							StreamName: "table1",
+							Name:       "b",
+						},
+						Name:  "b",
+						AName: "",
+					},
+				},
+				isAggregate: false,
+				sendMeta:    false,
+			}.Init(),
+		},
+		{ // 1
+			sql: `SELECT src1.a, table1.b FROM src1 INNER JOIN table1 ON table1.b > 20 AND src1.c < 40 AND src1.id = table1.id`,
+			p: ProjectPlan{
+				baseLogicalPlan: baseLogicalPlan{
+					children: []LogicalPlan{
+						FilterPlan{
+							baseLogicalPlan: baseLogicalPlan{
+								children: []LogicalPlan{
+									LookupPlan{
+										baseLogicalPlan: baseLogicalPlan{
+											children: []LogicalPlan{
+												FilterPlan{
+													baseLogicalPlan: baseLogicalPlan{
+														children: []LogicalPlan{
+															DataSourcePlan{
+																baseLogicalPlan: baseLogicalPlan{},
+																name:            "src1",
+																streamFields: []interface{}{
+																	"a",
+																},
+																streamStmt: streams["src1"],
+																metaFields: []string{},
+															}.Init(),
+														},
+													},
+													condition: &ast.BinaryExpr{
+														OP: ast.LT,
+														LHS: &ast.FieldRef{
+															StreamName: "src1",
+															Name:       "c",
+														},
+														RHS: &ast.IntegerLiteral{Val: 40},
+													},
+												}.Init(),
+											},
+										},
+										joinExpr: ast.Join{
+											Name:     "table1",
+											Alias:    "",
+											JoinType: ast.INNER_JOIN,
+											Expr: &ast.BinaryExpr{
+												OP: ast.AND,
+												RHS: &ast.BinaryExpr{
+													OP: ast.EQ,
+													LHS: &ast.FieldRef{
+														StreamName: "src1",
+														Name:       "id",
+													},
+													RHS: &ast.FieldRef{
+														StreamName: "table1",
+														Name:       "id",
+													},
+												},
+												LHS: &ast.BinaryExpr{
+													OP: ast.AND,
+													LHS: &ast.BinaryExpr{
+														OP: ast.GT,
+														LHS: &ast.FieldRef{
+															StreamName: "table1",
+															Name:       "b",
+														},
+														RHS: &ast.IntegerLiteral{Val: 20},
+													},
+													RHS: &ast.BinaryExpr{
+														OP: ast.LT,
+														LHS: &ast.FieldRef{
+															StreamName: "src1",
+															Name:       "c",
+														},
+														RHS: &ast.IntegerLiteral{Val: 40},
+													},
+												},
+											},
+										},
+										keys: []string{"id"},
+										valvars: []ast.Expr{
+											&ast.FieldRef{
+												StreamName: "src1",
+												Name:       "id",
+											},
+										},
+										options: &ast.Options{
+											DATASOURCE:        "table1",
+											TYPE:              "sql",
+											STRICT_VALIDATION: true,
+											KIND:              "lookup",
+										},
+										conditions: &ast.BinaryExpr{
+											OP: ast.AND,
+											LHS: &ast.BinaryExpr{
+												OP: ast.GT,
+												LHS: &ast.FieldRef{
+													StreamName: "table1",
+													Name:       "b",
+												},
+												RHS: &ast.IntegerLiteral{Val: 20},
+											},
+											RHS: &ast.BinaryExpr{
+												OP: ast.LT,
+												LHS: &ast.FieldRef{
+													StreamName: "src1",
+													Name:       "c",
+												},
+												RHS: &ast.IntegerLiteral{Val: 40},
+											},
+										},
+									}.Init(),
+								},
+							},
+							condition: &ast.BinaryExpr{
+								OP: ast.GT,
+								LHS: &ast.FieldRef{
+									StreamName: "table1",
+									Name:       "b",
+								},
+								RHS: &ast.IntegerLiteral{Val: 20},
+							},
+						}.Init(),
+					},
+				},
+				fields: []ast.Field{
+					{
+						Expr: &ast.FieldRef{
+							StreamName: "src1",
+							Name:       "a",
+						},
+						Name:  "a",
+						AName: "",
+					},
+					{
+						Expr: &ast.FieldRef{
+							StreamName: "table1",
+							Name:       "b",
+						},
+						Name:  "b",
+						AName: "",
+					},
+				},
+				isAggregate: false,
+				sendMeta:    false,
+			}.Init(),
+		},
+		{ // 0
+			sql: `SELECT src1.a, table1.b, table2.c FROM src1 INNER JOIN table1 ON src1.id = table1.id INNER JOIN table2 on table1.id = table2.id`,
+			p: ProjectPlan{
+				baseLogicalPlan: baseLogicalPlan{
+					children: []LogicalPlan{
+						LookupPlan{
+							baseLogicalPlan: baseLogicalPlan{
+								children: []LogicalPlan{
+									LookupPlan{
+										baseLogicalPlan: baseLogicalPlan{
+											children: []LogicalPlan{
+												DataSourcePlan{
+													baseLogicalPlan: baseLogicalPlan{},
+													name:            "src1",
+													streamFields: []interface{}{
+														"a",
+													},
+													streamStmt: streams["src1"],
+													metaFields: []string{},
+												}.Init(),
+											},
+										},
+										joinExpr: ast.Join{
+											Name:     "table1",
+											Alias:    "",
+											JoinType: ast.INNER_JOIN,
+											Expr: &ast.BinaryExpr{
+												OP: ast.EQ,
+												LHS: &ast.FieldRef{
+													StreamName: "src1",
+													Name:       "id",
+												},
+												RHS: &ast.FieldRef{
+													StreamName: "table1",
+													Name:       "id",
+												},
+											},
+										},
+										keys: []string{"id"},
+										valvars: []ast.Expr{
+											&ast.FieldRef{
+												StreamName: "src1",
+												Name:       "id",
+											},
+										},
+										options: &ast.Options{
+											DATASOURCE:        "table1",
+											TYPE:              "sql",
+											STRICT_VALIDATION: true,
+											KIND:              "lookup",
+										},
+										conditions: nil,
+									}.Init(),
+								},
+							},
+							joinExpr: ast.Join{
+								Name:     "table2",
+								Alias:    "",
+								JoinType: ast.INNER_JOIN,
+								Expr: &ast.BinaryExpr{
+									OP: ast.EQ,
+									LHS: &ast.FieldRef{
+										StreamName: "table1",
+										Name:       "id",
+									},
+									RHS: &ast.FieldRef{
+										StreamName: "table2",
+										Name:       "id",
+									},
+								},
+							},
+							keys: []string{"id"},
+							valvars: []ast.Expr{
+								&ast.FieldRef{
+									StreamName: "table1",
+									Name:       "id",
+								},
+							},
+							options: &ast.Options{
+								DATASOURCE:        "table2",
+								TYPE:              "sql",
+								STRICT_VALIDATION: true,
+								KIND:              "lookup",
+							},
+						}.Init(),
+					},
+				},
+				fields: []ast.Field{
+					{
+						Expr: &ast.FieldRef{
+							StreamName: "src1",
+							Name:       "a",
+						},
+						Name:  "a",
+						AName: "",
+					},
+					{
+						Expr: &ast.FieldRef{
+							StreamName: "table1",
+							Name:       "b",
+						},
+						Name:  "b",
+						AName: "",
+					},
+					{
+						Expr: &ast.FieldRef{
+							StreamName: "table2",
+							Name:       "c",
+						},
+						Name:  "c",
+						AName: "",
+					},
+				},
+				isAggregate: false,
+				sendMeta:    false,
+			}.Init(),
+		},
+		{ // 3
+			sql: `SELECT src1.a, table1.b FROM src1 INNER JOIN table1 ON src1.id = table1.id GROUP BY TUMBLINGWINDOW(ss, 10)`,
+			p: ProjectPlan{
+				baseLogicalPlan: baseLogicalPlan{
+					children: []LogicalPlan{
+						LookupPlan{
+							baseLogicalPlan: baseLogicalPlan{
+								children: []LogicalPlan{
+									WindowPlan{
+										baseLogicalPlan: baseLogicalPlan{
+											children: []LogicalPlan{
+												DataSourcePlan{
+													baseLogicalPlan: baseLogicalPlan{},
+													name:            "src1",
+													streamFields: []interface{}{
+														"a",
+													},
+													streamStmt: streams["src1"],
+													metaFields: []string{},
+												}.Init(),
+											},
+										},
+										condition: nil,
+										wtype:     ast.TUMBLING_WINDOW,
+										length:    10000,
+										interval:  0,
+										limit:     0,
+									}.Init(),
+								},
+							},
+							joinExpr: ast.Join{
+								Name:     "table1",
+								Alias:    "",
+								JoinType: ast.INNER_JOIN,
+								Expr: &ast.BinaryExpr{
+									OP: ast.EQ,
+									LHS: &ast.FieldRef{
+										StreamName: "src1",
+										Name:       "id",
+									},
+									RHS: &ast.FieldRef{
+										StreamName: "table1",
+										Name:       "id",
+									},
+								},
+							},
+							keys: []string{"id"},
+							valvars: []ast.Expr{
+								&ast.FieldRef{
+									StreamName: "src1",
+									Name:       "id",
+								},
+							},
+							options: &ast.Options{
+								DATASOURCE:        "table1",
+								TYPE:              "sql",
+								STRICT_VALIDATION: true,
+								KIND:              "lookup",
+							},
+							conditions: nil,
+						}.Init(),
+					},
+				},
+				fields: []ast.Field{
+					{
+						Expr: &ast.FieldRef{
+							StreamName: "src1",
+							Name:       "a",
+						},
+						Name:  "a",
+						AName: "",
+					},
+					{
+						Expr: &ast.FieldRef{
+							StreamName: "table1",
+							Name:       "b",
+						},
+						Name:  "b",
+						AName: "",
+					},
+				},
+				isAggregate: false,
+				sendMeta:    false,
+			}.Init(),
+		},
+	}
+	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, &api.RuleOption{
+			IsEventTime:        false,
+			LateTol:            0,
+			Concurrency:        0,
+			BufferLength:       0,
+			SendMetaToSink:     false,
+			Qos:                0,
+			CheckpointInterval: 0,
+			SendError:          true,
+		}, store)
+		if !reflect.DeepEqual(tt.err, testx.Errstring(err)) {
+			t.Errorf("%d. %q: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.sql, tt.err, err)
+		} else if !reflect.DeepEqual(tt.p, p) {
+			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))
+		}
+	}
+}
diff --git a/internal/xsql/parser_tree_test.go b/internal/xsql/parser_tree_test.go
index 663c9d47..937f2e28 100644
--- a/internal/xsql/parser_tree_test.go
+++ b/internal/xsql/parser_tree_test.go
@@ -107,6 +107,28 @@ func TestParser_ParseTree(t *testing.T) {
 				StreamType: ast.TypeTable,
 			},
 		},
+		{
+			s: `CREATE TABLE table1 (
+					name STRING,
+					size BIGINT,
+					id BIGINT
+				) WITH (DATASOURCE="devices", KIND="LOOKUP", TYPE="sql");`,
+			stmt: &ast.StreamStmt{
+				Name: ast.StreamName("table1"),
+				StreamFields: []ast.StreamField{
+					{Name: "name", FieldType: &ast.BasicType{Type: ast.STRINGS}},
+					{Name: "size", FieldType: &ast.BasicType{Type: ast.BIGINT}},
+					{Name: "id", FieldType: &ast.BasicType{Type: ast.BIGINT}},
+				},
+				Options: &ast.Options{
+					DATASOURCE:        "devices",
+					STRICT_VALIDATION: true,
+					KIND:              ast.StreamKindLookup,
+					TYPE:              "sql",
+				},
+				StreamType: ast.TypeTable,
+			},
+		},
 		{
 			s:    `SHOW STREAMS`,
 			stmt: &ast.ShowStreamsStatement{},