feat(lookup): lookup table runtime

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

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=

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 {

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()
+}

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,7 @@ var (
 		"memory":      func() api.Sink { return memory.GetSink() },
 		"neuron":      func() api.Sink { return neuron.GetSink() },
 	}
+	lookupSources = map[string]NewLookupSourceFunc{}
 )
 
 type Manager struct{}
@@ -54,6 +56,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

internal/topo/node/lookup_node.go

@@ -0,0 +1,246 @@
+// 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/binder/io"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"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 {
+			props := getSourceConf(ctx, n.sourceType, n.srcOptions)
+			ctx.GetLogger().Infof("open lookup source node with props %v", conf.Printable(props))
+			// Create the lookup source according to the source options
+			ns, err := io.LookupSource(n.sourceType)
+			if err != nil {
+				return err
+			}
+			err = ns.Configure(n.srcOptions.DATASOURCE, props, n.Keys)
+			if err != nil {
+				return err
+			}
+			err = ns.Open(ctx)
+			if err != nil {
+				return err
+			}
+			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))
+	for i, val := range n.vals {
+		cvs[i] = ve.Eval(val)
+	}
+	r, e := ns.Lookup(ctx, 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,
+				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)))
+}

internal/topo/node/lookup_node_test.go

@@ -0,0 +1,340 @@
+// 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/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{}, _ []string) error {
+	return nil
+}
+
+// Lookup accept int value as the first array value
+func (m *mockLookupSrc) Lookup(_ api.StreamContext, values []interface{}) ([]map[string]interface{}, error) {
+	a1, ok := values[0].(int)
+	if ok {
+		var result []map[string]interface{}
+		c := a1 % 2
+		if c != 0 {
+			result = append(result, map[string]interface{}{
+				"newA": c,
+				"newB": c * 2,
+			})
+		}
+		c = a1 % 3
+		if c != 0 {
+			result = append(result, map[string]interface{}{
+				"newA": c,
+				"newB": c * 2,
+			})
+		}
+		c = a1 % 5
+		if c != 0 {
+			result = append(result, map[string]interface{}{
+				"newA": c,
+				"newB": c * 2,
+			})
+		}
+		c = a1 % 7
+		if c != 0 {
+			result = append(result, map[string]interface{}{
+				"newA": c,
+				"newB": c * 2,
+			})
+		}
+		return result, nil
+	} else {
+		return []map[string]interface{}{
+			{
+				"newA": 0,
+				"newB": 0,
+			},
+		}, nil
+	}
+}
+
+func (m *mockLookupSrc) Close(ctx 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,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	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",
+	}}, &ast.Options{
+		DATASOURCE:        "mock",
+		TYPE:              "mock",
+		STRICT_VALIDATION: true,
+		KIND:              "lookup",
+	}, &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
+		}
+	}
+}

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 {

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}
+}

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)
+			}
+		}
+	}
+}

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{

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))
+		}
+	}
+}

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":

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() {

internal/xsql/parser_tree_test.go

@@ -108,6 +108,28 @@ func TestParser_ParseTree(t *testing.T) {
 			},
 		},
 		{
+			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{},
 		},

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)

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
+	Configure(datasource string, props map[string]interface{}, lookupKeys []string) error
+	// Lookup receive lookup values to construct the query and return query results
+	Lookup(ctx StreamContext, values []interface{}) ([]map[string]interface{}, error)
+	Closable
 }
 
 type Sink interface {

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() {}

pkg/ast/token.go

@@ -136,6 +136,7 @@ const (
 	RETAIN_SIZE
 	SHARED
 	SCHEMAID
+	KIND
 
 	DD
 	HH