Repository: lf-edge/ekuiper
Pull request #2880 (2024-05-28 03:13:50)

feat: support store configuration in kv

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?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 `16.66667%` with `55 lines` in your changes are missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@8e1c6bd`). [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/2880?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/conf/yaml\_config\_storage.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&filepath=internal%2Fconf%2Fyaml_config_storage.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvY29uZi95YW1sX2NvbmZpZ19zdG9yYWdlLmdv) | 0.00% | [17 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/meta/sinkMeta.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&filepath=internal%2Fmeta%2FsinkMeta.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvbWV0YS9zaW5rTWV0YS5nbw==) | 0.00% | [10 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/meta/sourceMeta.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&filepath=internal%2Fmeta%2FsourceMeta.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvbWV0YS9zb3VyY2VNZXRhLmdv) | 9.09% | [10 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/binder/meta/bind.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&filepath=internal%2Fbinder%2Fmeta%2Fbind.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvYmluZGVyL21ldGEvYmluZC5nbw==) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/conf/yaml\_config\_ops.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&filepath=internal%2Fconf%2Fyaml_config_ops.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvY29uZi95YW1sX2NvbmZpZ19vcHMuZ28=) | 62.50% | [3 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/conf/conf.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&filepath=internal%2Fconf%2Fconf.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvY29uZi9jb25mLmdv) | 0.00% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/server.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?src=pr&el=tree&filepath=internal%2Fserver%2Fserver.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL3NlcnZlci5nbw==) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?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    #2880   +/-   ##
============================================
  Coverage             ?   68.80%           
============================================
  Files                ?      279           
  Lines                ?    26602           
  Branches             ?        0           
============================================
  Hits                 ?    18303           
  Misses               ?     6665           
  Partials             ?     1634           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2880?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/etc/kuiper.yaml b/etc/kuiper.yaml
index cef56fc5..4a16183b 100644
--- a/etc/kuiper.yaml
+++ b/etc/kuiper.yaml
@@ -57,8 +57,6 @@ basic:
     # 0 indicates unlimited
     maxConnections: 0
   rulePatrolInterval: 10s
-  # cfgStorageType indicates the storage type to store the config, support "file","sqlite" and "fdb"
-  cfgStorageType: file
   # enableOpenZiti indicates whether to enable OpenZiti for eKuiper REST service. Currently, it is only supported to work with EdgeX secure mode.
   enableOpenZiti: false
   # AES Key, base64 encoded
diff --git a/internal/binder/meta/bind.go b/internal/binder/meta/bind.go
index c3974a19..f16cb5af 100644
--- a/internal/binder/meta/bind.go
+++ b/internal/binder/meta/bind.go
@@ -31,12 +31,18 @@ func Bind() {
 	}); nil != err {
 		conf.Log.Errorf("readSourceMetaDir:%v", err)
 	}
+	if err := meta.ReadSourceMetaData(); err != nil {
+		conf.Log.Errorf("readSourceMetaData:%v", err)
+	}
 	if err := meta.ReadSinkMetaDir(func(name string) bool {
 		s, _ := io.Sink(name)
 		return s != nil
 	}); nil != err {
 		conf.Log.Errorf("readSinkMetaDir:%v", err)
 	}
+	if err := meta.ReadSinkMetaData(); err != nil {
+		conf.Log.Errorf("readSinkMetaData:%v", err)
+	}
 	if err := meta.ReadUiMsgDir(); nil != err {
 		conf.Log.Errorf("readUiMsgDir:%v", err)
 	}
diff --git a/internal/conf/conf.go b/internal/conf/conf.go
index 9c9e5736..843439cf 100644
--- a/internal/conf/conf.go
+++ b/internal/conf/conf.go
@@ -192,7 +192,6 @@ type KuiperConf struct {
 		IgnoreCase              bool              `yaml:"ignoreCase"`
 		SQLConf                 *SQLConf          `yaml:"sql"`
 		RulePatrolInterval      cast.DurationConf `yaml:"rulePatrolInterval"`
-		CfgStorageType          string            `yaml:"cfgStorageType"`
 		EnableOpenZiti          bool              `yaml:"enableOpenZiti"`
 		AesKey                  string            `yaml:"aesKey"`
 		GracefulShutdownTimeout cast.DurationConf `yaml:"gracefulShutdownTimeout"`
@@ -397,10 +396,6 @@ func InitConf() {
 	if Config.Source == nil {
 		Config.Source = &SourceConf{}
 	}
-	if Config.Basic.CfgStorageType == "" {
-		Config.Basic.CfgStorageType = "file"
-	}
-
 	_ = Config.Source.Validate()
 	if Config.Sink == nil {
 		Config.Sink = &SinkConf{}
diff --git a/internal/conf/yaml_config_ops.go b/internal/conf/yaml_config_ops.go
index e505e522..8b61fac6 100644
--- a/internal/conf/yaml_config_ops.go
+++ b/internal/conf/yaml_config_ops.go
@@ -17,12 +17,10 @@ package conf
 import (
 	"encoding/json"
 	"fmt"
-	"os"
 	"path"
 	"reflect"
 	"sync"
 
-	"github.com/lf-edge/ekuiper/v2/internal/pkg/filex"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 )
 
@@ -64,7 +62,7 @@ type ConfigKeys struct {
 	lock        sync.RWMutex
 	pluginName  string                            // source type, can be mqtt/edgex/httppull
 	etcCfg      map[string]map[string]interface{} // configs defined in etc/sources/yaml
-	dataCfg     map[string]map[string]interface{} // configs defined in etc/sources/
+	dataCfg     map[string]map[string]interface{}
 	// delCfgKey save the config key which needs to be deleted from the storage
 	delCfgKey map[string]struct{}
 	// saveCfgKey save the config key which needs to be saved or updated into the storage
@@ -328,18 +326,6 @@ func (c *SourceConfigKeysOps) SaveCfgToStorage() error {
 	switch c.storageType {
 	case cfgStoreKVStorage:
 		return c.ConfigKeys.saveCfgKeysIntoKVStorage("sources")
-	case cfgFileStorage:
-		pluginName := c.pluginName
-		confDir, err := GetDataLoc()
-		if nil != err {
-			return err
-		}
-
-		dir := path.Join(confDir, "sources")
-		filePath := path.Join(dir, pluginName+".yaml")
-		cfg := c.CopyUpdatableConfContent()
-		err = filex.WriteYamlMarshal(filePath, cfg)
-		return err
 	}
 	return fmt.Errorf("unknown source cfg storage type: %v", c.storageType)
 }
@@ -353,26 +339,6 @@ func (c *SinkConfigKeysOps) SaveCfgToStorage() error {
 	switch c.storageType {
 	case cfgStoreKVStorage:
 		return c.ConfigKeys.saveCfgKeysIntoKVStorage("sinks")
-	case cfgFileStorage:
-		pluginName := c.pluginName
-		confDir, err := GetDataLoc()
-		if nil != err {
-			return err
-		}
-
-		dir := path.Join(confDir, "sinks")
-		if IsTesting {
-			if _, err := os.Stat(dir); os.IsNotExist(err) {
-				err = os.MkdirAll(dir, 0o755)
-				if err != nil {
-					return err
-				}
-			}
-		}
-		filePath := path.Join(dir, pluginName+".yaml")
-		cfg := c.CopyUpdatableConfContent()
-		err = filex.WriteYamlMarshal(filePath, cfg)
-		return err
 	}
 	return fmt.Errorf("unknown sinks cfg storage type: %v", c.storageType)
 }
@@ -386,24 +352,6 @@ func (p *ConnectionConfigKeysOps) SaveCfgToStorage() error {
 	switch p.storageType {
 	case cfgStoreKVStorage:
 		return p.ConfigKeys.saveCfgKeysIntoKVStorage("connections")
-	case cfgFileStorage:
-		pluginName := p.pluginName
-		confDir, err := GetDataLoc()
-		if nil != err {
-			return err
-		}
-
-		cfg := p.CopyUpdatableConfContent()
-
-		yamlPath := path.Join(confDir, "connections/connection.yaml")
-
-		yamlData := make(map[string]interface{})
-		err = filex.ReadYamlUnmarshal(yamlPath, &yamlData)
-		if nil != err {
-			return err
-		}
-		yamlData[pluginName] = cfg
-		return filex.WriteYamlMarshal(yamlPath, yamlData)
 	}
 	return fmt.Errorf("unknown connection cfg storage type: %v", p.ConfigKeys.storageType)
 }
@@ -452,24 +400,12 @@ func NewConfigOperatorFromSourceStorage(pluginName string) (ConfigOperator, erro
 	// Just ignore error if yaml not found
 	_ = LoadConfigFromPath(filePath, &c.etcCfg)
 
-	if c.storageType != cfgFileStorage {
-		prefix := buildKey("sources", pluginName, "")
-		dataCfg, err := getCfgKeyFromStorageByPrefix(prefix)
-		if err != nil {
-			return nil, err
-		}
-		c.dataCfg = dataCfg
-	} else {
-		dataDir, err := GetDataLoc()
-		if nil != err {
-			return nil, err
-		}
-		dir = path.Join(dataDir, "sources")
-		fileName = pluginName
-
-		filePath = path.Join(dir, fileName+`.yaml`)
-		_ = filex.ReadYamlUnmarshal(filePath, &c.dataCfg)
+	prefix := buildKey("sources", pluginName, "")
+	dataCfg, err := getCfgKeyFromStorageByPrefix(prefix)
+	if err != nil {
+		return nil, err
 	}
+	c.dataCfg = dataCfg
 	return c, nil
 }
 
@@ -502,23 +438,12 @@ func NewConfigOperatorFromSinkStorage(pluginName string) (ConfigOperator, error)
 			saveCfgKey:  map[string]struct{}{},
 		},
 	}
-	if c.storageType != cfgFileStorage {
-		prefix := buildKey("sinks", pluginName, "")
-		dataCfg, err := getCfgKeyFromStorageByPrefix(prefix)
-		if err != nil {
-			return nil, err
-		}
-		c.dataCfg = dataCfg
-	} else {
-		dataDir, err := GetDataLoc()
-		if nil != err {
-			return nil, err
-		}
-		dir := path.Join(dataDir, "sinks")
-
-		filePath := path.Join(dir, pluginName+`.yaml`)
-		_ = filex.ReadYamlUnmarshal(filePath, &c.dataCfg)
+	prefix := buildKey("sinks", pluginName, "")
+	dataCfg, err := getCfgKeyFromStorageByPrefix(prefix)
+	if err != nil {
+		return nil, err
 	}
+	c.dataCfg = dataCfg
 	return c, nil
 }
 
@@ -578,43 +503,15 @@ func NewConfigOperatorFromConnectionStorage(pluginName string) (ConfigOperator,
 		return nil, fmt.Errorf("not find the target connection type: %s", c.pluginName)
 	}
 
-	if c.storageType != cfgFileStorage {
-		prefix := buildKey("connections", pluginName, "")
-		dataCfg, err := getCfgKeyFromStorageByPrefix(prefix)
-		if err != nil {
-			return nil, err
-		}
-		c.dataCfg = dataCfg
-	} else {
-		confDir, err = GetDataLoc()
-		if nil != err {
-			return nil, err
-		}
-		yamlPath = path.Join(confDir, "connections/connection.yaml")
-		yamlData = make(map[string]interface{})
-		_ = filex.ReadYamlUnmarshal(yamlPath, &yamlData)
-
-		if plgCnfs, ok := yamlData[pluginName]; ok {
-			if cf, ok1 := plgCnfs.(map[string]interface{}); ok1 {
-				for confKey, confVal := range cf {
-					if conf, ok := confVal.(map[string]interface{}); ok {
-						c.dataCfg[confKey] = conf
-					} else {
-						return nil, fmt.Errorf("file content is not right: %s.%v", confKey, confVal)
-					}
-				}
-			} else {
-				return nil, fmt.Errorf("file content is not right: %v", plgCnfs)
-			}
-		}
+	prefix := buildKey("connections", pluginName, "")
+	dataCfg, err := getCfgKeyFromStorageByPrefix(prefix)
+	if err != nil {
+		return nil, err
 	}
+	c.dataCfg = dataCfg
 	return c, nil
 }
 
 func getStorageType() string {
-	st := cfgFileStorage
-	if Config != nil && len(Config.Basic.CfgStorageType) > 0 {
-		st = Config.Basic.CfgStorageType
-	}
-	return st
+	return cfgStoreKVStorage
 }
diff --git a/internal/conf/yaml_config_storage.go b/internal/conf/yaml_config_storage.go
index 9de1eb2a..ff602694 100644
--- a/internal/conf/yaml_config_storage.go
+++ b/internal/conf/yaml_config_storage.go
@@ -16,15 +16,16 @@ package conf
 
 import (
 	"bytes"
-	"fmt"
+	"errors"
 	"strings"
 
+	"github.com/pingcap/failpoint"
+
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/pkg/kv"
 )
 
 const (
-	cfgFileStorage    = "file"
 	cfgStoreKVStorage = "kv"
 )
 
@@ -60,10 +61,39 @@ func (m *kvMemory) GetByPrefix(prefix string) (map[string]map[string]interface{}
 
 var (
 	mockMemoryKVStore *kvMemory
-	sqliteKVStore     *sqlKVStore
+	kvStore           *sqlKVStore
 )
 
-func getKVStorage() (cfgKVStorage, error) {
+// GetYamlConfigAllKeys get all plugin keys about sources/sinks/connections
+func GetYamlConfigAllKeys(typ string) (map[string]struct{}, error) {
+	s, err := getKVStorage()
+	if err != nil {
+		return nil, err
+	}
+	data, err := s.GetByPrefix(typ)
+	failpoint.Inject("getDataErr", func() {
+		err = errors.New("getDataErr")
+	})
+	if err != nil {
+		return nil, err
+	}
+	s1 := make(map[string]struct{})
+	for key := range data {
+		names := strings.Split(key, ".")
+		if len(names) != 3 {
+			continue
+		}
+		s1[names[1]] = struct{}{}
+	}
+	return s1, nil
+}
+
+func getKVStorage() (s cfgKVStorage, err error) {
+	defer func() {
+		failpoint.Inject("storageErr", func() {
+			err = errors.New("storageErr")
+		})
+	}()
 	if IsTesting {
 		if mockMemoryKVStore == nil {
 			mockMemoryKVStore = &kvMemory{}
@@ -71,18 +101,19 @@ func getKVStorage() (cfgKVStorage, error) {
 		}
 		return mockMemoryKVStore, nil
 	}
-	switch Config.Basic.CfgStorageType {
-	case cfgStoreKVStorage:
-		if sqliteKVStore == nil {
-			sqliteKVStorage, err := NewSqliteKVStore("confKVStorage")
-			if err != nil {
-				return nil, err
-			}
-			sqliteKVStore = sqliteKVStorage
+	if kvStore == nil {
+		sqliteKVStorage, err := NewSqliteKVStore("confKVStorage")
+		if err != nil {
+			return nil, err
 		}
-		return sqliteKVStore, nil
+		kvStore = sqliteKVStorage
 	}
-	return nil, fmt.Errorf("unknown cfg kv storage type: %v", Config.Basic.CfgStorageType)
+	return kvStore, nil
+}
+
+// SaveCfgKeyToKVInTest only used in unit test
+func SaveCfgKeyToKVInTest(key string, cfg map[string]interface{}) error {
+	return saveCfgKeyToKV(key, cfg)
 }
 
 func saveCfgKeyToKV(key string, cfg map[string]interface{}) error {
diff --git a/internal/converter/converter.go b/internal/converter/converter.go
index cf81a55f..83599482 100644
--- a/internal/converter/converter.go
+++ b/internal/converter/converter.go
@@ -30,9 +30,6 @@ import (
 
 func init() {
 	modules.RegisterConverter(message.FormatJson, func(_ api.StreamContext, _ string, _ string, _ string, schema map[string]*ast.JsonStreamField) (message.Converter, error) {
-		if schema == nil {
-			return json.GetConverter()
-		}
 		return json.NewFastJsonConverter(schema), nil
 	})
 	modules.RegisterConverter(message.FormatBinary, func(_ api.StreamContext, _ string, _ string, _ string, _ map[string]*ast.JsonStreamField) (message.Converter, error) {
diff --git a/internal/converter/json/converter.go b/internal/converter/json/converter.go
index 82114233..32fe61f3 100644
--- a/internal/converter/json/converter.go
+++ b/internal/converter/json/converter.go
@@ -25,35 +25,8 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
-	"github.com/lf-edge/ekuiper/v2/pkg/message"
 )
 
-type Converter struct{}
-
-var converter = &Converter{}
-
-func GetConverter() (message.Converter, error) {
-	return converter, nil
-}
-
-func (c *Converter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
-	return json.Marshal(d)
-}
-
-func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
-	defer func() {
-		if err != nil {
-			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
-		}
-	}()
-	var r0 interface{}
-	err = json.Unmarshal(b, &r0)
-	if err != nil {
-		return nil, err
-	}
-	return r0, nil
-}
-
 type FastJsonConverter struct {
 	sync.RWMutex
 	schema map[string]*ast.JsonStreamField
@@ -66,28 +39,30 @@ func NewFastJsonConverter(schema map[string]*ast.JsonStreamField) *FastJsonConve
 	return f
 }
 
-func (c *FastJsonConverter) ResetSchema(schema map[string]*ast.JsonStreamField) {
-	c.Lock()
-	defer c.Unlock()
-	c.schema = schema
+func (f *FastJsonConverter) ResetSchema(schema map[string]*ast.JsonStreamField) {
+	f.Lock()
+	defer f.Unlock()
+	f.schema = schema
 }
 
-func (c *FastJsonConverter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
+func (f *FastJsonConverter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
 	return json.Marshal(d)
 }
 
-func (c *FastJsonConverter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
+func (f *FastJsonConverter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 	defer func() {
 		if err != nil {
 			err = errorx.NewWithCode(errorx.CovnerterErr, err.Error())
 		}
 	}()
-	c.RLock()
-	defer c.RUnlock()
-	if c.schema == nil {
-		return converter.Decode(ctx, b)
+	f.RLock()
+	defer f.RUnlock()
+	if f.schema == nil {
+		var r any
+		err = json.Unmarshal(b, &r)
+		return r, err
 	}
-	return c.decodeWithSchema(b, c.schema)
+	return f.decodeWithSchema(b, f.schema)
 }
 
 func (f *FastJsonConverter) decodeWithSchema(b []byte, schema map[string]*ast.JsonStreamField) (interface{}, error) {
diff --git a/internal/converter/schema/schema.go b/internal/converter/schema/schema.go
index 4fba89d1..b71f6535 100644
--- a/internal/converter/schema/schema.go
+++ b/internal/converter/schema/schema.go
@@ -38,13 +38,15 @@ func NewSchemaLayer(ruleID, streamName string, schema map[string]*ast.JsonStream
 		streamMap:   make(map[string]string),
 		wildcardMap: make(map[string]struct{}),
 	}
-	s.schemaMap[ruleID] = schema
-	s.streamMap[ruleID] = streamName
-	if isWildcard {
-		s.wildcardMap[ruleID] = struct{}{}
+	if schema != nil || isWildcard {
+		s.schemaMap[ruleID] = schema
+		s.streamMap[ruleID] = streamName
+		if isWildcard {
+			s.wildcardMap[ruleID] = struct{}{}
+		}
+		s.schema = schema
+		s.storeSchema()
 	}
-	s.schema = schema
-	s.storeSchema()
 	return s
 }
 
diff --git a/internal/meta/sinkMeta.go b/internal/meta/sinkMeta.go
index 0f346a10..5b469c4c 100644
--- a/internal/meta/sinkMeta.go
+++ b/internal/meta/sinkMeta.go
@@ -187,36 +187,25 @@ func newUiSink(fi *fileSink) (*uiSink, error) {
 
 var gSinkmetadata = make(map[string]*uiSink) // immutable
 
-func ReadSinkMetaDir(checker InstallChecker) error {
-	confDir, err := conf.GetConfLoc()
-	if nil != err {
+func ReadSinkMetaData() error {
+	keys, err := conf.GetYamlConfigAllKeys("sinks")
+	if err != nil {
 		return err
 	}
-
-	dir := path.Join(confDir, "sinks")
-	files, err := os.ReadDir(dir)
-	if nil != err {
-		return err
-	}
-	for _, file := range files {
-		fname := file.Name()
-		if !strings.HasSuffix(fname, ".json") {
-			continue
-		}
-
-		filePath := path.Join(dir, fname)
-		if err := ReadSinkMetaFile(filePath, checker(strings.TrimSuffix(fname, ".json"))); nil != err {
-			return err
-		}
+	for key := range keys {
+		loadConfigOperatorForSink(key)
 	}
+	return nil
+}
 
-	confDir, err = conf.GetDataLoc()
+func ReadSinkMetaDir(checker InstallChecker) error {
+	confDir, err := conf.GetConfLoc()
 	if nil != err {
 		return err
 	}
 
-	dir = path.Join(confDir, "sinks")
-	files, err = os.ReadDir(dir)
+	dir := path.Join(confDir, "sinks")
+	files, err := os.ReadDir(dir)
 	if nil != err {
 		return err
 	}
diff --git a/internal/meta/sourceMeta.go b/internal/meta/sourceMeta.go
index 34978c55..0f036611 100644
--- a/internal/meta/sourceMeta.go
+++ b/internal/meta/sourceMeta.go
@@ -108,11 +108,9 @@ func ReadSourceMetaFile(filePath string, isScan bool, isLookup bool) error {
 	gSourcemetaLock.Lock()
 	gSourcemetadata[fileName] = meta
 	gSourcemetaLock.Unlock()
-
 	loadConfigOperatorForSource(strings.TrimSuffix(fileName, `.json`))
 	loadConfigOperatorForConnection(strings.TrimSuffix(fileName, `.json`))
-
-	return err
+	return nil
 }
 
 func ReadSourceMetaDir(scanChecker InstallChecker, lookupChecker InstallChecker) error {
@@ -150,37 +148,24 @@ func ReadSourceMetaDir(scanChecker InstallChecker, lookupChecker InstallChecker)
 			}
 		}
 	}
+	return nil
+}
 
-	// load data/sources meta data
-	confDir, err = conf.GetDataLoc()
-	if nil != err {
+func ReadSourceMetaData() error {
+	keys, err := conf.GetYamlConfigAllKeys("sources")
+	if err != nil {
 		return err
 	}
-
-	dir = path.Join(confDir, "sources")
-	dirEntries, err = os.ReadDir(dir)
-	if nil != err {
+	for key := range keys {
+		loadConfigOperatorForSource(key)
+	}
+	keys, err = conf.GetYamlConfigAllKeys("connections")
+	if err != nil {
 		return err
 	}
-
-	for _, entry := range dirEntries {
-		fileName := entry.Name()
-		if strings.HasSuffix(fileName, ".json") {
-			name := strings.TrimSuffix(fileName, ".json")
-			isScan := scanChecker(name)
-			isLookup := lookupChecker(name)
-			if isScan || isLookup {
-				filePath := path.Join(dir, fileName)
-				if err = ReadSourceMetaFile(filePath, isScan, isLookup); nil != err {
-					return err
-				}
-				conf.Log.Infof("Loading metadata file for source : %s", fileName)
-			} else {
-				conf.Log.Warnf("Find source metadata file but not installed : %s", fileName)
-			}
-		}
+	for key := range keys {
+		loadConfigOperatorForConnection(key)
 	}
-
 	return nil
 }
 
diff --git a/internal/processor/rule_query.go b/internal/processor/rule_query.go
index bf5741a1..5d419cc2 100644
--- a/internal/processor/rule_query.go
+++ b/internal/processor/rule_query.go
@@ -13,7 +13,6 @@
 // limitations under the License.
 
 //go:build rpc || !core
-// +build rpc !core
 
 package processor
 
diff --git a/internal/topo/node/decode_op.go b/internal/topo/node/decode_op.go
index 34a0ba4f..75650508 100644
--- a/internal/topo/node/decode_op.go
+++ b/internal/topo/node/decode_op.go
@@ -34,18 +34,35 @@ type DecodeOp struct {
 	*defaultSinkNode
 	converter message.Converter
 	sLayer    *schemaLayer.SchemaLayer
-	// When receiving list, send them one by one, this is the sending interval between each
-	// Typically set by file source
-	sendInterval time.Duration
+
+	c *dconf
+	// This is for first level decode, add the payload field to schema to make sure it is decoded
+	forPayload     bool
+	additionSchema string
 }
 
 type dconf struct {
-	SendInterval time.Duration `json:"sendInterval"`
+	// When receiving list, send them one by one, this is the sending interval between each
+	// Typically set by file source
+	SendInterval      time.Duration `json:"sendInterval"`
+	PayloadField      string        `json:"payloadField"`
+	PayloadBatchField string        `json:"payloadBatchField"`
+	PayloadFormat     string        `json:"payloadFormat"`
+	PayloadSchemaId   string        `json:"payloadSchemaId"`
+	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 {
@@ -67,13 +84,40 @@ func (o *DecodeOp) DetachSchema(ctx api.StreamContext, ruleId string) {
 	}
 }
 
-func NewDecodeOp(ctx api.StreamContext, 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) {
+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)
+	if e != nil {
+		return nil, e
+	}
+	if forPayload && dc.PayloadFormat == "" {
+		return nil, fmt.Errorf("payloadFormat is missing")
+	}
+	var additionSchema string
+	// It is payload decoder
+	if forPayload {
+		options = &ast.Options{
+			FORMAT:    dc.PayloadFormat,
+			SCHEMAID:  dc.PayloadSchemaId,
+			DELIMITER: dc.PayloadDelimiter,
+		}
+	} else {
+		if dc.PayloadBatchField != "" {
+			additionSchema = dc.PayloadBatchField
+		} else if dc.PayloadField != "" {
+			additionSchema = dc.PayloadField
+		}
+	}
+
 	options.Schema = nil
 	options.IsWildCard = isWildcard
 	options.IsSchemaLess = isSchemaless
 	if schema != nil {
 		options.Schema = schema
 		options.StreamName = StreamName
+		if additionSchema != "" {
+			options.Schema[additionSchema] = nil
+		}
 	}
 	options.RuleID = ruleId
 	converterTool, err := converter.GetOrCreateConverter(ctx, options)
@@ -81,17 +125,16 @@ func NewDecodeOp(ctx api.StreamContext, name, StreamName string, ruleId string,
 		msg := fmt.Sprintf("cannot get converter from format %s, schemaId %s: %v", options.FORMAT, options.SCHEMAID, err)
 		return nil, fmt.Errorf(msg)
 	}
-	dc := &dconf{}
-	e := cast.MapToStruct(props, dc)
-	if e != nil {
-		return nil, e
-	}
-	return &DecodeOp{
+	o := &DecodeOp{
 		defaultSinkNode: newDefaultSinkNode(name, rOpt),
 		converter:       converterTool,
 		sLayer:          schemaLayer.NewSchemaLayer(ruleId, StreamName, schema, isWildcard),
-		sendInterval:    dc.SendInterval,
-	}, nil
+		c:               dc,
+		forPayload:      forPayload,
+		additionSchema:  additionSchema,
+	}
+
+	return o, nil
 }
 
 // Exec decode op receives raw data and converts it to message
@@ -101,8 +144,18 @@ func (o *DecodeOp) Exec(ctx api.StreamContext, errCh chan<- error) {
 		defer func() {
 			o.Close()
 		}()
+		var w workerFunc
+		if o.forPayload {
+			if o.c.PayloadBatchField != "" {
+				w = o.PayloadBatchDecodeWorker
+			} else {
+				w = o.PayloadDecodeWorker
+			}
+		} else {
+			w = o.Worker
+		}
 		err := infra.SafeRun(func() error {
-			runWithOrderAndInterval(ctx, o.defaultSinkNode, o.concurrency, o.Worker, o.sendInterval)
+			runWithOrderAndInterval(ctx, o.defaultSinkNode, o.concurrency, w, o.c.SendInterval)
 			return nil
 		})
 		if err != nil {
@@ -115,8 +168,6 @@ func (o *DecodeOp) Worker(ctx api.StreamContext, item any) []any {
 	o.statManager.ProcessTimeStart()
 	defer o.statManager.ProcessTimeEnd()
 	switch d := item.(type) {
-	case error:
-		return []any{d}
 	case *xsql.RawTuple:
 		result, err := o.converter.Decode(ctx, d.Raw())
 		if err != nil {
@@ -124,18 +175,18 @@ func (o *DecodeOp) Worker(ctx api.StreamContext, item any) []any {
 		}
 		switch r := result.(type) {
 		case map[string]interface{}:
-			return []any{o.toTuple(r, d)}
+			return []any{toTupleFromRawTuple(r, d)}
 		case []map[string]interface{}:
 			rr := make([]any, len(r))
 			for i, v := range r {
-				rr[i] = o.toTuple(v, d)
+				rr[i] = toTupleFromRawTuple(v, d)
 			}
 			return rr
 		case []interface{}:
 			rr := make([]any, len(r))
 			for i, v := range r {
 				if vc, ok := v.(map[string]interface{}); ok {
-					rr[i] = o.toTuple(vc, d)
+					rr[i] = toTupleFromRawTuple(vc, d)
 				} else {
 					rr[i] = fmt.Errorf("only map[string]any inside a list is supported but got: %v", v)
 				}
@@ -149,7 +200,182 @@ func (o *DecodeOp) Worker(ctx api.StreamContext, item any) []any {
 	}
 }
 
-func (o *DecodeOp) toTuple(v map[string]any, d *xsql.RawTuple) *xsql.Tuple {
+// PayloadDecodeWorker each input has one message with the payload field to decode
+//
+//	{
+//		"payloadField":"data","otherField":1
+//	}
+//
+//	{
+//		// parsed fields
+//		"parsedField": 1,
+//		"parsedField2": 2,
+//		// keep the original field if in schema
+//		"payloadField":"data",
+//		"otherField":1
+//	}
+//
+// If parse result is a list, it will also output a list
+func (o *DecodeOp) PayloadDecodeWorker(ctx api.StreamContext, item any) []any {
+	o.statManager.ProcessTimeStart()
+	defer o.statManager.ProcessTimeEnd()
+	switch d := item.(type) {
+	case *xsql.Tuple:
+		// extract payload
+		payload, ok := d.Value(o.c.PayloadField, "")
+		if !ok {
+			ctx.GetLogger().Warnf("payload field %s not found, ignore it", o.c.PayloadField)
+			return nil
+		}
+		delete(d.Message, o.c.PayloadField)
+		raw, err := cast.ToByteA(payload, cast.CONVERT_SAMEKIND)
+		if err != nil {
+			return []any{fmt.Errorf("payload is not bytes: %v", err)}
+		}
+		result, err := o.converter.Decode(ctx, raw)
+		if err != nil {
+			return []any{err}
+		}
+		return transTuple(d, result)
+	default:
+		return []any{fmt.Errorf("unsupported data received: %v", d)}
+	}
+}
+
+// TODO do not update the tuple directly
+// Currently, this op is generated implicitly and it is guarantee to not share the data, so we mutate it directly
+func transTuple(d *xsql.Tuple, result any) []any {
+	switch r := result.(type) {
+	case map[string]any:
+		tupleAppend(d, r)
+		return []any{d}
+	case []map[string]any:
+		rr := make([]any, len(r))
+		for i, v := range r {
+			dd := cloneTuple(d)
+			tupleAppend(dd, v)
+			rr[i] = dd
+		}
+		return rr
+	case []any:
+		rr := make([]any, len(r))
+		for i, v := range r {
+			if vc, ok := v.(map[string]any); ok {
+				dd := cloneTuple(d)
+				tupleAppend(dd, vc)
+				rr[i] = dd
+			} else {
+				rr[i] = fmt.Errorf("only map[string]any inside a list is supported but got: %v", v)
+			}
+		}
+		return rr
+	default:
+		return []any{fmt.Errorf("unsupported decode result: %v", r)}
+	}
+}
+
+// PayloadBatchDecodeWorker deals with payload like
+//
+//	{
+//		"ts": 123456,
+//		"batchField": [
+//			{"payloadField":"data","otherField":1},
+//			{"payloadField":"data2","otherField":2}
+//		]
+//	}
+//
+// It merges all payload result into one
+//
+//	{
+//		"ts": 123456,
+//		// parsed fields are merged
+//		"parsedField": 1,
+//		"parsedField": 2,
+//		// other fields also merged and keep the latest
+//		"otherField": 2
+//	}
+//
+// If parse result is a list, it will also merge them in
+func (o *DecodeOp) PayloadBatchDecodeWorker(ctx api.StreamContext, item any) []any {
+	o.statManager.ProcessTimeStart()
+	defer o.statManager.ProcessTimeEnd()
+	switch d := item.(type) {
+	case *xsql.Tuple:
+		// extract batch field
+		batch, ok := d.Value(o.c.PayloadBatchField, "")
+		if !ok {
+			ctx.GetLogger().Warnf("payload batch field %s not found, ignore it", o.c.PayloadBatchField)
+			return nil
+		}
+		delete(d.Message, o.c.PayloadBatchField)
+		batchVal, ok := batch.([]any)
+		if !ok {
+			return []any{fmt.Errorf("payload batch field is not array: %v", batch)}
+		}
+		r := cloneTuple(d)
+		for _, val := range batchVal {
+			var vv xsql.Valuer
+			switch vt := val.(type) {
+			case xsql.Valuer:
+				vv = vt
+			case map[string]any:
+				vv = xsql.Message(vt)
+			}
+			payload, ok := vv.Value(o.c.PayloadField, "")
+			if !ok {
+				ctx.GetLogger().Warnf("payload field %s not found, ignore it", o.c.PayloadField)
+				continue
+			}
+			raw, err := cast.ToByteA(payload, cast.CONVERT_SAMEKIND)
+			if err != nil {
+				ctx.GetLogger().Warnf("payload is not bytes: %v", err)
+				continue
+			}
+			result, err := o.converter.Decode(ctx, raw)
+			if err != nil {
+				ctx.GetLogger().Warnf("cannot decode payload: %v", err)
+				continue
+			}
+			delete(val.(map[string]any), o.c.PayloadField)
+			mergeTuple(ctx, r, val)
+			mergeTuple(ctx, r, result)
+		}
+		return []any{r}
+	default:
+		return []any{fmt.Errorf("unsupported data received: %v", d)}
+	}
+}
+
+// TODO do not update the tuple directly
+// Currently, this op is generated implicitly and it is guarantee to not share the data, so we mutate it directly
+func mergeTuple(ctx api.StreamContext, d *xsql.Tuple, result any) {
+	switch r := result.(type) {
+	case map[string]any:
+		for k, v := range r {
+			d.Message[k] = v
+		}
+	case []map[string]interface{}:
+		for _, m := range r {
+			for k, v := range m {
+				d.Message[k] = v
+			}
+		}
+	case []any:
+		for _, a := range r {
+			if m, ok := a.(map[string]any); ok {
+				for k, v := range m {
+					d.Message[k] = v
+				}
+			} else {
+				ctx.GetLogger().Warnf("decode payload list receive non map %v", a)
+			}
+		}
+	default:
+		ctx.GetLogger().Warnf("unsupported decode result: %v", r)
+	}
+}
+
+func toTupleFromRawTuple(v map[string]any, d *xsql.RawTuple) *xsql.Tuple {
 	return &xsql.Tuple{
 		Message:   v,
 		Metadata:  d.Metadata,
@@ -158,4 +384,24 @@ func (o *DecodeOp) toTuple(v map[string]any, d *xsql.RawTuple) *xsql.Tuple {
 	}
 }
 
+func cloneTuple(d *xsql.Tuple) *xsql.Tuple {
+	m := make(map[string]any, len(d.Message))
+	for k, v := range d.Message {
+		m[k] = v
+	}
+	return &xsql.Tuple{
+		Message:   m,
+		Metadata:  d.Metadata,
+		Timestamp: d.Timestamp,
+		Emitter:   d.Emitter,
+	}
+}
+
+func tupleAppend(d *xsql.Tuple, mv map[string]any) {
+	m := d.Message
+	for k, v := range mv {
+		m[k] = v
+	}
+}
+
 var _ SchemaNode = &DecodeOp{}
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index 8f28f85a..6834f94d 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -74,6 +74,7 @@ func transformSourceNode(ctx api.StreamContext, t *DataSourcePlan, mockSourcesPr
 type SourcePropsForSplit struct {
 	Decompression string `json:"decompression"`
 	SelId         string `json:"connectionSelector"`
+	PayloadFormat string `json:"payloadFormat"`
 }
 
 func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, options *def.RuleOption, index int, ruleId string, pp node.UnOperation) (node.DataSourceNode, []node.OperatorNode, int, error) {
@@ -131,7 +132,7 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 
 	if info.NeedDecode {
 		// Create the decode node
-		decodeNode, err := node.NewDecodeOp(ctx, fmt.Sprintf("%d_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
+		decodeNode, err := node.NewDecodeOp(ctx, false, fmt.Sprintf("%d_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
 		if err != nil {
 			return nil, nil, 0, err
 		}
@@ -139,6 +140,16 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 		ops = append(ops, decodeNode)
 	}
 
+	if sp.PayloadFormat != "" {
+		// Create the decode node
+		payloadDecodeNode, err := node.NewDecodeOp(ctx, true, fmt.Sprintf("%d_payload_decoder", index), string(t.streamStmt.Name), ruleId, options, t.streamStmt.Options, t.isWildCard, t.isSchemaless, t.streamFields, props)
+		if err != nil {
+			return nil, nil, 0, err
+		}
+		index++
+		ops = append(ops, payloadDecodeNode)
+	}
+
 	// Create the preprocessor node if needed
 	if pp != nil {
 		ops = append(ops, Transform(pp, fmt.Sprintf("%d_preprocessor", index), options))
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index dd4f78d7..c4ab599d 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -400,9 +400,12 @@ func (s *Topo) ResetStreamOffset(name string, input map[string]interface{}) erro
 }
 
 func (s *Topo) WaitClose() {
+	if s == nil {
+		return
+	}
 	// wait all operators close
 	if s.opsWg != nil {
 		s.opsWg.Wait()
+		conf.Log.Infof("rule %s stopped", s.ctx.GetRuleId())
 	}
-	conf.Log.Infof("rule %s stopped", s.ctx.GetRuleId())
 }
diff --git a/pkg/cast/cast.go b/pkg/cast/cast.go
index 50a8c3b7..2a378aa4 100644
--- a/pkg/cast/cast.go
+++ b/pkg/cast/cast.go
@@ -831,14 +831,18 @@ func ToBytes(input interface{}, sn Strictness) ([]byte, error) {
 }
 
 // ToByteA converts to eKuiper internal byte array
-func ToByteA(input interface{}, _ Strictness) ([]byte, error) {
+func ToByteA(input interface{}, sn Strictness) ([]byte, error) {
 	switch b := input.(type) {
 	case []byte:
 		return b, nil
 	case string:
 		r, err := base64.StdEncoding.DecodeString(b)
 		if err != nil {
-			return nil, fmt.Errorf("illegal string %s, must be base64 encoded string", b)
+			if sn != STRICT {
+				r = []byte(b)
+			} else {
+				return nil, fmt.Errorf("illegal string %s, must be base64 encoded string", b)
+			}
 		}
 		return r, nil
 	}


Test patch:
diff --git a/internal/conf/yaml_config_ops_test.go b/internal/conf/yaml_config_ops_test.go
index 4a658401..5bffbcca 100644
--- a/internal/conf/yaml_config_ops_test.go
+++ b/internal/conf/yaml_config_ops_test.go
@@ -23,6 +23,7 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/pingcap/failpoint"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
@@ -293,14 +294,6 @@ func TestNewConfigOperatorForConnection(t *testing.T) {
 
 func TestConfigKeys_LoadFromKV(t *testing.T) {
 	InitConf()
-	Config.Basic.CfgStorageType = ""
-	InitConf()
-	// assert default
-	require.Equal(t, Config.Basic.CfgStorageType, cfgFileStorage)
-	defer func() {
-		Config.Basic.CfgStorageType = cfgFileStorage
-	}()
-	Config.Basic.CfgStorageType = cfgStoreKVStorage
 	mqttCfg, err := NewConfigOperatorFromSourceStorage("mqtt")
 	require.NoError(t, err)
 	require.NoError(t, mqttCfg.AddConfKey("key1", map[string]interface{}{
@@ -351,6 +344,15 @@ func TestConfigKeys_LoadFromKV(t *testing.T) {
 	mConn.storageType = "mock"
 	err = mConn.SaveCfgToStorage()
 	require.Error(t, err, fmt.Errorf("unknown source cfg storage type: %v", "mock"))
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/conf/storageErr", "return(true)")
+	_, err = NewConfigOperatorFromSourceStorage("mqtt")
+	require.Error(t, err)
+	_, err = NewConfigOperatorFromSinkStorage("mqtt")
+	require.Error(t, err)
+	_, err = NewConfigOperatorFromConnectionStorage("mqtt")
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/conf/storageErr")
 }
 
 func marshalUn(input, output interface{}) error {
diff --git a/internal/conf/yaml_config_storage_test.go b/internal/conf/yaml_config_storage_test.go
index 87bb6aea..f25fff9f 100644
--- a/internal/conf/yaml_config_storage_test.go
+++ b/internal/conf/yaml_config_storage_test.go
@@ -17,6 +17,7 @@ package conf
 import (
 	"testing"
 
+	"github.com/pingcap/failpoint"
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
@@ -53,3 +54,41 @@ func TestSQLiteStorage(t *testing.T) {
 		},
 	}, v)
 }
+
+func TestGetYamlConfigAllKeys(t *testing.T) {
+	dataDir, err := GetDataLoc()
+	require.NoError(t, err)
+	require.NoError(t, store.SetupDefault(dataDir))
+	require.NoError(t, saveCfgKeyToKV(buildKey("sources", "mqtt", "conf1"), map[string]interface{}{"a": 1}))
+	require.NoError(t, saveCfgKeyToKV(buildKey("sources", "sql", "conf1"), map[string]interface{}{"a": 1}))
+	m, err := GetYamlConfigAllKeys("sources")
+	require.NoError(t, err)
+	require.Equal(t, map[string]struct{}{
+		"mqtt": {}, "sql": {},
+	}, m)
+	require.NoError(t, err)
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/conf/storageErr", "return(true)")
+	_, err = GetYamlConfigAllKeys("sources")
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/conf/storageErr")
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/conf/getDataErr", "return(true)")
+	_, err = GetYamlConfigAllKeys("sources")
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/conf/getDataErr")
+}
+
+func TestGetStorage(t *testing.T) {
+	IsTesting = false
+	defer func() {
+		IsTesting = true
+	}()
+	dataDir, err := GetDataLoc()
+	require.NoError(t, err)
+	require.NoError(t, store.SetupDefault(dataDir))
+	kvStore = nil
+	s, err := getKVStorage()
+	require.NoError(t, err)
+	require.NotNil(t, s)
+}
diff --git a/internal/converter/json/convert_bench_test.go b/internal/converter/json/convert_bench_test.go
index 934717bd..a2bd30b9 100644
--- a/internal/converter/json/convert_bench_test.go
+++ b/internal/converter/json/convert_bench_test.go
@@ -106,16 +106,9 @@ func benchmarkByFiles(filePath string, b *testing.B, schema map[string]*ast.Json
 	if err != nil {
 		b.Fatalf(err.Error())
 	}
-	if schema != nil {
-		f := NewFastJsonConverter(schema)
-		b.ResetTimer()
-		for i := 0; i < b.N; i++ {
-			f.Decode(ctx, payload)
-		}
-	} else {
-		b.ResetTimer()
-		for i := 0; i < b.N; i++ {
-			converter.Decode(ctx, payload)
-		}
+	f := NewFastJsonConverter(schema)
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		f.Decode(ctx, payload)
 	}
 }
diff --git a/internal/converter/json/converter_test.go b/internal/converter/json/converter_test.go
index 5964d375..6124111d 100644
--- a/internal/converter/json/converter_test.go
+++ b/internal/converter/json/converter_test.go
@@ -18,70 +18,14 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
-	"os"
-	"path"
-	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
-	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
-func TestMessageDecode(t *testing.T) {
-	image, err := os.ReadFile(path.Join("../../../docs", "cover.jpg"))
-	if err != nil {
-		t.Errorf("Cannot read image: %v", err)
-	}
-	b64img := base64.StdEncoding.EncodeToString(image)
-	tests := []struct {
-		payload []byte
-		format  string
-		result  map[string]interface{}
-		results []interface{}
-	}{
-		{
-			payload: []byte(fmt.Sprintf(`{"format":"jpg","content":"%s"}`, b64img)),
-			format:  "json",
-			result: map[string]interface{}{
-				"format":  "jpg",
-				"content": b64img,
-			},
-		},
-		{
-			payload: []byte(`[{"a":1},{"a":2}]`),
-			format:  "json",
-			results: []interface{}{
-				map[string]interface{}{
-					"a": float64(1),
-				},
-				map[string]interface{}{
-					"a": float64(2),
-				},
-			},
-		},
-	}
-	conv, _ := GetConverter()
-	ctx := mockContext.NewMockContext("test", "op1")
-	for i, tt := range tests {
-		result, err := conv.Decode(ctx, tt.payload)
-		if err != nil {
-			t.Errorf("%d decode error: %v", i, err)
-		}
-		if len(tt.results) > 0 {
-			if !reflect.DeepEqual(tt.results, result) {
-				t.Errorf("%d result mismatch:\n\nexp=%s\n\ngot=%s\n\n", i, tt.result, result)
-			}
-		} else {
-			if !reflect.DeepEqual(tt.result, result) {
-				t.Errorf("%d result mismatch:\n\nexp=%s\n\ngot=%s\n\n", i, tt.result, result)
-			}
-		}
-	}
-}
-
 func TestFastJsonConverterWithSchema(t *testing.T) {
 	origin := "123"
 	encode := base64.StdEncoding.EncodeToString([]byte(origin))
@@ -713,19 +657,3 @@ func TestSchemaless(t *testing.T) {
 		require.Equal(t, tc.expect, v)
 	}
 }
-
-func TestJsonError(t *testing.T) {
-	ctx := mockContext.NewMockContext("test", "op1")
-	_, err := converter.Decode(ctx, nil)
-	require.Error(t, err)
-	errWithCode, ok := err.(errorx.ErrorWithCode)
-	require.True(t, ok)
-	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
-	// fastjson
-	c := NewFastJsonConverter(nil)
-	_, err = c.Decode(ctx, nil)
-	require.Error(t, err)
-	errWithCode, ok = err.(errorx.ErrorWithCode)
-	require.True(t, ok)
-	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
-}
diff --git a/internal/meta/sinkMeta_test.go b/internal/meta/sinkMeta_test.go
index 14f512d8..f8892b1b 100644
--- a/internal/meta/sinkMeta_test.go
+++ b/internal/meta/sinkMeta_test.go
@@ -19,9 +19,11 @@ import (
 	"path"
 	"testing"
 
+	"github.com/pingcap/failpoint"
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
 )
 
@@ -51,3 +53,21 @@ func TestMetaError(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, errorx.ConfKeyError, ewc.Code())
 }
+
+func TestReadMetaData(t *testing.T) {
+	dataDir, err := conf.GetDataLoc()
+	require.NoError(t, err)
+	require.NoError(t, store.SetupDefault(dataDir))
+	require.NoError(t, conf.SaveCfgKeyToKVInTest("sources.mqtt.conf1", map[string]interface{}{"a": 1}))
+	require.NoError(t, conf.SaveCfgKeyToKVInTest("sinks.mqtt.conf1", map[string]interface{}{"a": 1}))
+	require.NoError(t, conf.SaveCfgKeyToKVInTest("connections.mqtt.conf1", map[string]interface{}{"a": 1}))
+	require.NoError(t, ReadSourceMetaData())
+	require.NoError(t, ReadSinkMetaData())
+
+	failpoint.Enable("github.com/lf-edge/ekuiper/v2/internal/conf/storageErr", "return(true)")
+	err = ReadSourceMetaData()
+	require.Error(t, err)
+	err = ReadSinkMetaData()
+	require.Error(t, err)
+	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/conf/storageErr")
+}
diff --git a/internal/topo/node/decode_op_test.go b/internal/topo/node/decode_op_test.go
index dbb9d048..18714f7b 100644
--- a/internal/topo/node/decode_op_test.go
+++ b/internal/topo/node/decode_op_test.go
@@ -16,6 +16,7 @@ package node
 
 import (
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 
@@ -40,7 +41,7 @@ func TestJSON(t *testing.T) {
 	ctx := mockContext.NewMockContext("test", "Test")
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			op, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{
+			op, err := NewDecodeOp(ctx, false, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{
 				"sendInterval": "10ms",
 			})
 			assert.NoError(t, err)
@@ -94,7 +95,7 @@ func TestJSON(t *testing.T) {
 // This is useful when a node is much slower
 func BenchmarkThrougput(b *testing.B) {
 	ctx := mockContext.NewMockContext("test1", "decode_test")
-	op, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: 10, Debug: true}, &ast.Options{FORMAT: "mock"}, false, true, nil, nil)
+	op, err := NewDecodeOp(ctx, false, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: 10, Debug: true}, &ast.Options{FORMAT: "mock"}, false, true, nil, nil)
 	assert.NoError(b, err)
 	out := make(chan any, 100)
 	err = op.AddOutput(out, "test")
@@ -131,16 +132,18 @@ func TestJSONWithSchema(t *testing.T) {
 	ctx := mockContext.NewMockContext("test1", "decode_test")
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			op, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json", SHARED: true}, false, false, map[string]*ast.JsonStreamField{
+			op, err := NewDecodeOp(ctx, false, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true, Concurrency: tt.concurrency}, &ast.Options{FORMAT: "json", SHARED: true}, false, false, map[string]*ast.JsonStreamField{
 				"a": {
 					Type: "bigint",
 				},
-			}, nil)
+			}, map[string]any{
+				"payloadField": "c", "payloadFormat": "json",
+			})
+			// payload field will add to schema automatically
 			assert.NoError(t, err)
 			out := make(chan any, 100)
 			err = op.AddOutput(out, "test")
 			assert.NoError(t, err)
-			ctx := mockContext.NewMockContext("test1", "decode_test")
 			errCh := make(chan error)
 			op.Exec(ctx, errCh)
 
@@ -152,7 +155,7 @@ func TestJSONWithSchema(t *testing.T) {
 				{&xsql.Tuple{Emitter: "test", Message: map[string]interface{}{"a": int64(1)}, Timestamp: time.UnixMilli(111), Metadata: map[string]any{"topic": "demo", "qos": 1}}},
 				{
 					&xsql.Tuple{Emitter: "test", Message: map[string]interface{}{"a": int64(1)}, Timestamp: time.UnixMilli(111), Metadata: map[string]any{"topic": "demo", "qos": 1}},
-					&xsql.Tuple{Emitter: "test", Message: map[string]interface{}{"a": int64(3)}, Timestamp: time.UnixMilli(111), Metadata: map[string]any{"topic": "demo", "qos": 1}},
+					&xsql.Tuple{Emitter: "test", Message: map[string]interface{}{"a": int64(3), "c": "hello"}, Timestamp: time.UnixMilli(111), Metadata: map[string]any{"topic": "demo", "qos": 1}},
 				},
 			}
 
@@ -179,7 +182,7 @@ func TestJSONWithSchema(t *testing.T) {
 				{&xsql.Tuple{Emitter: "test", Message: map[string]interface{}{"a": int64(1), "b": 2.0}, Timestamp: time.UnixMilli(111), Metadata: map[string]any{"topic": "demo", "qos": 1}}},
 				{
 					&xsql.Tuple{Emitter: "test", Message: map[string]interface{}{"a": int64(1), "b": 2.0}, Timestamp: time.UnixMilli(111), Metadata: map[string]any{"topic": "demo", "qos": 1}},
-					&xsql.Tuple{Emitter: "test", Message: map[string]interface{}{"a": int64(3), "b": 4.0}, Timestamp: time.UnixMilli(111), Metadata: map[string]any{"topic": "demo", "qos": 1}},
+					&xsql.Tuple{Emitter: "test", Message: map[string]interface{}{"a": int64(3), "b": 4.0, "c": "hello"}, Timestamp: time.UnixMilli(111), Metadata: map[string]any{"topic": "demo", "qos": 1}},
 				},
 			}
 
@@ -226,13 +229,298 @@ func TestJSONWithSchema(t *testing.T) {
 
 func TestValidate(t *testing.T) {
 	ctx := mockContext.NewMockContext("test1", "decode_test")
-	_, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "cann"}, false, true, nil, nil)
+	_, err := NewDecodeOp(ctx, false, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "cann"}, false, true, nil, nil)
 	assert.Error(t, err)
 	assert.Equal(t, "cannot get converter from format cann, schemaId : format type cann not supported", err.Error())
-	_, err = NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "none"})
+	_, err = NewDecodeOp(ctx, false, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "none"})
 	assert.Error(t, err)
 	assert.EqualError(t, err, "1 error(s) decoding:\n\n* error decoding 'sendInterval': time: invalid duration \"none\"")
-	do, err := NewDecodeOp(ctx, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "12s"})
+	do, err := NewDecodeOp(ctx, false, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"sendInterval": "12s"})
+	assert.NoError(t, err)
+	assert.Equal(t, 12*time.Second, do.c.SendInterval)
+	_, err = NewDecodeOp(ctx, true, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"payloadField": "abc"})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "payloadFormat is missing")
+	_, err = NewDecodeOp(ctx, true, "test", "streamName", "test1", &def.RuleOption{BufferLength: 10, SendError: true}, &ast.Options{FORMAT: "json"}, false, true, nil, map[string]any{"payloadField": "abc", "payloadFormat": "test"})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "cannot get converter from format test, schemaId : format type test not supported")
+}
+
+func TestPayloadDecodeWithSchema(t *testing.T) {
+	tests := []struct {
+		name   string
+		input  any
+		schema map[string]*ast.JsonStreamField
+		result []any
+	}{
+		{
+			name: "normal",
+			input: &xsql.Tuple{
+				Emitter:  "test",
+				Message:  map[string]any{"payload": []byte(`{"a":23,"b":34}`)},
+				Metadata: map[string]any{"topic": "a"},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+			},
+			result: []any{&xsql.Tuple{
+				Emitter:  "test",
+				Metadata: map[string]any{"topic": "a"},
+				Message:  map[string]any{"b": 34.0},
+			}},
+		},
+		{
+			name: "list with one payload field not found",
+			input: &xsql.Tuple{
+				Emitter:  "test",
+				Message:  map[string]any{"n": "outside", "payload": []byte(`[{"a":23,"b":34},{"a":99},{"a":55,"b":66}]`)},
+				Metadata: map[string]any{"topic": "a"},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+				"n": nil,
+			},
+			result: []any{
+				&xsql.Tuple{
+					Emitter:  "test",
+					Metadata: map[string]any{"topic": "a"},
+					Message:  map[string]any{"b": 34.0, "n": "outside"},
+				},
+				&xsql.Tuple{
+					Emitter:  "test",
+					Metadata: map[string]any{"topic": "a"},
+					Message:  map[string]any{"n": "outside"},
+				},
+				&xsql.Tuple{
+					Emitter:  "test",
+					Metadata: map[string]any{"topic": "a"},
+					Message:  map[string]any{"b": 66.0, "n": "outside"},
+				},
+			},
+		},
+		{
+			name: "no payload field",
+			input: &xsql.Tuple{
+				Emitter:  "test",
+				Message:  map[string]any{"n": "outside"},
+				Metadata: map[string]any{"topic": "a"},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+			},
+			result: []any{},
+		},
+		{
+			name: "wrong payload field",
+			input: &xsql.Tuple{
+				Emitter:  "test",
+				Message:  map[string]any{"n": "outside", "payload": 34},
+				Metadata: map[string]any{"topic": "a"},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+			},
+			result: []any{
+				errors.New("payload is not bytes: cannot convert int(34) to bytea"),
+			},
+		},
+		{
+			name: "wrong input type",
+			input: &xsql.RawTuple{
+				Emitter:  "test",
+				Metadata: map[string]any{"topic": "a"},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+			},
+			result: []any{
+				errors.New("unsupported data received: &{test 0001-01-01 00:00:00 +0000 UTC [] map[topic:a] map[]}"),
+			},
+		},
+	}
+	ctx, cancel := mockContext.NewMockContext("test1", "decode_test").WithCancel()
+	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{
+		"payloadField": "payload", "payloadFormat": "json",
+	})
+
+	assert.NoError(t, err)
+	out := make(chan any, 100)
+	err = op.AddOutput(out, "test")
 	assert.NoError(t, err)
-	assert.Equal(t, 12*time.Second, do.sendInterval)
+	errCh := make(chan error)
+	op.Exec(ctx, errCh)
+	for i, tt := range tests {
+		if i > 0 {
+			op.DetachSchema(ctx, fmt.Sprintf("stream%d", i))
+		}
+		op.AttachSchema(ctx, fmt.Sprintf("stream%d", i), tt.schema, false)
+		t.Run(tt.name, func(t *testing.T) {
+			op.input <- tt.input
+			for _, exp := range tt.result {
+				r := <-out
+				assert.Equal(t, exp, r)
+			}
+		})
+	}
+	cancel()
+}
+
+func TestPayloadBatchDecodeWithSchema(t *testing.T) {
+	tests := []struct {
+		name   string
+		input  any
+		schema map[string]*ast.JsonStreamField
+		result []any
+	}{
+		{
+			name: "normal",
+			input: &xsql.Tuple{
+				Emitter: "test",
+				Message: map[string]any{
+					"n": "outside", "frames": []any{
+						map[string]any{
+							"payload": []byte(`{"a":23,"b":34}`),
+							"inner":   123,
+						},
+						map[string]any{
+							"payload": []byte(`{"a":33,"b":44}`),
+							"inner":   123,
+						},
+					},
+				},
+				Metadata: map[string]any{"topic": "a"},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b":     nil,
+				"n":     nil,
+				"inner": nil,
+			},
+			result: []any{&xsql.Tuple{
+				Emitter:  "test",
+				Metadata: map[string]any{"topic": "a"},
+				Message:  map[string]any{"b": 44.0, "inner": 123, "n": "outside"},
+			}},
+		},
+		{
+			name: "list with one payload field not found",
+			input: &xsql.Tuple{
+				Emitter: "test",
+				Message: map[string]any{"frames": []any{
+					map[string]any{
+						"payload": []byte(`{"a":23,"b":34}`),
+						"inner":   123,
+					},
+					map[string]any{
+						"payload": []byte(`[{"a":23,"b":54},{"a":99},{"a":55,"b":66}]`),
+						"inner":   456,
+					},
+				}},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+				"inner": nil,
+			},
+			result: []any{
+				&xsql.Tuple{
+					Emitter: "test",
+					Message: map[string]any{"b": 66.0, "inner": 456},
+				},
+			},
+		},
+		{
+			name: "no batch payload field",
+			input: &xsql.Tuple{
+				Emitter:  "test",
+				Message:  map[string]any{"n": "outside"},
+				Metadata: map[string]any{"topic": "a"},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+			},
+			result: []any{},
+		},
+		{
+			name: "no payload field",
+			input: &xsql.Tuple{
+				Emitter: "test",
+				Message: map[string]any{
+					"n": "outside", "frames": []any{
+						map[string]any{
+							"payload": []byte(`{"a":23,"b":34}`),
+							"inner":   123,
+						},
+						map[string]any{
+							"inner2": 333,
+						},
+						map[string]any{
+							"payload": 444,
+							"inner":   243,
+						},
+					},
+				},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+			},
+			result: []any{&xsql.Tuple{
+				Emitter: "test",
+				Message: map[string]any{"b": 34.0, "inner": 123, "n": "outside"},
+			}},
+		},
+		{
+			name: "wrong input type",
+			input: &xsql.RawTuple{
+				Emitter:  "test",
+				Metadata: map[string]any{"topic": "a"},
+			},
+			schema: map[string]*ast.JsonStreamField{
+				"b": {
+					Type: "float",
+				},
+			},
+			result: []any{
+				errors.New("unsupported data received: &{test 0001-01-01 00:00:00 +0000 UTC [] map[topic:a] map[]}"),
+			},
+		},
+	}
+	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{
+		"payloadField": "payload", "payloadFormat": "json", "payloadBatchField": "frames",
+	})
+
+	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)
+	for i, tt := range tests {
+		if i > 0 {
+			op.DetachSchema(ctx, fmt.Sprintf("stream%d", i))
+		}
+		op.AttachSchema(ctx, fmt.Sprintf("stream%d", i), tt.schema, false)
+		t.Run(tt.name, func(t *testing.T) {
+			op.input <- tt.input
+			for _, exp := range tt.result {
+				r := <-out
+				assert.Equal(t, exp, r)
+			}
+		})
+	}
 }
diff --git a/internal/topo/planner/planner_source_test.go b/internal/topo/planner/planner_source_test.go
index a395006b..d00715b5 100644
--- a/internal/topo/planner/planner_source_test.go
+++ b/internal/topo/planner/planner_source_test.go
@@ -70,6 +70,8 @@ func TestPlanTopo(t *testing.T) {
 			conf: map[string]any{
 				"fileType":      "csv",
 				"decompression": "gzip",
+				"payloadFormat": "json",
+				"payloadField":  "col1",
 			},
 			p: "file",
 			k: "csv",
@@ -240,9 +242,12 @@ func TestPlanTopo(t *testing.T) {
 				Sources: []string{"source_fs2"},
 				Edges: map[string][]any{
 					"source_fs2": {
-						"op_2_project",
+						"op_2_payload_decoder",
 					},
-					"op_2_project": {
+					"op_2_payload_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
 						"op_logToMemory_0_0_transform",
 					},
 					"op_logToMemory_0_0_transform": {
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index 7c39f787..e5e55e5b 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -4785,11 +4785,11 @@ func TestTransformSourceNode(t *testing.T) {
 	props := nodeConf.GetSourceConf("mqtt", &ast.Options{TYPE: "mqtt", DATASOURCE: "topic1"})
 	srcNode, err := node.NewSourceNode(tp.GetContext(), "test", &mqtt.SourceConnector{}, props, &def.RuleOption{SendError: false})
 	assert.NoError(t, err)
-	decodeNode, err := node.NewDecodeOp(tp.GetContext(), "2_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
+	decodeNode, err := node.NewDecodeOp(tp.GetContext(), false, "2_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
 	assert.NoError(t, err)
 	decomNode, err := node.NewDecompressOp("2_decompressor", &def.RuleOption{SendError: false}, "gzip")
 	assert.NoError(t, err)
-	decodeNode2, err := node.NewDecodeOp(tp.GetContext(), "3_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
+	decodeNode2, err := node.NewDecodeOp(tp.GetContext(), false, "3_decoder", "test", "test", &def.RuleOption{SendError: false}, &ast.Options{TYPE: "mqtt"}, false, false, schema, nil)
 	assert.NoError(t, err)
 	props2 := nodeConf.GetSourceConf("mqtt", &ast.Options{TYPE: "mqtt", CONF_KEY: "testCom", DATASOURCE: "topic2"})
 	srcNode2, err := node.NewSourceNode(tp.GetContext(), "test", &mqtt.SourceConnector{}, props2, &def.RuleOption{SendError: false})
diff --git a/pkg/cast/cast_test.go b/pkg/cast/cast_test.go
index ff6dd89b..8dea6aa0 100644
--- a/pkg/cast/cast_test.go
+++ b/pkg/cast/cast_test.go
@@ -888,8 +888,8 @@ func TestToByteA(t *testing.T) {
 		err    string
 	}{
 		{
-			input: "foo",
-			err:   "illegal string foo, must be base64 encoded string",
+			input:  "foo",
+			output: []byte("foo"),
 		}, {
 			input:  []byte("foo"),
 			output: []byte("foo"),