Merge branch 'arana-db:master' into master

conf/config.yaml

@@ -29,6 +29,8 @@ data:
         password: "123456"
         # TODO It is recommended that the initialization of the subsequent system library be handled by arana internally
         database: __arana_sys
+        weight: r10w10
+        parameters:
       users:
         - username: root
           password: "123456"

integration_test/config/db/config.yaml

@@ -34,6 +34,8 @@ data:
         password: "123456"
         # TODO It is recommended that the initialization of the subsequent system library be handled by arana internally
         database: __arana_sys
+        weight: r10w10
+        parameters:
       clusters:
         - name: employees
           type: mysql

integration_test/config/db_tbl/config.yaml

@@ -34,6 +34,8 @@ data:
         password: "123456"
         # TODO It is recommended that the initialization of the subsequent system library be handled by arana internally
         database: __arana_sys
+        weight: r10w10
+        parameters:
       clusters:
         - name: employees
           type: mysql

integration_test/config/db_tbl_rw/config.yaml

@@ -34,6 +34,15 @@ data:
           password: "123456"
         - username: arana
           password: "123456"
+      sys_db:
+        host: arana-mysql
+        port: 3306
+        username: root
+        password: "123456"
+        # TODO It is recommended that the initialization of the subsequent system library be handled by arana internally
+        database: __arana_sys
+        weight: r10w10
+        parameters:
 
   clusters:
     - name: employees

integration_test/config/tbl/config.yaml

@@ -34,6 +34,8 @@ data:
         password: "123456"
         # TODO It is recommended that the initialization of the subsequent system library be handled by arana internally
         database: __arana_sys
+        weight: r10w10
+        parameters:
       clusters:
         - name: employees
           type: mysql

pkg/config/api.go

@@ -69,6 +69,7 @@ const (
 
 const (
 	ConfigItemSpec         = "spec"
+	ConfigItemSysDB        = "sys_db"
 	ConfigItemUsers        = "users"
 	ConfigItemClusters     = "clusters"
 	ConfigItemShardingRule = "sharding_rule"

pkg/config/config_reader.go

@@ -172,6 +172,12 @@ func (c *configReader) compositeConfiguration(loadFilter map[PathKey]string) *Te
 		}
 	}
 
+	if _, ok := loadFilter[c.pathInfo.DefaultConfigSysDBPath]; ok {
+		if val := c.holders[c.pathInfo.DefaultConfigSysDBPath].Load(); val != nil {
+			conf.SysDB = val.(*Tenant).SysDB
+		}
+	}
+
 	if _, ok := loadFilter[c.pathInfo.DefaultConfigDataNodesPath]; ok {
 		if val := c.holders[c.pathInfo.DefaultConfigDataNodesPath].Load(); val != nil {
 			conf.Nodes = val.(*Tenant).Nodes

pkg/config/config_test.go

@@ -299,6 +299,7 @@ func Test_configReader_LoadAll(t *testing.T) {
 		}
 
 		data, _ := yaml.Marshal(nodes)
+		sysdb, _ := yaml.Marshal(config.NewEmptyTenant().SysDB)
 		users, _ := yaml.Marshal(config.NewEmptyTenant().Users)
 		shadow, _ := yaml.Marshal(config.NewEmptyTenant().ShadowRule)
 		sharding, _ := yaml.Marshal(config.NewEmptyTenant().ShardingRule)
@@ -311,6 +312,9 @@ func Test_configReader_LoadAll(t *testing.T) {
 			Do(func(interface{}) {
 				atomic.AddInt32(&callCnt, 1)
 			})
+		mockStoreOperator.EXPECT().Get(config.NewPathInfo("arana").DefaultConfigSysDBPath).
+			AnyTimes().
+			Return(sysdb, nil)
 		mockStoreOperator.EXPECT().Get(config.NewPathInfo("arana").DefaultConfigDataUsersPath).
 			AnyTimes().
 			Return(users, nil)
@@ -357,6 +361,7 @@ func Test_configReader_LoadAll(t *testing.T) {
 		mockStoreOperator.EXPECT().Watch(gomock.Any()).AnyTimes().Return(shareCh, nil)
 
 		nodes, _ := yaml.Marshal(config.NewEmptyTenant().Nodes)
+		sysdb, _ := yaml.Marshal(config.NewEmptyTenant().SysDB)
 		users, _ := yaml.Marshal(config.NewEmptyTenant().Users)
 		shadow, _ := yaml.Marshal(config.NewEmptyTenant().ShadowRule)
 		sharding, _ := yaml.Marshal(config.NewEmptyTenant().ShardingRule)
@@ -366,6 +371,9 @@ func Test_configReader_LoadAll(t *testing.T) {
 		mockStoreOperator.EXPECT().Get(config.NewPathInfo("arana").DefaultConfigDataNodesPath).
 			AnyTimes().
 			Return(nodes, nil)
+		mockStoreOperator.EXPECT().Get(config.NewPathInfo("arana").DefaultConfigSysDBPath).
+			AnyTimes().
+			Return(sysdb, nil)
 		mockStoreOperator.EXPECT().Get(config.NewPathInfo("arana").DefaultConfigDataUsersPath).
 			AnyTimes().
 			Return(users, nil)

pkg/config/model.go

@@ -419,6 +419,7 @@ func NewEmptyTenant() *Tenant {
 		Spec: Spec{
 			Metadata: map[string]interface{}{},
 		},
+		SysDB:              nil,
 		Users:              make([]*User, 0, 1),
 		DataSourceClusters: make([]*DataSourceCluster, 0, 1),
 		ShardingRule:       new(ShardingRule),

pkg/config/path.go

@@ -28,6 +28,7 @@ import (
 type PathInfo struct {
 	DefaultConfigSpecPath               PathKey
 	DefaultTenantBaseConfigPath         PathKey
+	DefaultConfigSysDBPath              PathKey
 	DefaultConfigDataNodesPath          PathKey
 	DefaultConfigDataUsersPath          PathKey
 	DefaultConfigDataSourceClustersPath PathKey
@@ -45,6 +46,7 @@ func NewPathInfo(tenant string) *PathInfo {
 
 	p.DefaultTenantBaseConfigPath = PathKey(filepath.Join(string(DefaultRootPath), fmt.Sprintf("tenants/%s", tenant)))
 	p.DefaultConfigSpecPath = PathKey(filepath.Join(string(p.DefaultTenantBaseConfigPath), "spec"))
+	p.DefaultConfigSysDBPath = PathKey(filepath.Join(string(p.DefaultTenantBaseConfigPath), "sys_db"))
 	p.DefaultConfigDataNodesPath = PathKey(filepath.Join(string(p.DefaultTenantBaseConfigPath), "nodes"))
 	p.DefaultConfigDataUsersPath = PathKey(filepath.Join(string(p.DefaultTenantBaseConfigPath), "users"))
 	p.DefaultConfigDataSourceClustersPath = PathKey(filepath.Join(string(p.DefaultTenantBaseConfigPath), "dataSourceClusters"))
@@ -63,6 +65,9 @@ func NewPathInfo(tenant string) *PathInfo {
 		p.DefaultConfigSpecPath: func(cfg *Tenant) interface{} {
 			return &cfg.Spec
 		},
+		p.DefaultConfigSysDBPath: func(cfg *Tenant) interface{} {
+			return &cfg.SysDB
+		},
 		p.DefaultConfigDataUsersPath: func(cfg *Tenant) interface{} {
 			return &cfg.Users
 		},
@@ -82,6 +87,7 @@ func NewPathInfo(tenant string) *PathInfo {
 
 	p.ConfigKeyMapping = map[PathKey]string{
 		p.DefaultConfigSpecPath:               ConfigItemSpec,
+		p.DefaultConfigSysDBPath:              ConfigItemSysDB,
 		p.DefaultConfigDataUsersPath:          ConfigItemUsers,
 		p.DefaultConfigDataSourceClustersPath: ConfigItemClusters,
 		p.DefaultConfigDataShardingRulePath:   ConfigItemShardingRule,

pkg/runtime/function/cume_dist.go

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 function
+
+import (
+	"context"
+)
+
+import (
+	"github.com/pkg/errors"
+)
+
+import (
+	"github.com/arana-db/arana/pkg/proto"
+)
+
+// FuncCumeDist is  https://dev.mysql.com/doc/refman/8.0/en/window-function-descriptions.html
+const FuncCumeDist = "CUME_DIST"
+
+var _ proto.Func = (*cumedistFunc)(nil)
+
+func init() {
+	proto.RegisterFunc(FuncCumeDist, cumedistFunc{})
+}
+
+type cumedistFunc struct{}
+
+func (a cumedistFunc) Apply(ctx context.Context, inputs ...proto.Valuer) (proto.Value, error) {
+	first, err := inputs[0].Value(ctx)
+	if first == nil || err != nil {
+		return nil, errors.Wrapf(err, "cannot eval %s", FuncCumeDist)
+	}
+	firstDec, _ := first.Float64()
+	firstNum := 0
+
+	for _, it := range inputs[1:] {
+		val, err := it.Value(ctx)
+		if val == nil || err != nil {
+			return nil, errors.Wrapf(err, "cannot eval %s", FuncCumeDist)
+		}
+		valDec, _ := val.Float64()
+
+		if valDec <= firstDec {
+			firstNum++
+		}
+	}
+
+	r := 0.0
+	if len(inputs) > 1 {
+		r = float64(firstNum) / float64(len(inputs)-1)
+	}
+	return proto.NewValueFloat64(r), nil
+}
+
+func (a cumedistFunc) NumInput() int {
+	return 0
+}

pkg/runtime/function/cume_dist_test.go

@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 function
+
+import (
+	"context"
+	"fmt"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/arana-db/arana/pkg/proto"
+)
+
+func TestFuncCumeDist(t *testing.T) {
+	fn := proto.MustGetFunc(FuncCumeDist)
+	type tt struct {
+		inputs []proto.Value
+		want   string
+	}
+	for _, v := range []tt{
+		{
+			[]proto.Value{
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(2),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(5),
+			},
+			"0.2222222222222222",
+		},
+		{
+			[]proto.Value{
+				proto.NewValueFloat64(2),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(2),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(5),
+			},
+			"0.3333333333333333",
+		},
+		{
+			[]proto.Value{
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(2),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(5),
+			},
+			"0.6666666666666666",
+		},
+		{
+			[]proto.Value{
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(2),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(5),
+			},
+			"0.8888888888888888",
+		},
+		{
+			[]proto.Value{
+				proto.NewValueFloat64(5),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(1),
+				proto.NewValueFloat64(2),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(3),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(4),
+				proto.NewValueFloat64(5),
+			},
+			"1",
+		},
+	} {
+		t.Run(v.want, func(t *testing.T) {
+			var inputs []proto.Valuer
+			for i := range v.inputs {
+				inputs = append(inputs, proto.ToValuer(v.inputs[i]))
+			}
+			out, err := fn.Apply(context.Background(), inputs...)
+			assert.NoError(t, err)
+			assert.Equal(t, v.want, fmt.Sprint(out))
+		})
+	}
+}