Repository: lf-edge/ekuiper
Pull request #2923 (2024-06-13 09:01:38)

feat(lookup): support lookup embedded format

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2923?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 `89.47368%` with `6 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@080fead`). [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/2923?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/topo/node/lookup\_node.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2923?src=pr&el=tree&filepath=internal%2Ftopo%2Fnode%2Flookup_node.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvdG9wby9ub2RlL2xvb2t1cF9ub2RlLmdv) | 89.47% | [5 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2923?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    #2923   +/-   ##
============================================
  Coverage             ?   69.33%           
============================================
  Files                ?      297           
  Lines                ?    28806           
  Branches             ?        0           
============================================
  Hits                 ?    19972           
  Misses               ?     7110           
  Partials             ?     1724           
```



</details>

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


Patch:
diff --git a/internal/converter/json/converter.go b/internal/converter/json/converter.go
index 32fe61f3..d81ea012 100644
--- a/internal/converter/json/converter.go
+++ b/internal/converter/json/converter.go
@@ -188,11 +188,14 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 		case fastjson.TypeNull:
 			m[key] = nil
 		case fastjson.TypeObject:
-			valid := f.checkSchema(key, "struct", schema)
+			add, valid := f.checkSchema(key, "struct", schema)
 			if !valid {
 				err = fmt.Errorf("%v has wrong type:%v, expect:%v", key, v.Type().String(), getType(schema[key]))
 				return
 			}
+			if !add {
+				return
+			}
 			childObj, err2 := v.Object()
 			if err2 != nil {
 				err = err2
@@ -211,11 +214,14 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 				m[key] = childMap
 			}
 		case fastjson.TypeArray:
-			valid := f.checkSchema(key, "array", schema)
+			add, valid := f.checkSchema(key, "array", schema)
 			if !valid {
 				err = fmt.Errorf("%v has wrong type:%v, expect:%v", key, v.Type().String(), getType(schema[key]))
 				return
 			}
+			if !add {
+				return
+			}
 			childArray, err2 := v.Array()
 			if err2 != nil {
 				err = err2
@@ -286,12 +292,22 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 	return m, nil
 }
 
-func (f *FastJsonConverter) checkSchema(key, typ string, schema map[string]*ast.JsonStreamField) (valid bool) {
-	v := schema[key]
+func (f *FastJsonConverter) checkSchema(key, typ string, schema map[string]*ast.JsonStreamField) (add, valid bool) {
+	// for schemaless, allow to decode the key value
+	if schema == nil {
+		return true, true
+	}
+	// for defined schema, skip to decode undefined key
+	v, ok := schema[key]
+	if !ok {
+		return false, true
+	}
+	// for the schema we didn't parse,allow to decode eg: results[0].a.b
 	if v == nil {
-		return true
+		return true, true
 	}
-	return v.Type == typ
+	// for the defined schema type, directly to check
+	return v.Type == typ, v.Type == typ
 }
 
 func (f *FastJsonConverter) extractNumberValue(name string, v *fastjson.Value, field *ast.JsonStreamField) (interface{}, error) {
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/node/lookup_node.go b/internal/topo/node/lookup_node.go
index e7950003..8db80c76 100644
--- a/internal/topo/node/lookup_node.go
+++ b/internal/topo/node/lookup_node.go
@@ -37,19 +37,28 @@ type LookupConf struct {
 	CacheMissingKey bool          `json:"cacheMissingKey"`
 }
 
+type srcConf struct {
+	PayloadField     string `json:"payloadField"`
+	PayloadFormat    string `json:"payloadFormat"`
+	PayloadSchemaId  string `json:"payloadSchemaId"`
+	PayloadDelimiter string `json:"payloadDelimiter"`
+}
+
 // LookupNode will look up the data from the external source when receiving an event
 type LookupNode struct {
 	*defaultSinkNode
 
 	conf *LookupConf
+	c    *srcConf
 
 	joinType ast.JoinType
 	vals     []ast.Expr
 	fields   []string
 	keys     []string
 	// If lookupByteSource, the decoders are needed
-	isBytesLookup bool
-	formatDecoder message.Converter
+	isBytesLookup  bool
+	formatDecoder  message.Converter
+	payloadDecoder message.Converter
 }
 
 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) {
@@ -70,22 +79,55 @@ func NewLookupNode(ctx api.StreamContext, name string, isBytesLookup bool, field
 	}
 	n.defaultSinkNode = newDefaultSinkNode(name, options)
 	if isBytesLookup {
+		sc := &srcConf{}
+		e := cast.MapToStruct(props, sc)
+		if e != nil {
+			return nil, e
+		}
+		if (sc.PayloadFormat == "" && sc.PayloadField != "") || (sc.PayloadFormat != "" && sc.PayloadField == "") {
+			return nil, fmt.Errorf("payloadFormat and payloadField must set together")
+		}
 		sch := make(map[string]*ast.JsonStreamField, len(fields))
 		for _, field := range fields {
 			sch[field] = nil
 		}
+		var fsch map[string]*ast.JsonStreamField
+		if sc.PayloadField != "" {
+			fsch = map[string]*ast.JsonStreamField{
+				sc.PayloadField: nil,
+			}
+		} else {
+			fsch = sch
+		}
+
 		decoder, err := converter.GetOrCreateConverter(ctx, &ast.Options{
 			FORMAT:     srcOptions.FORMAT,
 			SCHEMAID:   srcOptions.SCHEMAID,
 			DELIMITER:  srcOptions.DELIMITER,
 			IsWildCard: false,
-			Schema:     sch,
+			Schema:     fsch,
 		})
 		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
+
+		if sc.PayloadField != "" {
+			payloadDecoder, err := converter.GetOrCreateConverter(ctx, &ast.Options{
+				FORMAT:     sc.PayloadFormat,
+				SCHEMAID:   sc.PayloadSchemaId,
+				DELIMITER:  sc.PayloadDelimiter,
+				IsWildCard: false,
+				Schema:     sch,
+			})
+			if err != nil {
+				return nil, fmt.Errorf("cannot get payload converter from payloadFormat %s, schemaId %s: %v", sc.PayloadFormat, sc.PayloadSchemaId, err)
+			}
+			n.payloadDecoder = payloadDecoder
+		}
+
+		n.c = sc
 	}
 	return n, nil
 }
@@ -239,39 +281,6 @@ func (n *LookupNode) lookup(ctx api.StreamContext, d xsql.Row, fv *xsql.Function
 	}
 }
 
-func (n *LookupNode) merge(ctx api.StreamContext, d xsql.Row, 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: timex.GetNow(),
-		}
-		merged.AddTuple(t)
-		sets.Content = append(sets.Content, merged)
-	}
-
-	n.Broadcast(sets)
-	n.statManager.ProcessTimeEnd()
-	n.statManager.IncTotalRecordsOut()
-	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)
@@ -303,5 +312,45 @@ func (n *LookupNode) doLookup(ctx api.StreamContext, ns api.Source, cvs []any) (
 // 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)
+	r, e := n.formatDecoder.Decode(ctx, row)
+	if e == nil && n.payloadDecoder != nil {
+		switch rt := r.(type) {
+		case map[string]any:
+			return decodePayload(ctx, n.payloadDecoder, rt, n.c.PayloadField)
+		case []map[string]any:
+			result := make([]map[string]any, 0, len(rt))
+			for _, mm := range rt {
+				rr, e := decodePayload(ctx, n.payloadDecoder, mm, n.c.PayloadField)
+				if e != nil {
+					ctx.GetLogger().Warnf("decode payload of %v got error %v", mm, e)
+				} else {
+					switch rrt := rr.(type) {
+					case map[string]any:
+						result = append(result, rrt)
+					case []map[string]any:
+						result = append(result, rrt...)
+					default:
+						return nil, fmt.Errorf("payload decoder return non map or map slice")
+					}
+				}
+			}
+			return result, nil
+		default:
+			return nil, fmt.Errorf("decoder return non map or map slice")
+		}
+	}
+	return r, e
+}
+
+func decodePayload(ctx api.StreamContext, decoder message.Converter, rt map[string]any, field string) (any, error) {
+	payload, ok := rt[field]
+	if !ok {
+		ctx.GetLogger().Warnf("cannot find payload field %s", field)
+		return nil, nil
+	}
+	raw, err := cast.ToByteA(payload, cast.CONVERT_SAMEKIND)
+	if err != nil {
+		return nil, fmt.Errorf("payload is not bytes: %v", err)
+	}
+	return decoder.Decode(ctx, raw)
 }
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
 


Test patch:
diff --git a/internal/converter/json/converter_test.go b/internal/converter/json/converter_test.go
index 6124111d..3d012381 100644
--- a/internal/converter/json/converter_test.go
+++ b/internal/converter/json/converter_test.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
@@ -657,3 +658,53 @@ func TestSchemaless(t *testing.T) {
 		require.Equal(t, tc.expect, v)
 	}
 }
+
+func TestIssue(t *testing.T) {
+	originSchema := map[string]*ast.JsonStreamField{
+		"results": nil,
+	}
+	f := NewFastJsonConverter(originSchema)
+	data := `{
+    "results": [
+        {
+            "location": {
+                "id": "WTMKQ069CCJ7",
+                "name": "杭州",
+                "country": "CN",
+                "path": "杭州,杭州,浙江,中国",
+                "timezone": "Asia/Shanghai",
+                "timezone_offset": "+08:00"
+            },
+            "now": {
+                "text": "多云",
+                "code": "4",
+                "temperature": "31",
+                "feels_like": "36",
+                "pressure": "997",
+                "humidity": "58",
+                "visibility": "9.7",
+                "wind_direction": "东",
+                "wind_direction_degree": "87",
+                "wind_speed": "14.0",
+                "wind_scale": "3",
+                "clouds": "100",
+                "dew_point": ""
+            },
+            "last_update": "2024-06-13T16:37:50+08:00"
+        }
+    ]
+}`
+	m, err := f.Decode(context.Background(), []byte(data))
+	require.NoError(t, err)
+	expected := make(map[string]interface{})
+	json.Unmarshal([]byte(data), &expected)
+	require.Equal(t, expected, m)
+
+	schmema2 := map[string]*ast.JsonStreamField{
+		"others": nil,
+	}
+	f2 := NewFastJsonConverter(schmema2)
+	m, err = f2.Decode(context.Background(), []byte(data))
+	require.NoError(t, err)
+	require.Len(t, m, 0)
+}
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/lookup_node_test.go b/internal/topo/node/lookup_node_test.go
index c1201e9c..7ddf113e 100644
--- a/internal/topo/node/lookup_node_test.go
+++ b/internal/topo/node/lookup_node_test.go
@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -44,6 +45,24 @@ func TestLookupInit(t *testing.T) {
 		assert.Error(t, err)
 		assert.EqualError(t, err, "cannot get converter from format json1, schemaId : format type json1 not supported")
 	})
+
+	t.Run("wrong payload field", func(t *testing.T) {
+		_, err := NewLookupNode(ctx, "test2", true, []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{"payloadField": 1})
+		assert.Error(t, err)
+		assert.EqualError(t, err, "1 error(s) decoding:\n\n* 'payloadField' expected type 'string', got unconvertible type 'int', value: '1'")
+	})
+
+	t.Run("missing payload 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: "json"}, &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{"payloadField": "a"})
+		assert.Error(t, err)
+		assert.EqualError(t, err, "payloadFormat and payloadField must set together")
+	})
+
+	t.Run("wrong payload 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: "json"}, &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{"payloadField": "a", "payloadFormat": "json1"})
+		assert.Error(t, err)
+		assert.EqualError(t, err, "cannot get payload converter from payloadFormat json1, schemaId : format type json1 not supported")
+	})
 }
 
 func TestLookup(t *testing.T) {
@@ -211,7 +230,7 @@ func TestLookup(t *testing.T) {
 			name: "empty",
 			input: &xsql.Tuple{
 				Emitter: "stream1",
-				Message: map[string]any{"a": 0},
+				Message: map[string]any{"a": "empty"},
 			},
 			result: &xsql.JoinTuples{
 				Content: []*xsql.JoinTuple{
@@ -219,7 +238,7 @@ func TestLookup(t *testing.T) {
 						Tuples: []xsql.Row{
 							&xsql.Tuple{
 								Emitter: "stream1",
-								Message: map[string]any{"a": 0},
+								Message: map[string]any{"a": "empty"},
 							},
 						},
 					},
@@ -321,6 +340,198 @@ func TestLookup(t *testing.T) {
 	}
 }
 
+func TestLookupPayload(t *testing.T) {
+	tests := []struct {
+		name   string
+		input  any
+		result any // join tuple or error
+	}{
+		{
+			name: "normal payload",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "payload"},
+			},
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": "payload"},
+							},
+							&xsql.Tuple{
+								Emitter:   "testP",
+								Message:   map[string]any{"la": 11.0, "lb": 12.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name:   "invalid stream input",
+			input:  "test",
+			result: errors.New("run lookup node error: invalid input type but got string(test)"),
+		},
+		{
+			name: "use cache",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "payload"},
+			},
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": "payload"},
+							},
+							&xsql.Tuple{
+								Emitter:   "testP",
+								Message:   map[string]any{"la": 11.0, "lb": 12.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "normal payload array",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "payloadA"},
+			},
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": "payloadA"},
+							},
+							&xsql.Tuple{
+								Emitter:   "testP",
+								Message:   map[string]any{"la": 11.0, "lb": 12.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "payloadEmpty",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "payloadEmpty"},
+			},
+			result: &xsql.JoinTuples{
+				Content: make([]*xsql.JoinTuple, 0),
+			},
+		},
+		{
+			name: "payload not bytes",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "payload not bytes"},
+			},
+			result: &xsql.JoinTuples{
+				Content: make([]*xsql.JoinTuple, 0),
+			},
+		},
+		{
+			name: "array payload array",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "payloadAA"},
+			},
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": "payloadAA"},
+							},
+							&xsql.Tuple{
+								Emitter:   "testP",
+								Message:   map[string]any{"la": 11.0, "lb": 12.0},
+								Timestamp: timex.GetNow(),
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "array payload array array",
+			input: &xsql.Tuple{
+				Emitter: "stream1",
+				Message: map[string]any{"a": "payloadAAA"},
+			},
+			result: &xsql.JoinTuples{
+				Content: []*xsql.JoinTuple{
+					{
+						Tuples: []xsql.Row{
+							&xsql.Tuple{
+								Emitter: "stream1",
+								Message: map[string]any{"a": "payloadAAA"},
+							},
+							&xsql.Tuple{
+								Emitter:   "testP",
+								Message:   map[string]any{"la": 11.0, "lb": 12.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, "testP", true, []string{"la", "lb"}, []string{"test1"}, ast.INNER_JOIN, []ast.Expr{&ast.FieldRef{
+		StreamName: "",
+		Name:       "a",
+	}}, &ast.Options{TYPE: "mock", FORMAT: "json"}, &def.RuleOption{BufferLength: 10, SendError: true}, map[string]any{
+		"payloadField": "payload", "payloadFormat": "json", "lookup": map[string]any{"cache": true, "cacheTtl": "1s"},
+	})
+	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 testP is not found")
+	// run table
+	err = lookup.CreateInstance("testP", "mock", &ast.Options{
+		DATASOURCE: "testP",
+		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)
+		})
+	}
+	timex.Add(20 * time.Second)
+}
+
 type MockLookupBytes struct{}
 
 func (m *MockLookupBytes) Provision(ctx api.StreamContext, configs map[string]any) error {
@@ -336,22 +547,48 @@ func (m *MockLookupBytes) Connect(ctx api.StreamContext) error {
 }
 
 func (m *MockLookupBytes) Lookup(ctx api.StreamContext, fields []string, keys []string, values []any) ([][]byte, error) {
-	if values[0] == "wrong" {
+	switch values[0] {
+	case "wrong":
 		return nil, fmt.Errorf("mock lookup error")
-	} else if values[0] == 0 {
+	case "empty":
 		return nil, nil
-	} else if values[0] == "jsonerror" {
+	case "jsonerror":
 		return [][]byte{
 			[]byte("{wrong jso}n"),
 		}, nil
-	} else if values[0] == "array" {
+	case "array":
 		return [][]byte{
 			[]byte(`[{"la":1,"lb":1,"lc":1},{"la":3,"lb":4,"lc":1}]`),
 		}, nil
+	case "payload":
+		return [][]byte{
+			[]byte(`{"la":1,"lb":1,"payload":"{\"la\":11, \"lb\":12, \"lc\":1}"}`),
+		}, nil
+	case "payloadA":
+		return [][]byte{
+			[]byte(`{"la":1,"lb":1,"payload":"[{\"la\":11, \"lb\":12, \"lc\":1}]"}`),
+		}, nil
+	case "payloadAA":
+		return [][]byte{
+			[]byte(`[{"la":1,"lb":1,"payload":"{\"la\":11, \"lb\":12, \"lc\":1}"},{"la":1,"lb":1,"payload":"invalid"}]`),
+		}, nil
+	case "payloadAAA":
+		return [][]byte{
+			[]byte(`[{"la":1,"lb":1,"payload":"[{\"la\":11, \"lb\":12, \"lc\":1}]"}]`),
+		}, nil
+	case "payloadEmpty":
+		return [][]byte{
+			[]byte(`{"la":1,"lb":1,"data":"[{\"la\":11, \"lb\":12, \"lc\":1}]"}`),
+		}, nil
+	case "payloadNotByte":
+		return [][]byte{
+			[]byte(`{"la":1,"lb":1,"payload":34`),
+		}, nil
+	default:
+		return [][]byte{
+			[]byte(`{"la":1,"lb":1,"lc":1}`),
+			[]byte(`{"la":2,"lb":2,"lc":1}`),
+			[]byte(`{"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/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) {