Repository: lf-edge/ekuiper
Pull request #2937 (2024-06-18 09:49:26)

feat: support bump sql source conf
bump sql source configuration from :

```yaml
default:
  interval: 10000
  url: mysql://user:test@140.210.204.147/user?parseTime=true
  internalSqlQueryCfg:
    table: test
    limit: 1
    indexField: registerTime
    indexValue: "2022-04-21 10:23:55"
    indexFieldType: "DATETIME"
    dateTimeFormat: "YYYY-MM-dd HH:mm:ss"
  templateSqlQueryCfg:
    TemplateSql: "select * from table where entry_data > {{.entry_data}}"
    indexField: entry_data
    indexValue: "2022-04-13 06:22:32.233"
    indexFieldType: "DATETIME"
    dateTimeFormat: "YYYY-MM-dd HH:mm:ssSSS"
```

into

```yaml
default:
  interval: "10s"
  dburl: mysql://user:test@140.210.204.147/user?parseTime=true
  internalSqlQueryCfg:
    table: test
    limit: 1
    indexFields:
    - indexField: registerTime
      indexValue: "2022-04-21 10:23:55"
      indexFieldType: "DATETIME"
      dateTimeFormat: "YYYY-MM-dd HH:mm:ss"
  templateSqlQueryCfg:
    templateSql: "select * from table where entry_data > {{.entry_data}}"
    indexFields:
    - indexField: entry_data
      indexValue: "2022-04-13 06:22:32.233"
      indexFieldType: "DATETIME"
      dateTimeFormat: "YYYY-MM-dd HH:mm:ssSSS"
```

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2937?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 `22.72727%` with `51 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@9c9096e`). [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/2937?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/server/bump/bump3.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2937?src=pr&el=tree&filepath=internal%2Fserver%2Fbump%2Fbump3.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL2J1bXAvYnVtcDMuZ28=) | 22.22% | [45 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2937?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/server/bump/bump\_manager.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2937?src=pr&el=tree&filepath=internal%2Fserver%2Fbump%2Fbump_manager.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvc2VydmVyL2J1bXAvYnVtcF9tYW5hZ2VyLmdv) | 33.33% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2937?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    #2937   +/-   ##
============================================
  Coverage             ?   69.53%           
============================================
  Files                ?      302           
  Lines                ?    29233           
  Branches             ?        0           
============================================
  Hits                 ?    20325           
  Misses               ?     7152           
  Partials             ?     1756           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2937?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/conf/yaml_config_storage.go b/internal/conf/yaml_config_storage.go
index 4bd0d5a5..5dcdc3c6 100644
--- a/internal/conf/yaml_config_storage.go
+++ b/internal/conf/yaml_config_storage.go
@@ -228,3 +228,19 @@ func GetCfgFromKVStorage(typ string, plugin string, confKey string) (map[string]
 	}
 	return kvStorage.GetByPrefix(key)
 }
+
+// ClearKVStorage only used in unit test
+func ClearKVStorage() error {
+	kvStorage, err := getKVStorage()
+	if err != nil {
+		return err
+	}
+	km, err := kvStorage.GetByPrefix("")
+	if err != nil {
+		return err
+	}
+	for key := range km {
+		kvStorage.Delete(key)
+	}
+	return nil
+}
diff --git a/internal/server/bump/bump3.go b/internal/server/bump/bump3.go
new file mode 100644
index 00000000..f6309fb8
--- /dev/null
+++ b/internal/server/bump/bump3.go
@@ -0,0 +1,146 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bump
+
+import (
+	"encoding/json"
+	"time"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/store"
+)
+
+func bumpFrom2TO3() error {
+	return rewriteSQLSourceConfiguration()
+}
+
+func rewriteSQLSourceConfiguration() error {
+	keyProps, err := conf.GetCfgFromKVStorage("sources", "sql", "")
+	if err != nil {
+		return err
+	}
+	for key, props := range keyProps {
+		cfg := &OriginSqlSourceCfg{}
+		if err := cast.MapToStruct(props, cfg); err != nil || cfg == nil {
+			continue
+		}
+		data := rewriteCfg(cfg)
+		_, _, confKey, _ := extractKey(key)
+		conf.WriteCfgIntoKVStorage("sources", "sql", confKey, data)
+	}
+	return nil
+}
+
+func rewriteCfg(cfg *OriginSqlSourceCfg) map[string]interface{} {
+	m := make(map[string]interface{})
+	m["dburl"] = cfg.Url
+	m["interval"] = time.Duration(cfg.Interval).String()
+	if cfg.InternalSqlQueryCfg != nil {
+		icfg := cfg.InternalSqlQueryCfg
+		im := make(map[string]interface{})
+		im["table"] = icfg.Table
+		im["limit"] = icfg.Limit
+		f := extractIndexField(icfg)
+		if f != nil {
+			icfg.IndexFields = append(icfg.IndexFields, f)
+		}
+		if len(icfg.IndexFields) > 0 {
+			im["indexFields"] = icfg.IndexFields
+		}
+		m["internalSqlQueryCfg"] = im
+	}
+	if cfg.TemplateSqlQueryCfg != nil {
+		tcfg := cfg.TemplateSqlQueryCfg
+		tm := make(map[string]interface{})
+		tm["templateSql"] = tcfg.TemplateSQL
+		f := extractIndexField2(tcfg)
+		if f != nil {
+			tcfg.IndexFields = append(tcfg.IndexFields, f)
+		}
+		if len(tcfg.IndexFields) > 0 {
+			tm["indexFields"] = tcfg.IndexFields
+		}
+		m["templateSqlQueryCfg"] = tm
+	}
+	bs, _ := json.Marshal(m)
+	newm := make(map[string]interface{})
+	json.Unmarshal(bs, &newm)
+	return newm
+}
+
+func extractIndexField(icfg *OriginInternalSqlQueryCfg) *store.IndexField {
+	if len(icfg.IndexFieldName) < 1 {
+		return nil
+	}
+	f := &store.IndexField{
+		IndexFieldName: icfg.IndexFieldName,
+	}
+	if icfg.IndexFieldValue != nil {
+		f.IndexFieldValue = icfg.IndexFieldValue
+	}
+	if len(icfg.IndexFieldDataType) > 0 {
+		f.IndexFieldDataType = icfg.IndexFieldDataType
+	}
+	if len(icfg.IndexFieldDateTimeFormat) > 0 {
+		f.IndexFieldDateTimeFormat = icfg.IndexFieldDateTimeFormat
+	}
+	return f
+}
+
+func extractIndexField2(icfg *OriginTemplateSqlQueryCfg) *store.IndexField {
+	if len(icfg.IndexFieldName) < 1 {
+		return nil
+	}
+	f := &store.IndexField{
+		IndexFieldName: icfg.IndexFieldName,
+	}
+	if icfg.IndexFieldValue != nil {
+		f.IndexFieldValue = icfg.IndexFieldValue
+	}
+	if len(icfg.IndexFieldDataType) > 0 {
+		f.IndexFieldDataType = icfg.IndexFieldDataType
+	}
+	if len(icfg.IndexFieldDateTimeFormat) > 0 {
+		f.IndexFieldDateTimeFormat = icfg.IndexFieldDateTimeFormat
+	}
+	return f
+}
+
+type OriginSqlSourceCfg struct {
+	Url                 string                     `json:"url"`
+	Interval            cast.DurationConf          `json:"interval"`
+	InternalSqlQueryCfg *OriginInternalSqlQueryCfg `json:"internalSqlQueryCfg,omitempty"`
+	TemplateSqlQueryCfg *OriginTemplateSqlQueryCfg `json:"templateSqlQueryCfg,omitempty"`
+}
+
+type OriginInternalSqlQueryCfg struct {
+	Table                    string              `json:"table"`
+	Limit                    int                 `json:"limit"`
+	IndexFieldName           string              `json:"indexField"`
+	IndexFieldValue          interface{}         `json:"indexValue"`
+	IndexFieldDataType       string              `json:"indexFieldType"`
+	IndexFieldDateTimeFormat string              `json:"dateTimeFormat"`
+	IndexFields              []*store.IndexField `json:"indexFields"`
+}
+
+type OriginTemplateSqlQueryCfg struct {
+	TemplateSQL              string              `json:"templateSql"`
+	IndexFieldName           string              `json:"indexField"`
+	IndexFieldValue          interface{}         `json:"indexValue"`
+	IndexFieldDataType       string              `json:"indexFieldType"`
+	IndexFieldDateTimeFormat string              `json:"dateTimeFormat"`
+	IndexFields              []*store.IndexField `json:"indexFields"`
+}
diff --git a/internal/server/bump/bump_manager.go b/internal/server/bump/bump_manager.go
index a760aad5..b85ed3e0 100644
--- a/internal/server/bump/bump_manager.go
+++ b/internal/server/bump/bump_manager.go
@@ -30,7 +30,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/kv"
 )
 
-const currentVersion = 2
+const currentVersion = 3
 
 var GlobalBumpManager *BumpManager
 
@@ -81,6 +81,10 @@ func BumpToCurrentVersion(dataDir string) error {
 			if err := bumpFrom1TO2(); err != nil {
 				return err
 			}
+		case 3:
+			if err := bumpFrom2TO3(); err != nil {
+				return err
+			}
 		}
 	}
 	return nil


Test patch:
diff --git a/internal/server/bump/bump3_test.go b/internal/server/bump/bump3_test.go
new file mode 100644
index 00000000..beddaa64
--- /dev/null
+++ b/internal/server/bump/bump3_test.go
@@ -0,0 +1,202 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bump
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/store"
+)
+
+func TestRewriteSQLConf(t *testing.T) {
+	testcases := []struct {
+		oldCfg   *OriginSqlSourceCfg
+		expected *TargetSqlSourceCfg
+	}{
+		{
+			oldCfg: &OriginSqlSourceCfg{
+				Url:      "123",
+				Interval: cast.DurationConf(10 * time.Second),
+			},
+			expected: &TargetSqlSourceCfg{
+				DbUrl:    "123",
+				Interval: cast.DurationConf(10 * time.Second),
+			},
+		},
+		{
+			oldCfg: &OriginSqlSourceCfg{
+				Url:      "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				InternalSqlQueryCfg: &OriginInternalSqlQueryCfg{
+					Table:              "t",
+					Limit:              10,
+					IndexFieldName:     "a",
+					IndexFieldValue:    1,
+					IndexFieldDataType: "bigint",
+				},
+			},
+			expected: &TargetSqlSourceCfg{
+				DbUrl:    "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				InternalSqlQueryCfg: &TargetInternalSqlQueryCfg{
+					Table: "t",
+					Limit: 10,
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "a",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+			},
+		},
+		{
+			oldCfg: &OriginSqlSourceCfg{
+				Url:      "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				TemplateSqlQueryCfg: &OriginTemplateSqlQueryCfg{
+					TemplateSQL:        "select * from t",
+					IndexFieldName:     "a",
+					IndexFieldValue:    1,
+					IndexFieldDataType: "bigint",
+				},
+			},
+			expected: &TargetSqlSourceCfg{
+				DbUrl:    "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				TemplateSqlQueryCfg: &TargetTemplateSqlQueryCfg{
+					TemplateSQL: "select * from t",
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "a",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+			},
+		},
+		{
+			oldCfg: &OriginSqlSourceCfg{
+				Url:      "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				InternalSqlQueryCfg: &OriginInternalSqlQueryCfg{
+					Table:              "t",
+					Limit:              10,
+					IndexFieldName:     "a",
+					IndexFieldValue:    1,
+					IndexFieldDataType: "bigint",
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "b",
+							IndexFieldValue:    1,
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+				TemplateSqlQueryCfg: &OriginTemplateSqlQueryCfg{
+					TemplateSQL:        "select * from t",
+					IndexFieldName:     "a",
+					IndexFieldValue:    1,
+					IndexFieldDataType: "bigint",
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "b",
+							IndexFieldValue:    1,
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+			},
+			expected: &TargetSqlSourceCfg{
+				DbUrl:    "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				InternalSqlQueryCfg: &TargetInternalSqlQueryCfg{
+					Table: "t",
+					Limit: 10,
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "b",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+						{
+							IndexFieldName:     "a",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+				TemplateSqlQueryCfg: &TargetTemplateSqlQueryCfg{
+					TemplateSQL: "select * from t",
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "b",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+						{
+							IndexFieldName:     "a",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for index, tc := range testcases {
+		require.NoError(t, conf.ClearKVStorage())
+		originData, err := json.Marshal(tc.oldCfg)
+		require.NoError(t, err)
+		m := make(map[string]interface{})
+		require.NoError(t, json.Unmarshal(originData, &m))
+		cfgKey := fmt.Sprintf("%v", index)
+		require.NoError(t, conf.WriteCfgIntoKVStorage("sources", "sql", cfgKey, m))
+		require.NoError(t, rewriteSQLSourceConfiguration())
+		gotm, err := conf.GetCfgFromKVStorage("sources", "sql", cfgKey)
+		require.NoError(t, err)
+		tcfg := &TargetSqlSourceCfg{}
+		cast.MapToStruct(gotm[fmt.Sprintf("sources.sql.%s", cfgKey)], tcfg)
+		require.Equal(t, tc.expected, tcfg)
+	}
+}
+
+type TargetSqlSourceCfg struct {
+	DbUrl               string                     `json:"dburl"`
+	Interval            cast.DurationConf          `json:"interval"`
+	InternalSqlQueryCfg *TargetInternalSqlQueryCfg `json:"internalSqlQueryCfg,omitempty"`
+	TemplateSqlQueryCfg *TargetTemplateSqlQueryCfg `json:"templateSqlQueryCfg,omitempty"`
+}
+
+type TargetInternalSqlQueryCfg struct {
+	Table       string              `json:"table"`
+	Limit       int                 `json:"limit"`
+	IndexFields []*store.IndexField `json:"indexFields"`
+}
+
+type TargetTemplateSqlQueryCfg struct {
+	TemplateSQL string              `json:"templateSql"`
+	IndexFields []*store.IndexField `json:"indexFields"`
+}