Repository: lf-edge/ekuiper
Pull request #2922 (2024-06-13 06:49:13)

feat: revise connection interface

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `74.50980%` with `13 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@1c0784f`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/io/connection/connection\_pool.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?src=pr&el=tree&filepath=internal%2Fio%2Fconnection%2Fconnection_pool.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vY29ubmVjdGlvbi9jb25uZWN0aW9uX3Bvb2wuZ28=) | 80.00% | [4 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/mqtt/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?src=pr&el=tree&filepath=internal%2Fio%2Fmqtt%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vbXF0dC9zaW5rLmdv) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/mqtt/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?src=pr&el=tree&filepath=internal%2Fio%2Fmqtt%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vbXF0dC9zb3VyY2UuZ28=) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/mqtt/client/client.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?src=pr&el=tree&filepath=internal%2Fio%2Fmqtt%2Fclient%2Fclient.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vbXF0dC9jbGllbnQvY2xpZW50Lmdv) | 75.00% | [1 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #2922   +/-   ##
============================================
  Coverage             ?   68.74%           
============================================
  Files                ?      297           
  Lines                ?    28702           
  Branches             ?        0           
============================================
  Hits                 ?    19729           
  Misses               ?     7256           
  Partials             ?     1717           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2922?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/contract/api/source.go b/contract/api/source.go
index 9a2c1e04..7095294d 100644
--- a/contract/api/source.go
+++ b/contract/api/source.go
@@ -66,13 +66,20 @@ type Rewindable interface {
 	ResetOffset(input map[string]any) error
 }
 
-// LookupSource is a source feature to query the source on demand (TO be modified later)
+// LookupSource is a source feature to query the source on demand
 type LookupSource interface {
 	Source
 	// Lookup receive lookup values to construct the query and return query results
 	Lookup(ctx StreamContext, fields []string, keys []string, values []any) ([]map[string]any, error)
 }
 
+// LookupBytesSource looks up with the bytes payload pushed by the external source
+type LookupBytesSource interface {
+	Source
+	// Lookup receive multiple rows of bytes
+	Lookup(ctx StreamContext, fields []string, keys []string, values []any) ([][]byte, error)
+}
+
 /// helper function definition
 
 type ErrorIngest func(ctx StreamContext, err error)
diff --git a/internal/binder/factory.go b/internal/binder/factory.go
index 486a7688..dacea399 100644
--- a/internal/binder/factory.go
+++ b/internal/binder/factory.go
@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -20,7 +20,7 @@ import (
 
 type SourceFactory interface {
 	Source(name string) (api.Source, error)
-	LookupSource(name string) (api.LookupSource, error)
+	LookupSource(name string) (api.Source, error)
 }
 
 type SinkFactory interface {
diff --git a/internal/binder/io/binder.go b/internal/binder/io/binder.go
index 4faf6c4c..671ca392 100644
--- a/internal/binder/io/binder.go
+++ b/internal/binder/io/binder.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -20,6 +20,7 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/binder"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
 var ( // init once and read only
@@ -63,6 +64,9 @@ func Source(name string) (api.Source, error) {
 			errs = errors.Join(errs, fmt.Errorf("%s:%v", sourceFactoriesNames[i], err))
 		}
 		if r != nil {
+			if !modules.IsStreamSource(r) {
+				return nil, fmt.Errorf("got non stream source %s", name)
+			}
 			return r, errs
 		}
 	}
@@ -83,7 +87,7 @@ func Sink(name string) (api.Sink, error) {
 	return nil, errs
 }
 
-func LookupSource(name string) (api.LookupSource, error) {
+func LookupSource(name string) (api.Source, error) {
 	var errs error
 	for i, sf := range sourceFactories {
 		r, err := sf.LookupSource(name)
@@ -91,6 +95,9 @@ func LookupSource(name string) (api.LookupSource, error) {
 			errs = errors.Join(errs, fmt.Errorf("%s:%v", sourceFactoriesNames[i], err))
 		}
 		if r != nil {
+			if !modules.IsLookupSource(r) {
+				return nil, fmt.Errorf("got non lookup source %s", name)
+			}
 			return r, errs
 		}
 	}
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index 7bb314ef..42987322 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -59,7 +59,7 @@ func (m *Manager) Source(name string) (api.Source, error) {
 	return nil, nil
 }
 
-func (m *Manager) LookupSource(name string) (api.LookupSource, error) {
+func (m *Manager) LookupSource(name string) (api.Source, error) {
 	if s, ok := modules.LookupSources[name]; ok {
 		return s(), nil
 	}
diff --git a/internal/binder/mock/mock_factory.go b/internal/binder/mock/mock_factory.go
index d21fab0e..b66dc005 100644
--- a/internal/binder/mock/mock_factory.go
+++ b/internal/binder/mock/mock_factory.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -35,7 +35,7 @@ func (f *MockFactory) Source(name string) (api.Source, error) {
 	}
 }
 
-func (f *MockFactory) LookupSource(name string) (api.LookupSource, error) {
+func (f *MockFactory) LookupSource(name string) (api.Source, error) {
 	return nil, nil
 }
 
@@ -95,6 +95,10 @@ func (m *mockSource) Close(_ api.StreamContext) error {
 	return nil
 }
 
+func (m *mockSource) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest) error {
+	return nil
+}
+
 type mockSink struct{}
 
 func (m *mockSink) Provision(ctx api.StreamContext, configs map[string]any) error {
diff --git a/internal/io/memory/lookupsource.go b/internal/io/memory/lookupsource.go
index 2bc080d3..d205a1bd 100644
--- a/internal/io/memory/lookupsource.go
+++ b/internal/io/memory/lookupsource.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -87,6 +87,6 @@ func (s *lookupsource) Close(ctx api.StreamContext) error {
 	return store.Unreg(s.topic, s.key)
 }
 
-func GetLookupSource() api.LookupSource {
+func GetLookupSource() api.Source {
 	return &lookupsource{}
 }
diff --git a/internal/io/mqtt/client/client.go b/internal/io/mqtt/client/client.go
index fc006da1..460ee433 100644
--- a/internal/io/mqtt/client/client.go
+++ b/internal/io/mqtt/client/client.go
@@ -25,9 +25,9 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cert"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
@@ -37,7 +37,6 @@ type Connection struct {
 	selId     string
 	logger    api.Logger
 	connected atomic.Bool
-	refCount  atomic.Int32
 	// key is the topic. Each topic will have only one connector
 	subscriptions map[string]*SubscriptionInfo
 }
@@ -90,18 +89,8 @@ func (conn *Connection) onReconnecting(_ pahoMqtt.Client, _ *pahoMqtt.ClientOpti
 	conn.logger.Infof("Reconnecting to mqtt broker")
 }
 
-// Do not call this directly. Call connection pool Attach method to get the connection
-func (conn *Connection) Attach(ctx api.StreamContext) {
-	conn.refCount.Add(1)
-}
-
-func (conn *Connection) Ref(ctx api.StreamContext) int {
-	return int(conn.refCount.Load())
-}
-
 // Do not call this directly. Call connection pool Detach method to release the connection
 func (conn *Connection) DetachSub(ctx api.StreamContext, props map[string]any) {
-	conn.refCount.Add(-1)
 	topic, err := getTopicFromProps(props)
 	if err != nil {
 		return
@@ -110,10 +99,6 @@ func (conn *Connection) DetachSub(ctx api.StreamContext, props map[string]any) {
 	conn.Client.Unsubscribe(topic)
 }
 
-func (conn *Connection) DetachPub(ctx api.StreamContext, props map[string]any) {
-	conn.refCount.Add(-1)
-}
-
 func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error {
 	conn.subscriptions[topic] = info
 	token := conn.Client.Subscribe(topic, info.Qos, info.Handler)
@@ -132,12 +117,12 @@ func (conn *Connection) Ping(ctx api.StreamContext) error {
 	}
 }
 
-func CreateConnection(ctx api.StreamContext, selId string, props map[string]any) (modules.Connection, error) {
-	return CreateClient(ctx, selId, props)
+func CreateConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
+	return CreateClient(ctx, props)
 }
 
 // CreateClient creates a new mqtt client. It is anonymous and does not require a name.
-func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*Connection, error) {
+func CreateClient(ctx api.StreamContext, props map[string]any) (*Connection, error) {
 	c, err := ValidateConfig(props)
 	if err != nil {
 		return nil, err
@@ -175,9 +160,6 @@ func CreateClient(ctx api.StreamContext, selId string, props map[string]any) (*C
 	}
 	ctx.GetLogger().Infof("new mqtt client created")
 	con.Client = cli
-	if len(selId) > 0 {
-		con.Attach(ctx)
-	}
 	return con, nil
 }
 
@@ -222,7 +204,7 @@ func ValidateConfig(props map[string]any) (*ConnectionConfig, error) {
 }
 
 func CreateAnonymousConnection(ctx api.StreamContext, props map[string]any) (*Connection, error) {
-	cli, err := CreateClient(ctx, "", props)
+	cli, err := CreateClient(ctx, props)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/io/mqtt/sink.go b/internal/io/mqtt/sink.go
index 0384464a..2ef6fa37 100644
--- a/internal/io/mqtt/sink.go
+++ b/internal/io/mqtt/sink.go
@@ -19,9 +19,9 @@ import (
 	"strings"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 )
 
 // AdConf is the advanced configuration for the mqtt sink
@@ -64,27 +64,17 @@ func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error {
 
 func (ms *Sink) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Connecting to mqtt server")
-	var cli *client.Connection
 	var err error
-	if len(ms.adconf.SelId) > 0 {
-		conn, err := connection.GetNameConnection(ms.adconf.SelId)
-		if err != nil {
-			return err
-		}
-		c, ok := conn.(*client.Connection)
-		if !ok {
-			return fmt.Errorf("connection %s should be mqtt connection", ms.adconf.SelId)
-		}
-		cli = c
-	} else {
-		id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
-		conn, err := connection.CreateNonStoredConnection(ctx, id, "mqtt", ms.config)
-		if err != nil {
-			return err
-		}
-		cli = conn.(*client.Connection)
+	id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
+	conn, err := connection.FetchConnection(ctx, id, "mqtt", ms.config)
+	if err != nil {
+		return err
+	}
+	c, ok := conn.(*client.Connection)
+	if !ok {
+		return fmt.Errorf("connection %s should be mqtt connection", ms.adconf.SelId)
 	}
-	ms.cli = cli
+	ms.cli = c
 	return err
 }
 
@@ -111,12 +101,8 @@ func (ms *Sink) Collect(ctx api.StreamContext, item api.RawTuple) error {
 func (ms *Sink) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Info("Closing mqtt sink connector")
 	if ms.cli != nil {
-		if len(ms.adconf.SelId) < 1 {
-			id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
-			connection.DropNonStoredConnection(ctx, id)
-		} else {
-			ms.cli.DetachPub(ctx, nil)
-		}
+		id := fmt.Sprintf("%s-%s-%s-mqtt-sink", ctx.GetRuleId(), ctx.GetOpId(), ms.adconf.Tpc)
+		connection.DetachConnection(ctx, id, ms.config)
 	}
 	return nil
 }
diff --git a/internal/io/mqtt/source.go b/internal/io/mqtt/source.go
index c446a7ce..18317c41 100644
--- a/internal/io/mqtt/source.go
+++ b/internal/io/mqtt/source.go
@@ -20,10 +20,10 @@ import (
 	pahoMqtt "github.com/eclipse/paho.mqtt.golang"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
@@ -75,24 +75,12 @@ func (ms *SourceConnector) Connect(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Connecting to mqtt server")
 	var cli *client.Connection
 	var err error
-	if len(ms.cfg.SelId) > 0 {
-		conn, err := connection.GetNameConnection(ms.cfg.SelId)
-		if err != nil {
-			return err
-		}
-		c, ok := conn.(*client.Connection)
-		if !ok {
-			return fmt.Errorf("connection %s should be mqtt connection", ms.cfg.SelId)
-		}
-		cli = c
-	} else {
-		id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
-		conn, err := connection.CreateNonStoredConnection(ctx, id, "mqtt", ms.props)
-		if err != nil {
-			return err
-		}
-		cli = conn.(*client.Connection)
+	id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
+	conn, err := connection.FetchConnection(ctx, id, "mqtt", ms.props)
+	if err != nil {
+		return err
 	}
+	cli = conn.(*client.Connection)
 	ms.cli = cli
 	return err
 }
@@ -126,12 +114,9 @@ func (ms *SourceConnector) onMessage(ctx api.StreamContext, msg pahoMqtt.Message
 func (ms *SourceConnector) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Closing mqtt source connector to topic %s.", ms.tpc)
 	if ms.cli != nil {
-		if len(ms.cfg.SelId) < 1 {
-			id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
-			connection.DropNonStoredConnection(ctx, id)
-		} else {
-			ms.cli.DetachSub(ctx, ms.props)
-		}
+		id := fmt.Sprintf("%s-%s-%s-mqtt-source", ctx.GetRuleId(), ctx.GetOpId(), ms.tpc)
+		connection.DetachConnection(ctx, id, ms.props)
+		ms.cli.DetachSub(ctx, ms.props)
 	}
 	return nil
 }
diff --git a/internal/server/connection.go b/internal/server/connection.go
index 841c2c1c..5157befe 100644
--- a/internal/server/connection.go
+++ b/internal/server/connection.go
@@ -21,8 +21,8 @@ import (
 
 	"github.com/gorilla/mux"
 
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 )
 
 type ConnectionRequest struct {
diff --git a/internal/server/server.go b/internal/server/server.go
index b3930489..ec648f82 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -35,7 +35,6 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/binder/io"
 	"github.com/lf-edge/ekuiper/v2/internal/binder/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/keyedstate"
 	meta2 "github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
@@ -48,6 +47,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
diff --git a/internal/topo/lookup/table.go b/internal/topo/lookup/table.go
index e2c12abf..57d39b39 100644
--- a/internal/topo/lookup/table.go
+++ b/internal/topo/lookup/table.go
@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -31,7 +31,7 @@ import (
 // It will only stop once the table is dropped.
 
 type info struct {
-	ls    api.LookupSource
+	ls    api.Source
 	count int32
 }
 
@@ -41,7 +41,7 @@ var (
 )
 
 // Attach called by lookup nodes. Add a count to the info
-func Attach(name string) (api.LookupSource, error) {
+func Attach(name string) (api.Source, error) {
 	lock.Lock()
 	defer lock.Unlock()
 	if i, ok := instances[name]; ok {
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index 75650508..3e94f7ca 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -52,38 +52,6 @@ type dconf struct {
 	PayloadDelimiter  string        `json:"payloadDelimiter"`
 }
 
-func (o *DecodeOp) AttachSchema(ctx api.StreamContext, dataSource string, schema map[string]*ast.JsonStreamField, isWildcard bool) {
-	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
-		ctx.GetLogger().Infof("attach schema to shared stream")
-		// append payload field to schema
-		if o.additionSchema != "" {
-			newSchema := make(map[string]*ast.JsonStreamField, len(schema)+1)
-			for k, v := range schema {
-				newSchema[k] = v
-			}
-			newSchema[o.additionSchema] = nil
-		}
-		if err := o.sLayer.MergeSchema(ctx.GetRuleId(), dataSource, schema, isWildcard); err != nil {
-			ctx.GetLogger().Warnf("merge schema to shared stream failed, err: %v", err)
-		} else {
-			ctx.GetLogger().Infof("attach schema become %+v", o.sLayer.GetSchema())
-			fastDecoder.ResetSchema(o.sLayer.GetSchema())
-		}
-	}
-}
-
-func (o *DecodeOp) DetachSchema(ctx api.StreamContext, ruleId string) {
-	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
-		ctx.GetLogger().Infof("detach schema for shared stream rule %v", ruleId)
-		if err := o.sLayer.DetachSchema(ruleId); err != nil {
-			ctx.GetLogger().Infof("detach schema for shared stream rule %v failed, err:%v", ruleId, err)
-		} else {
-			fastDecoder.ResetSchema(o.sLayer.GetSchema())
-			ctx.GetLogger().Infof("detach schema become %+v", o.sLayer.GetSchema())
-		}
-	}
-}
-
 func NewDecodeOp(ctx api.StreamContext, forPayload bool, name, StreamName string, ruleId string, rOpt *def.RuleOption, options *ast.Options, isWildcard, isSchemaless bool, schema map[string]*ast.JsonStreamField, props map[string]any) (*DecodeOp, error) {
 	dc := &dconf{}
 	e := cast.MapToStruct(props, dc)
@@ -200,6 +168,38 @@ func (o *DecodeOp) Worker(ctx api.StreamContext, item any) []any {
 	}
 }
 
+func (o *DecodeOp) AttachSchema(ctx api.StreamContext, dataSource string, schema map[string]*ast.JsonStreamField, isWildcard bool) {
+	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
+		ctx.GetLogger().Infof("attach schema to shared stream")
+		// append payload field to schema
+		if o.additionSchema != "" {
+			newSchema := make(map[string]*ast.JsonStreamField, len(schema)+1)
+			for k, v := range schema {
+				newSchema[k] = v
+			}
+			newSchema[o.additionSchema] = nil
+		}
+		if err := o.sLayer.MergeSchema(ctx.GetRuleId(), dataSource, schema, isWildcard); err != nil {
+			ctx.GetLogger().Warnf("merge schema to shared stream failed, err: %v", err)
+		} else {
+			ctx.GetLogger().Infof("attach schema become %+v", o.sLayer.GetSchema())
+			fastDecoder.ResetSchema(o.sLayer.GetSchema())
+		}
+	}
+}
+
+func (o *DecodeOp) DetachSchema(ctx api.StreamContext, ruleId string) {
+	if fastDecoder, ok := o.converter.(message.SchemaResetAbleConverter); ok {
+		ctx.GetLogger().Infof("detach schema for shared stream rule %v", ruleId)
+		if err := o.sLayer.DetachSchema(ruleId); err != nil {
+			ctx.GetLogger().Infof("detach schema for shared stream rule %v failed, err:%v", ruleId, err)
+		} else {
+			fastDecoder.ResetSchema(o.sLayer.GetSchema())
+			ctx.GetLogger().Infof("detach schema become %+v", o.sLayer.GetSchema())
+		}
+	}
+}
+
 // PayloadDecodeWorker each input has one message with the payload field to decode
 //
 //	{
@@ -320,6 +320,10 @@ func (o *DecodeOp) PayloadBatchDecodeWorker(ctx api.StreamContext, item any) []a
 				vv = vt
 			case map[string]any:
 				vv = xsql.Message(vt)
+			default:
+				return []any{
+					fmt.Errorf("unsupported payload received, must be a slice of maps: %v", batchVal),
+				}
 			}
 			payload, ok := vv.Value(o.c.PayloadField, "")
 			if !ok {
diff --git a/internal/topo/node/lookup_node.go b/internal/topo/node/lookup_node.go
index a12759fa..e7950003 100644
--- a/internal/topo/node/lookup_node.go
+++ b/internal/topo/node/lookup_node.go
@@ -19,14 +19,15 @@ import (
 	"time"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/converter"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/lookup"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/lookup/cache"
-	nodeConf "github.com/lf-edge/ekuiper/v2/internal/topo/node/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/infra"
+	"github.com/lf-edge/ekuiper/v2/pkg/message"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
@@ -39,22 +40,19 @@ type LookupConf struct {
 // LookupNode will look up the data from the external source when receiving an event
 type LookupNode struct {
 	*defaultSinkNode
-	sourceType string
-	joinType   ast.JoinType
-	vals       []ast.Expr
 
-	srcOptions *ast.Options
-	conf       *LookupConf
-	fields     []string
-	keys       []string
+	conf *LookupConf
+
+	joinType ast.JoinType
+	vals     []ast.Expr
+	fields   []string
+	keys     []string
+	// If lookupByteSource, the decoders are needed
+	isBytesLookup bool
+	formatDecoder message.Converter
 }
 
-func NewLookupNode(name string, fields []string, keys []string, joinType ast.JoinType, vals []ast.Expr, srcOptions *ast.Options, options *def.RuleOption) (*LookupNode, error) {
-	t := srcOptions.TYPE
-	if t == "" {
-		return nil, fmt.Errorf("source type is not specified")
-	}
-	props := nodeConf.GetSourceConf(t, srcOptions)
+func NewLookupNode(ctx api.StreamContext, name string, isBytesLookup bool, fields []string, keys []string, joinType ast.JoinType, vals []ast.Expr, srcOptions *ast.Options, options *def.RuleOption, props map[string]any) (*LookupNode, error) {
 	lookupConf := &LookupConf{}
 	if lc, ok := props["lookup"].(map[string]interface{}); ok {
 		err := cast.MapToStruct(lc, lookupConf)
@@ -63,15 +61,32 @@ func NewLookupNode(name string, fields []string, keys []string, joinType ast.Joi
 		}
 	}
 	n := &LookupNode{
-		fields:     fields,
-		keys:       keys,
-		srcOptions: srcOptions,
-		conf:       lookupConf,
-		sourceType: t,
-		joinType:   joinType,
-		vals:       vals,
+		fields:        fields,
+		keys:          keys,
+		conf:          lookupConf,
+		joinType:      joinType,
+		vals:          vals,
+		isBytesLookup: isBytesLookup,
 	}
 	n.defaultSinkNode = newDefaultSinkNode(name, options)
+	if isBytesLookup {
+		sch := make(map[string]*ast.JsonStreamField, len(fields))
+		for _, field := range fields {
+			sch[field] = nil
+		}
+		decoder, err := converter.GetOrCreateConverter(ctx, &ast.Options{
+			FORMAT:     srcOptions.FORMAT,
+			SCHEMAID:   srcOptions.SCHEMAID,
+			DELIMITER:  srcOptions.DELIMITER,
+			IsWildCard: false,
+			Schema:     sch,
+		})
+		if err != nil {
+			msg := fmt.Sprintf("cannot get converter from format %s, schemaId %s: %v", srcOptions.FORMAT, srcOptions.SCHEMAID, err)
+			return nil, fmt.Errorf(msg)
+		}
+		n.formatDecoder = decoder
+	}
 	return n, nil
 }
 
@@ -164,7 +179,7 @@ func (n *LookupNode) Exec(ctx api.StreamContext, errCh chan<- error) {
 }
 
 // lookup will lookup the cache firstly, if expires, read the external source
-func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.Row, fv *xsql.FunctionValuer, ns api.LookupSource, tuples *xsql.JoinTuples, c *cache.Cache) error {
+func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.Row, fv *xsql.FunctionValuer, ns api.Source, tuples *xsql.JoinTuples, c *cache.Cache) error {
 	ve := &xsql.ValuerEval{Valuer: xsql.MultiValuer(d, fv)}
 	cvs := make([]interface{}, len(n.vals))
 	hasNil := false
@@ -184,14 +199,14 @@ func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.Row, fv *xsql.Function
 			k := fmt.Sprintf("%v", cvs)
 			r, ok = c.Get(k)
 			if !ok {
-				r, e = ns.Lookup(ctx, n.fields, n.keys, cvs)
+				r, e = n.doLookup(ctx, ns, cvs)
 				if e != nil {
 					return e
 				}
 				c.Set(k, r)
 			}
 		} else {
-			r, e = ns.Lookup(ctx, n.fields, n.keys, cvs)
+			r, e = n.doLookup(ctx, ns, cvs)
 		}
 	}
 	if e != nil {
@@ -256,3 +271,37 @@ func (n *LookupNode) merge(ctx api.StreamContext, d xsql.Row, r []map[string]int
 	n.statManager.IncTotalMessagesProcessed(1)
 	n.statManager.SetBufferLength(int64(len(n.input)))
 }
+
+func (n *LookupNode) doLookup(ctx api.StreamContext, ns api.Source, cvs []any) ([]map[string]any, error) {
+	if n.isBytesLookup {
+		rawRows, err := ns.(api.LookupBytesSource).Lookup(ctx, n.fields, n.keys, cvs)
+		if err != nil {
+			return nil, err
+		}
+		result := make([]map[string]any, 0, len(rawRows))
+		for _, row := range rawRows {
+			r, e := n.decode(ctx, row)
+			if e != nil {
+				ctx.GetLogger().Errorf("decode row %v error: %v", row, e)
+			} else {
+				switch rt := r.(type) {
+				case []map[string]any:
+					result = append(result, rt...)
+				case map[string]any:
+					result = append(result, rt)
+				default:
+					ctx.GetLogger().Errorf("decode row %v got unknow result: %v", row, rt)
+				}
+			}
+		}
+		return result, nil
+	} else {
+		return ns.(api.LookupSource).Lookup(ctx, n.fields, n.keys, cvs)
+	}
+}
+
+// Only called when isBytesLookup is true
+// Must guarantee decoders are set
+func (n *LookupNode) decode(ctx api.StreamContext, row []byte) (any, error) {
+	return n.formatDecoder.Decode(ctx, row)
+}
diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
index bf05883b..79e4dd23 100644
--- a/internal/topo/planner/planner.go
+++ b/internal/topo/planner/planner.go
@@ -253,7 +253,7 @@ func buildOps(lp LogicalPlan, tp *topo.Topo, options *def.RuleOption, sources ma
 	case *DedupTriggerPlan:
 		op = node.NewDedupTriggerNode(fmt.Sprintf("%d_dedup_trigger", newIndex), options, t.aliasName, t.startField.Name, t.endField.Name, t.nowField.Name, t.expire)
 	case *LookupPlan:
-		op, err = node.NewLookupNode(t.joinExpr.Name, t.fields, t.keys, t.joinExpr.JoinType, t.valvars, t.options, options)
+		op, err = planLookupSource(tp.GetContext(), t, options)
 	case *JoinAlignPlan:
 		op, err = node.NewJoinAlignNode(fmt.Sprintf("%d_join_aligner", newIndex), t.Emitters, options)
 	case *JoinPlan:
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index 4b09b79a..11bde073 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -271,3 +271,24 @@ func checkByteSource(ss api.Source) model.NodeInfo {
 		return model.NodeInfo{}
 	}
 }
+
+func planLookupSource(ctx api.StreamContext, t *LookupPlan, ruleOption *def.RuleOption) (node.Emitter, error) {
+	si, err := io.LookupSource(t.options.TYPE)
+	if err != nil {
+		return nil, err
+	}
+	if si == nil {
+		return nil, fmt.Errorf("lookup source type %s not found", t.options.TYPE)
+	}
+	props := nodeConf.GetSourceConf(t.options.TYPE, t.options)
+	switch si.(type) {
+	case api.LookupSource:
+		return node.NewLookupNode(ctx, t.joinExpr.Name, false, t.fields, t.keys, t.joinExpr.JoinType, t.valvars, t.options, ruleOption, props)
+	case api.LookupBytesSource:
+		if t.options.FORMAT == "" {
+			return nil, fmt.Errorf("lookup source type %s must specify format", t.options.TYPE)
+		}
+		return node.NewLookupNode(ctx, t.joinExpr.Name, true, t.fields, t.keys, t.joinExpr.JoinType, t.valvars, t.options, ruleOption, props)
+	}
+	return nil, fmt.Errorf("lookup source type %s is found but not a valid lookup source", t.options.TYPE)
+}
diff --git a/internal/io/connection/connection_mock.go b/pkg/connection/connection_mock.go
similarity index 78%
rename from internal/io/connection/connection_mock.go
rename to pkg/connection/connection_mock.go
index 863ff238..7490fff7 100644
--- a/internal/io/connection/connection_mock.go
+++ b/pkg/connection/connection_mock.go
@@ -61,8 +61,8 @@ func (m *mockConnection) Ref(ctx api.StreamContext) int {
 	return m.ref
 }
 
-func CreateMockConnection(ctx api.StreamContext, id string, props map[string]any) (modules.Connection, error) {
-	m := &mockConnection{id: id, ref: 0}
+func CreateMockConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
+	m := &mockConnection{ref: 0}
 	return m, nil
 }
 
@@ -76,22 +76,10 @@ func (m mockErrConnection) Close(ctx api.StreamContext) {
 	return
 }
 
-func (m mockErrConnection) Attach(ctx api.StreamContext) {
-	return
-}
-
 func (m mockErrConnection) DetachSub(ctx api.StreamContext, props map[string]any) {
 	return
 }
 
-func (m mockErrConnection) DetachPub(ctx api.StreamContext, props map[string]any) {
-	return
-}
-
-func (m mockErrConnection) Ref(ctx api.StreamContext) int {
-	return 0
-}
-
-func CreateMockErrConnection(ctx api.StreamContext, id string, props map[string]any) (modules.Connection, error) {
+func CreateMockErrConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
 	return nil, errors.New("mockErr")
 }
diff --git a/internal/io/connection/connection_pool.go b/pkg/connection/connection_pool.go
similarity index 73%
rename from internal/io/connection/connection_pool.go
rename to pkg/connection/connection_pool.go
index c5e3c33e..cdd2daaa 100644
--- a/internal/io/connection/connection_pool.go
+++ b/pkg/connection/connection_pool.go
@@ -31,7 +31,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
-func storeConnection(plugin, id string, props map[string]interface{}) error {
+func storeConnectionMeta(plugin, id string, props map[string]interface{}) error {
 	err := conf.WriteCfgIntoKVStorage("connections", plugin, id, props)
 	failpoint.Inject("storeConnectionErr", func() {
 		err = errors.New("storeConnectionErr")
@@ -47,6 +47,16 @@ func dropConnectionStore(plugin, id string) error {
 	return err
 }
 
+func GetConnectionRef(id string) int {
+	globalConnectionManager.RLock()
+	defer globalConnectionManager.RUnlock()
+	meta, ok := globalConnectionManager.connectionPool[id]
+	if !ok {
+		return 0
+	}
+	return meta.refCount
+}
+
 func GetAllConnectionStatus(ctx api.StreamContext) map[string]ConnectionStatus {
 	globalConnectionManager.RLock()
 	defer globalConnectionManager.RUnlock()
@@ -83,26 +93,71 @@ func GetAllConnectionsID() []string {
 }
 
 func PingConnection(ctx api.StreamContext, id string) error {
-	conn, err := GetNameConnection(id)
-	if err != nil {
-		return err
+	if id == "" {
+		return fmt.Errorf("connection id should be defined")
 	}
-	return conn.Ping(ctx)
+	globalConnectionManager.RLock()
+	defer globalConnectionManager.RUnlock()
+	meta, ok := globalConnectionManager.connectionPool[id]
+	if !ok {
+		return fmt.Errorf("connection %s not existed", id)
+	}
+	return meta.conn.Ping(ctx)
 }
 
-func GetNameConnection(selId string) (modules.Connection, error) {
-	if selId == "" {
+func FetchConnection(ctx api.StreamContext, id, typ string, props map[string]interface{}) (modules.Connection, error) {
+	if id == "" {
 		return nil, fmt.Errorf("connection id should be defined")
 	}
-	globalConnectionManager.RLock()
-	defer globalConnectionManager.RUnlock()
-	meta, ok := globalConnectionManager.connectionPool[selId]
+	selID := extractSelID(props)
+	if len(selID) < 1 {
+		return CreateNonStoredConnection(ctx, id, typ, props)
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	return attachConnection(selID)
+}
+
+func attachConnection(id string) (modules.Connection, error) {
+	if id == "" {
+		return nil, fmt.Errorf("connection id should be defined")
+	}
+	meta, ok := globalConnectionManager.connectionPool[id]
 	if !ok {
-		return nil, fmt.Errorf("connection %s not existed", selId)
+		return nil, fmt.Errorf("connection %s not existed", id)
 	}
+	meta.refCount++
 	return meta.conn, nil
 }
 
+func DetachConnection(ctx api.StreamContext, id string, props map[string]interface{}) error {
+	if id == "" {
+		return fmt.Errorf("connection id should be defined")
+	}
+	globalConnectionManager.Lock()
+	defer globalConnectionManager.Unlock()
+	selID := extractSelID(props)
+	if len(selID) < 1 {
+		return detachConnection(ctx, id, true)
+	}
+	return detachConnection(ctx, selID, false)
+}
+
+func detachConnection(ctx api.StreamContext, id string, remove bool) error {
+	meta, ok := globalConnectionManager.connectionPool[id]
+	if !ok {
+		return nil
+	}
+	if remove {
+		conn := meta.conn
+		conn.Close(ctx)
+		delete(globalConnectionManager.connectionPool, id)
+		return nil
+	}
+	meta.refCount--
+	return nil
+}
+
 func CreateNamedConnection(ctx api.StreamContext, id, typ string, props map[string]any) (modules.Connection, error) {
 	if id == "" || typ == "" {
 		return nil, fmt.Errorf("connection id and type should be defined")
@@ -113,12 +168,12 @@ func CreateNamedConnection(ctx api.StreamContext, id, typ string, props map[stri
 	if ok {
 		return nil, fmt.Errorf("connection %v already been created", id)
 	}
-	meta := ConnectionMeta{
+	meta := &ConnectionMeta{
 		ID:    id,
 		Typ:   typ,
 		Props: props,
 	}
-	if err := storeConnection(typ, id, props); err != nil {
+	if err := storeConnectionMeta(typ, id, props); err != nil {
 		return nil, err
 	}
 	conn, err := createNamedConnection(ctx, meta)
@@ -143,7 +198,7 @@ func CreateNonStoredConnection(ctx api.StreamContext, id, typ string, props map[
 	if ok {
 		return nil, fmt.Errorf("connection %v already been created", id)
 	}
-	meta := ConnectionMeta{
+	meta := &ConnectionMeta{
 		ID:    id,
 		Typ:   typ,
 		Props: props,
@@ -157,25 +212,9 @@ func CreateNonStoredConnection(ctx api.StreamContext, id, typ string, props map[
 	return conn, nil
 }
 
-func DropNonStoredConnection(ctx api.StreamContext, selId string) error {
-	if selId == "" {
-		return fmt.Errorf("connection id should be defined")
-	}
-	globalConnectionManager.Lock()
-	defer globalConnectionManager.Unlock()
-	meta, ok := globalConnectionManager.connectionPool[selId]
-	if !ok {
-		return nil
-	}
-	conn := meta.conn
-	conn.Close(ctx)
-	delete(globalConnectionManager.connectionPool, selId)
-	return nil
-}
-
 var mockErr = true
 
-func createNamedConnection(ctx api.StreamContext, meta ConnectionMeta) (modules.Connection, error) {
+func createNamedConnection(ctx api.StreamContext, meta *ConnectionMeta) (modules.Connection, error) {
 	var conn modules.Connection
 	var err error
 	connRegister, ok := modules.ConnectionRegister[strings.ToLower(meta.Typ)]
@@ -183,7 +222,7 @@ func createNamedConnection(ctx api.StreamContext, meta ConnectionMeta) (modules.
 		return nil, fmt.Errorf("unknown connection type")
 	}
 	err = backoff.Retry(func() error {
-		conn, err = connRegister(ctx, meta.ID, meta.Props)
+		conn, err = connRegister(ctx, meta.Props)
 		failpoint.Inject("createConnectionErr", func() {
 			if mockErr {
 				err = errorx.NewIOErr("createConnectionErr")
@@ -215,15 +254,14 @@ func DropNameConnection(ctx api.StreamContext, selId string) error {
 		}
 		return nil
 	}
-	conn := meta.conn
-	if conn.Ref(ctx) > 0 {
+	if meta.refCount > 0 {
 		return fmt.Errorf("connection %s can't be dropped due to reference", selId)
 	}
 	err := dropConnectionStore(meta.Typ, selId)
 	if err != nil {
 		return fmt.Errorf("drop connection %s failed, err:%v", selId, err)
 	}
-	conn.Close(ctx)
+	meta.conn.Close(ctx)
 	delete(globalConnectionManager.connectionPool, selId)
 	return nil
 }
@@ -238,7 +276,7 @@ func InitConnectionManager4Test() error {
 
 func InitConnectionManager() {
 	globalConnectionManager = &ConnectionManager{
-		connectionPool: make(map[string]ConnectionMeta),
+		connectionPool: make(map[string]*ConnectionMeta),
 		failConnection: make(map[string]string),
 	}
 	if conf.IsTesting {
@@ -259,7 +297,7 @@ func ReloadConnection() error {
 		}
 		typ := names[1]
 		id := names[2]
-		meta := ConnectionMeta{
+		meta := &ConnectionMeta{
 			ID:    id,
 			Typ:   typ,
 			Props: props,
@@ -278,15 +316,16 @@ func ReloadConnection() error {
 
 type ConnectionManager struct {
 	sync.RWMutex
-	connectionPool map[string]ConnectionMeta
+	connectionPool map[string]*ConnectionMeta
 	failConnection map[string]string
 }
 
 type ConnectionMeta struct {
-	ID    string             `json:"id"`
-	Typ   string             `json:"typ"`
-	Props map[string]any     `json:"props"`
-	conn  modules.Connection `json:"-"`
+	ID       string             `json:"id"`
+	Typ      string             `json:"typ"`
+	Props    map[string]any     `json:"props"`
+	conn     modules.Connection `json:"-"`
+	refCount int                `json:"-"`
 }
 
 func NewExponentialBackOff() *backoff.ExponentialBackOff {
@@ -313,3 +352,18 @@ type ConnectionStatus struct {
 	Status string
 	ErrMsg string
 }
+
+func extractSelID(props map[string]interface{}) string {
+	if len(props) < 1 {
+		return ""
+	}
+	v, ok := props["connectionSelector"]
+	if !ok {
+		return ""
+	}
+	id, ok := v.(string)
+	if !ok {
+		return ""
+	}
+	return id
+}
diff --git a/pkg/modules/connection.go b/pkg/modules/connection.go
index e693777b..d2e1fb13 100644
--- a/pkg/modules/connection.go
+++ b/pkg/modules/connection.go
@@ -19,13 +19,10 @@ import "github.com/lf-edge/ekuiper/contract/v2/api"
 type Connection interface {
 	Ping(ctx api.StreamContext) error
 	Close(ctx api.StreamContext)
-	Attach(ctx api.StreamContext)
 	DetachSub(ctx api.StreamContext, props map[string]any)
-	DetachPub(ctx api.StreamContext, props map[string]any)
-	Ref(ctx api.StreamContext) int
 }
 
-type ConnectionProvider func(ctx api.StreamContext, id string, props map[string]any) (Connection, error)
+type ConnectionProvider func(ctx api.StreamContext, props map[string]any) (Connection, error)
 
 var ConnectionRegister map[string]ConnectionProvider
 
diff --git a/pkg/modules/io.go b/pkg/modules/io.go
index 1fa34ef2..4b0f6cbf 100644
--- a/pkg/modules/io.go
+++ b/pkg/modules/io.go
@@ -20,7 +20,7 @@ import (
 
 type (
 	NewSourceFunc       func() api.Source
-	NewLookupSourceFunc func() api.LookupSource
+	NewLookupSourceFunc func() api.Source
 	NewSinkFunc         func() api.Sink
 )
 
@@ -41,3 +41,39 @@ func RegisterSink(name string, f NewSinkFunc) {
 func RegisterLookupSource(name string, f NewLookupSourceFunc) {
 	LookupSources[name] = f
 }
+
+func IsStreamSource(s api.Source) bool {
+	switch s.(type) {
+	case api.TupleSource, api.BytesSource, api.PullTupleSource, api.PullBytesSource:
+		return true
+	default:
+		return false
+	}
+}
+
+func IsPushStreamSource(s api.Source) bool {
+	switch s.(type) {
+	case api.TupleSource, api.BytesSource:
+		return true
+	default:
+		return false
+	}
+}
+
+func IsPullStreamSource(s api.Source) bool {
+	switch s.(type) {
+	case api.PullTupleSource, api.PullBytesSource:
+		return true
+	default:
+		return false
+	}
+}
+
+func IsLookupSource(s api.Source) bool {
+	switch s.(type) {
+	case api.LookupSource, api.LookupBytesSource:
+		return true
+	default:
+		return false
+	}
+}


Test patch:
diff --git a/internal/binder/io/binder_test.go b/internal/binder/io/binder_test.go
index f7a00144..c07ff64d 100644
--- a/internal/binder/io/binder_test.go
+++ b/internal/binder/io/binder_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -22,6 +22,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/binder"
 	"github.com/lf-edge/ekuiper/v2/internal/binder/mock"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
 func TestBindings(t *testing.T) {
@@ -58,22 +59,24 @@ func TestBindings(t *testing.T) {
 			isSink:         true,
 		},
 	}
-	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	for _, tt := range tests {
-		_, err := Source(tt.name)
-		isSource := err == nil
-		if tt.isSource != isSource {
-			t.Errorf("%s is source: expect %v but got %v", tt.name, tt.isSource, isSource)
-		}
-		_, err = LookupSource(tt.name)
-		if tt.isLookupSource != (err == nil) {
-			t.Errorf("%s is lookup source: expect %v but got %v", tt.name, tt.isLookupSource, err == nil)
-		}
-		_, err = Sink(tt.name)
-		isSink := err == nil
-		if tt.isSink != isSink {
-			t.Errorf("%s is sink: expect %v but got %v", tt.name, tt.isSink, isSink)
-		}
+		t.Run(tt.name, func(t *testing.T) {
+			ss, _ := Source(tt.name)
+			isSource := modules.IsStreamSource(ss)
+			if tt.isSource != isSource {
+				t.Errorf("%s is source: expect %v but got %v", tt.name, tt.isSource, isSource)
+			}
+			ss, _ = LookupSource(tt.name)
+			isLookupSource := modules.IsLookupSource(ss)
+			if tt.isLookupSource != isLookupSource {
+				t.Errorf("%s is lookup source: expect %v but got %v", tt.name, tt.isLookupSource, err == nil)
+			}
+			_, err = Sink(tt.name)
+			isSink := err == nil
+			if tt.isSink != isSink {
+				t.Errorf("%s is sink: expect %v but got %v", tt.name, tt.isSink, isSink)
+			}
+		})
 	}
 }
 
diff --git a/internal/io/memory/lookupsource_test.go b/internal/io/memory/lookupsource_test.go
index 571e9140..5bc76e71 100644
--- a/internal/io/memory/lookupsource_test.go
+++ b/internal/io/memory/lookupsource_test.go
@@ -87,7 +87,7 @@ func TestLookupProvision(t *testing.T) {
 func TestUpdateLookup(t *testing.T) {
 	contextLogger := conf.Log.WithField("rule", "test")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	ls := GetLookupSource()
+	ls := GetLookupSource().(api.LookupSource)
 	err := ls.Provision(ctx, map[string]interface{}{"datasource": "test", "key": "ff"})
 	assert.NoError(t, err)
 	err = ls.Connect(ctx)
@@ -131,7 +131,7 @@ func TestUpdateLookup(t *testing.T) {
 func TestLookup(t *testing.T) {
 	contextLogger := conf.Log.WithField("rule", "test2")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	ls := GetLookupSource()
+	ls := GetLookupSource().(api.LookupSource)
 	err := ls.Provision(ctx, map[string]any{"datasource": "test2", "key": "gg"})
 	assert.NoError(t, err)
 	err = ls.Connect(ctx)
diff --git a/internal/io/mqtt/sink_test.go b/internal/io/mqtt/sink_test.go
index 487562b0..36694043 100644
--- a/internal/io/mqtt/sink_test.go
+++ b/internal/io/mqtt/sink_test.go
@@ -22,8 +22,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
diff --git a/internal/io/mqtt/source_sink_test.go b/internal/io/mqtt/source_sink_test.go
index f5890368..c4f4be3c 100644
--- a/internal/io/mqtt/source_sink_test.go
+++ b/internal/io/mqtt/source_sink_test.go
@@ -22,10 +22,10 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mockclock"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/mock"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
 )
diff --git a/internal/io/mqtt/source_test.go b/internal/io/mqtt/source_test.go
index e347e42f..b9910b1b 100644
--- a/internal/io/mqtt/source_test.go
+++ b/internal/io/mqtt/source_test.go
@@ -22,10 +22,10 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 29162df0..06786809 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -32,13 +32,13 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
-	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/processor"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/rule"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 )
 
diff --git a/internal/topo/node/decode_op_test.go b/internal/topo/node/decode_op_test.go
index 9c7ee342..fc48562b 100644
--- a/internal/topo/node/decode_op_test.go
+++ b/internal/topo/node/decode_op_test.go
@@ -498,6 +498,25 @@ func TestPayloadBatchDecodeWithSchema(t *testing.T) {
 				errors.New("unsupported data received: &{test 0001-01-01 00:00:00 +0000 UTC [] map[topic:a] map[]}"),
 			},
 		},
+		{
+			name: "wrong payload field type",
+			input: &xsql.Tuple{
+				Emitter: "test",
+				Message: map[string]any{
+					"n": "outside", "frames": []any{
+						[]byte(`{"a":23,"b":34}`),
+					},
+				},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+			},
+			result: []any{
+				errors.New("unsupported payload received, must be a slice of maps: [[123 34 97 34 58 50 51 44 34 98 34 58 51 52 125]]"),
+			},
+		},
 	}
 	ctx := mockContext.NewMockContext("test1", "decode_test")
 	op, err := NewDecodeOp(ctx, true, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: 10}, &ast.Options{FORMAT: "json", SHARED: true}, false, false, nil, map[string]any{
diff --git a/internal/topo/node/lookup_node_test.go b/internal/topo/node/lookup_node_test.go
new file mode 100644
index 00000000..c1201e9c
--- /dev/null
+++ b/internal/topo/node/lookup_node_test.go
@@ -0,0 +1,357 @@
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package node
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/lookup"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/ast"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+func TestLookupInit(t *testing.T) {
+	ctx := mockContext.NewMockContext("test1", "testInit")
+	t.Run("error decode lookup conf", func(t *testing.T) {
+		_, err := NewLookupNode(ctx, "test2", false, []string{"la", "lb"}, []string{"test1"}, ast.LEFT_JOIN, []ast.Expr{}, &ast.Options{TYPE: "mock", FORMAT: "json"}, &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{"lookup": map[string]any{"cacheTtl": "nosense"}})
+		assert.Error(t, err)
+		assert.EqualError(t, err, "1 error(s) decoding:\n\n* error decoding 'cacheTtl': time: invalid duration \"nosense\"")
+	})
+
+	t.Run("wrong decoder format", func(t *testing.T) {
+		_, err := NewLookupNode(ctx, "test2", true, []string{"la", "lb"}, []string{"test1"}, ast.LEFT_JOIN, []ast.Expr{}, &ast.Options{TYPE: "mock", FORMAT: "json1"}, &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{})
+		assert.Error(t, err)
+		assert.EqualError(t, err, "cannot get converter from format json1, schemaId : format type json1 not supported")
+	})
+}
+
+func TestLookup(t *testing.T) {
+	tests := []struct {
+		name   string
+		input  any
+		result any // join tuple or error
+	}{
+		{
+			name: "normal",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": 2},
+			},
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 2},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 1.0, "lb": 1.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 2},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 2.0, "lb": 2.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 2},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 3.0, "lb": 4.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "lookup error",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "wrong"},
+			},
+			result: errors.New("mock lookup error"),
+		},
+		{
+			name: "window",
+			input: &xsql.WindowTuples{
+				Content: []xsql.Row{
+					&xsql.Tuple{
+						Emitter: "stream1",
+						Message: map[string]any{"a": 2},
+					},
+					&xsql.Tuple{
+						Emitter: "stream1",
+						Message: map[string]any{"a": 3},
+					},
+				},
+			},
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 2},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 1.0, "lb": 1.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 2},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 2.0, "lb": 2.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 2},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 3.0, "lb": 4.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 3},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 1.0, "lb": 1.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 3},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 2.0, "lb": 2.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 3},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 3.0, "lb": 4.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "empty",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": 0},
+			},
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": 0},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "decode error",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "jsonerror"},
+			},
+			// decode error, but still have result
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": "jsonerror"},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "decode to array",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "array"},
+			},
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": "array"},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 1.0, "lb": 1.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": "array"},
+							},
+							&xsql.Tuple{
+								Emitter:   "test2",
+								Message:   map[string]any{"la": 3.0, "lb": 4.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	modules.RegisterLookupSource("mock", func() api.Source {
+		return &MockLookupBytes{}
+	})
+
+	ctx, cancel := mockContext.NewMockContext("testRule", "test").WithCancel()
+	defer cancel()
+	op, err := NewLookupNode(ctx, "test2", true, []string{"la", "lb"}, []string{"test1"}, ast.LEFT_JOIN, []ast.Expr{&ast.FieldRef{
+		StreamName: "",
+		Name:       "a",
+	}}, &ast.Options{TYPE: "mock", FORMAT: "json"}, &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{})
+	assert.NoError(t, err)
+	out := make(chan any, 100)
+	err = op.AddOutput(out, "test")
+	assert.NoError(t, err)
+	errCh := make(chan error)
+	op.Exec(ctx, errCh)
+	err = <-errCh
+	assert.Error(t, err)
+	assert.EqualError(t, err, "lookup table test2 is not found")
+	// run table
+	err = lookup.CreateInstance("test2", "mock", &ast.Options{
+		DATASOURCE: "test2",
+		TYPE:       "mock",
+		KIND:       "lookup",
+		KEY:        "id",
+	})
+	assert.NoError(t, err)
+	op.Exec(ctx, errCh)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			op.input <- tt.input
+			r := <-out
+			assert.Equal(t, tt.result, r)
+		})
+	}
+}
+
+type MockLookupBytes struct{}
+
+func (m *MockLookupBytes) Provision(ctx api.StreamContext, configs map[string]any) error {
+	return nil
+}
+
+func (m *MockLookupBytes) Close(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *MockLookupBytes) Connect(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *MockLookupBytes) Lookup(ctx api.StreamContext, fields []string, keys []string, values []any) ([][]byte, error) {
+	if values[0] == "wrong" {
+		return nil, fmt.Errorf("mock lookup error")
+	} else if values[0] == 0 {
+		return nil, nil
+	} else if values[0] == "jsonerror" {
+		return [][]byte{
+			[]byte("{wrong jso}n"),
+		}, nil
+	} else if values[0] == "array" {
+		return [][]byte{
+			[]byte(`[{"la":1,"lb":1,"lc":1},{"la":3,"lb":4,"lc":1}]`),
+		}, nil
+	}
+	return [][]byte{
+		[]byte(`{"la":1,"lb":1,"lc":1}`),
+		[]byte(`{"la":2,"lb":2,"lc":1}`),
+		[]byte(`{"la":3,"lb":4,"lc":1}`),
+	}, nil
+}
diff --git a/internal/topo/planner/planner_source_test.go b/internal/topo/planner/planner_source_test.go
index 4d5c8b8a..9b5c8bb3 100644
--- a/internal/topo/planner/planner_source_test.go
+++ b/internal/topo/planner/planner_source_test.go
@@ -24,12 +24,14 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt"
 	"github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
@@ -464,3 +466,44 @@ func TestPlanError(t *testing.T) {
 		})
 	}
 }
+
+func TestPlanLookup(t *testing.T) {
+	ctx := mockContext.NewMockContext("Test", "test")
+	t.Run("undefined source", func(t *testing.T) {
+		_, err := planLookupSource(ctx, &LookupPlan{options: &ast.Options{TYPE: "none"}}, &def.RuleOption{})
+		assert.Error(t, err)
+		assert.EqualError(t, err, "lookup source type none not found")
+	})
+	t.Run("missing format", func(t *testing.T) {
+		modules.RegisterLookupSource("mock", func() api.Source {
+			return &MockLookupBytes{}
+		})
+		_, err := planLookupSource(ctx, &LookupPlan{options: &ast.Options{TYPE: "mock"}}, &def.RuleOption{})
+		assert.Error(t, err)
+		assert.EqualError(t, err, "lookup source type mock must specify format")
+	})
+	t.Run("register wrong source", func(t *testing.T) {
+		modules.RegisterLookupSource("mock", mqtt.GetSource)
+		_, err := planLookupSource(ctx, &LookupPlan{options: &ast.Options{TYPE: "mock"}}, &def.RuleOption{})
+		assert.Error(t, err)
+		assert.EqualError(t, err, "got non lookup source mock")
+	})
+}
+
+type MockLookupBytes struct{}
+
+func (m *MockLookupBytes) Provision(ctx api.StreamContext, configs map[string]any) error {
+	return nil
+}
+
+func (m *MockLookupBytes) Close(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *MockLookupBytes) Connect(ctx api.StreamContext) error {
+	return nil
+}
+
+func (m *MockLookupBytes) Lookup(ctx api.StreamContext, fields []string, keys []string, values []any) ([][]byte, error) {
+	return nil, nil
+}
diff --git a/internal/io/connection/connection_pool_test.go b/pkg/connection/connection_pool_test.go
similarity index 71%
rename from internal/io/connection/connection_pool_test.go
rename to pkg/connection/connection_pool_test.go
index f9019bb8..4b9f845c 100644
--- a/internal/io/connection/connection_pool_test.go
+++ b/pkg/connection/connection_pool_test.go
@@ -35,37 +35,52 @@ func TestConnection(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, conn)
 	require.NoError(t, conn.Ping(ctx))
+	require.Equal(t, 0, GetConnectionRef("id1"))
 	_, err = CreateNamedConnection(ctx, "id1", "mock", nil)
 	require.Error(t, err)
-	require.Equal(t, 0, conn.Ref(ctx))
-	conn.Attach(ctx)
-	require.Equal(t, 1, conn.Ref(ctx))
-	conn.Attach(ctx)
-	require.Equal(t, 2, conn.Ref(ctx))
-	conn.DetachPub(ctx, nil)
-	require.Equal(t, 1, conn.Ref(ctx))
+	attachConnection("id1")
+	require.Equal(t, 1, GetConnectionRef("id1"))
+	attachConnection("id1")
+	require.Equal(t, 2, GetConnectionRef("id1"))
+	detachConnection(ctx, "id1", false)
+	require.Equal(t, 1, GetConnectionRef("id1"))
 	err = DropNameConnection(ctx, "id1")
 	require.Error(t, err)
-	conn2, err := GetNameConnection("id1")
-	require.NoError(t, err)
-	require.NotNil(t, conn2)
-	conn.DetachSub(ctx, nil)
-	require.Equal(t, 0, conn.Ref(ctx))
+	detachConnection(ctx, "id1", false)
+	require.Equal(t, 0, GetConnectionRef("id1"))
 	err = DropNameConnection(ctx, "id1")
 	require.NoError(t, err)
 	err = DropNameConnection(ctx, "id1")
 	require.NoError(t, err)
-	conn3, err := GetNameConnection("id1")
+	conn3, err := attachConnection("id1")
 	require.Error(t, err)
 	require.Nil(t, conn3)
 
+	conn, err = CreateNamedConnection(ctx, "id2", "mock", nil)
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+
+	conn, err = FetchConnection(ctx, "2222", "mock", map[string]interface{}{"connectionSelector": "id2"})
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+
+	require.Equal(t, 1, GetConnectionRef("id2"))
+}
+
+func TestConnectionErr(t *testing.T) {
+	dataDir, err := conf.GetDataLoc()
+	require.NoError(t, err)
+	require.NoError(t, store.SetupDefault(dataDir))
+	require.NoError(t, InitConnectionManager4Test())
+	ctx := context.Background()
+
 	_, err = CreateNamedConnection(ctx, "", "mock", nil)
 	require.Error(t, err)
 	err = DropNameConnection(ctx, "")
 	require.Error(t, err)
 	_, err = CreateNamedConnection(ctx, "12", "unknown", nil)
 	require.Error(t, err)
-	_, err = GetNameConnection("")
+	_, err = attachConnection("")
 	require.Error(t, err)
 	err = PingConnection(ctx, "")
 	require.Error(t, err)
@@ -77,41 +92,33 @@ func TestConnection(t *testing.T) {
 	require.NotNil(t, conn4)
 	_, err = CreateNonStoredConnection(ctx, "id2", "mock", nil)
 	require.Error(t, err)
-	err = DropNonStoredConnection(ctx, "")
+	err = DetachConnection(ctx, "", nil)
 	require.Error(t, err)
-	err = DropNonStoredConnection(ctx, "nonexists")
-	require.NoError(t, err)
-}
-
-func TestConnectionErr(t *testing.T) {
-	dataDir, err := conf.GetDataLoc()
+	err = DetachConnection(ctx, "nonexists", nil)
 	require.NoError(t, err)
-	require.NoError(t, store.SetupDefault(dataDir))
-	require.NoError(t, InitConnectionManager4Test())
-	ctx := context.Background()
 
-	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/io/connection/createConnectionErr", "return(true)")
-	conn, err := createNamedConnection(ctx, ConnectionMeta{
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/pkg/createConnectionErr", "return(true)")
+	conn, err := createNamedConnection(ctx, &ConnectionMeta{
 		ID:    "1",
 		Typ:   "mock",
 		Props: nil,
 	})
 	require.NoError(t, err)
 	require.NotNil(t, conn)
-	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/connection/createConnectionErr")
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/pkg/connection/createConnectionErr")
 
-	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/io/connection/storeConnectionErr", "return(true)")
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/pkg/connection/storeConnectionErr", "return(true)")
 	_, err = CreateNamedConnection(ctx, "qwe", "mock", nil)
 	require.Error(t, err)
-	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/connection/storeConnectionErr")
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/pkg/connection/storeConnectionErr")
 
 	_, err = CreateNamedConnection(ctx, "qwe", "mock", nil)
 	require.NoError(t, err)
 
-	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/io/connection/dropConnectionStoreErr", "return(true)")
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/pkg/connection/dropConnectionStoreErr", "return(true)")
 	err = DropNameConnection(ctx, "qwe")
 	require.Error(t, err)
-	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/connection/dropConnectionStoreErr")
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/pkg/connection/dropConnectionStoreErr")
 }
 
 func TestConnectionStatus(t *testing.T) {