opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-08-30 17:13:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: mapping mysql tables into clickhouse with Engine MySQL (#18877)

Browse Source 
Co-authored-by: Qiu Jian <qiujian@yunionyun.com>
This commit is contained in:
Jian Qiu
2023-12-05 07:08:17 +08:00
committed by GitHub
parent 0692c20341
commit 1b997dcb79
19 changed files with 373 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+2 -2
View File
@@ -87,14 +87,14 @@ require (
k8s.io/client-go v0.19.3
k8s.io/cluster-bootstrap v0.19.3
moul.io/http2curl/v2 v2.3.0
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231204065558-c4edc0938f2e
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231204130443-d54a8b16c7b7
yunion.io/x/executor v0.0.0-20230705125604-c5ac3141db32
yunion.io/x/jsonutils v1.0.1-0.20230613121553-0f3b41e2ef19
yunion.io/x/log v1.0.1-0.20230411060016-feb3f46ab361
yunion.io/x/ovsdb v0.0.0-20230306173834-f164f413a900
yunion.io/x/pkg v1.0.1-0.20231101105448-abef64cdc142
yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e
yunion.io/x/sqlchemy v1.1.2-0.20231201052514-97026b18ccf0
yunion.io/x/sqlchemy v1.1.2-0.20231204175132-1eb294922a51
yunion.io/x/structarg v0.0.0-20231017124457-df4d5009457c
)
go.sum
+4 -4
View File
@@ -1192,8 +1192,8 @@ sigs.k8s.io/structured-merge-diff/v4 v4.0.1/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK
sigs.k8s.io/yaml v1.1.0/go.mod h1:UJmg0vDUVViEyp3mgSv9WPwZCDxu4rQW1olrI1uml+o=
sigs.k8s.io/yaml v1.2.0 h1:kr/MCeFWJWTwyaHoR9c8EjH9OumOmoF9YGiZd7lFm/Q=
sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231204065558-c4edc0938f2e h1:QmYmnMJjVijGbLi5haZS49d2CaNgA/JLphLn7oL8q94=
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231204065558-c4edc0938f2e/go.mod h1:aj1gR9PPb6eqqKOwvANe26CoZFY8ydmXy0fuvgKYXH0=
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231204130443-d54a8b16c7b7 h1:fZ3sE1acOojrgXGtqLxXcfFhg4bT3shKx2o73J+Yat8=
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231204130443-d54a8b16c7b7/go.mod h1:aj1gR9PPb6eqqKOwvANe26CoZFY8ydmXy0fuvgKYXH0=
yunion.io/x/executor v0.0.0-20230705125604-c5ac3141db32 h1:v7POYkQwo1XzOxBoIoRVr/k0V9Y5JyjpshlIFa9raug=
yunion.io/x/executor v0.0.0-20230705125604-c5ac3141db32/go.mod h1:Uxuou9WQIeJXNpy7t2fPLL0BYLvLiMvGQwY7Qc6aSws=
yunion.io/x/jsonutils v0.0.0-20190625054549-a964e1e8a051/go.mod h1:4N0/RVzsYL3kH3WE/H1BjUQdFiWu50JGCFQuuy+Z634=
@@ -1211,7 +1211,7 @@ yunion.io/x/pkg v1.0.1-0.20231101105448-abef64cdc142 h1:L6LqxfP08eWUx+A6yQdrL6VB
yunion.io/x/pkg v1.0.1-0.20231101105448-abef64cdc142/go.mod h1:ksCJVQ+DwKrJ5QBEoU8pzrDFfDaZVAFH/iJ6yQCYxJk=
yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e h1:v+EzIadodSwkdZ/7bremd7J8J50Cise/HCylsOJngmo=
yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e/go.mod h1:0iFKpOs1y4lbCxeOmq3Xx/0AcQoewVPwj62eRluioEo=
yunion.io/x/sqlchemy v1.1.2-0.20231201052514-97026b18ccf0 h1:+MaykFV6YCakTLnHR3v31tovXhkvWgkBPFa83MuCekA=
yunion.io/x/sqlchemy v1.1.2-0.20231201052514-97026b18ccf0/go.mod h1:uuPVZEyEq3sWd5vf9VjGSy6lZzof22X87OEHw9sddJQ=
yunion.io/x/sqlchemy v1.1.2-0.20231204175132-1eb294922a51 h1:XgvhXxYul4W85vVJVKCKOWw6Ea/YroI3TBxyLcQ7vx4=
yunion.io/x/sqlchemy v1.1.2-0.20231204175132-1eb294922a51/go.mod h1:uuPVZEyEq3sWd5vf9VjGSy6lZzof22X87OEHw9sddJQ=
yunion.io/x/structarg v0.0.0-20231017124457-df4d5009457c h1:QuLab2kSRECZRxo4Lo2KcYn6XjQFDGaZ1+x0pYDVVwQ=
yunion.io/x/structarg v0.0.0-20231017124457-df4d5009457c/go.mod h1:EP6NSv2C0zzqBDTKumv8hPWLb3XvgMZDHQRfyuOrQng=
pkg/cloudcommon/consts/db.go
+17
View File
@@ -18,4 +18,21 @@ var (
QueryOffsetOptimization = false
OpsLogWithClickhouse = false
defaultDBDialect string
defaultDBConnectionString string
)
func SetDefaultDB(dialect, connStr string) {
defaultDBDialect = dialect
defaultDBConnectionString = connStr
}
func DefaultDBDialect() string {
return defaultDBDialect
}
func DefaultDBConnStr() string {
return defaultDBConnectionString
}
pkg/cloudcommon/database.go
+5 -2
View File
@@ -34,6 +34,7 @@ import (
"yunion.io/x/onecloud/pkg/cloudcommon/informer"
"yunion.io/x/onecloud/pkg/cloudcommon/notifyclient"
common_options "yunion.io/x/onecloud/pkg/cloudcommon/options"
"yunion.io/x/onecloud/pkg/util/dbutils"
)
func InitDB(options *common_options.DBOptions) {
@@ -77,6 +78,8 @@ func InitDB(options *common_options.DBOptions) {
log.Fatalf("cannot use clickhouse as primary database")
}
log.Infof("database dialect: %s sqlStr: %s", dialect, sqlStr)
// save configuration to consts
consts.SetDefaultDB(dialect, sqlStr)
dbConn, err := sql.Open(dialect, sqlStr)
if err != nil {
panic(err)
@@ -87,11 +90,11 @@ func InitDB(options *common_options.DBOptions) {
if err == nil {
// connect to clickcloud
// force convert sqlstr from clickhouse v2 to v1
sqlStr, err = clickhouseSqlStrV2ToV1(sqlStr)
sqlStr, err = dbutils.ClickhouseSqlStrV2ToV1(sqlStr)
if err != nil {
log.Fatalf("fail to convert clickhouse sqlstr from v2 to v1: %s", err)
}
err = validateClickhouseV1Str(sqlStr)
err = dbutils.ValidateClickhouseV1Str(sqlStr)
if err != nil {
log.Fatalf("invalid clickhouse sqlstr: %s", err)
}
pkg/cloudcommon/db/modelbase.go
+26
View File
@@ -28,12 +28,15 @@ import (
"yunion.io/x/pkg/util/rbacscope"
"yunion.io/x/pkg/util/version"
"yunion.io/x/sqlchemy"
"yunion.io/x/sqlchemy/backends/clickhouse"
"yunion.io/x/onecloud/pkg/apis"
"yunion.io/x/onecloud/pkg/appsrv"
"yunion.io/x/onecloud/pkg/cloudcommon/consts"
"yunion.io/x/onecloud/pkg/cloudcommon/policy"
"yunion.io/x/onecloud/pkg/httperrors"
"yunion.io/x/onecloud/pkg/mcclient"
"yunion.io/x/onecloud/pkg/util/dbutils"
"yunion.io/x/onecloud/pkg/util/logclient"
"yunion.io/x/onecloud/pkg/util/splitable"
"yunion.io/x/onecloud/pkg/util/stringutils2"
@@ -83,6 +86,29 @@ func NewModelBaseManagerWithSplitableDBName(model interface{}, tableName string,
return modelMan
}
func NewModelBaseManagerWithClickhouseMapping(manager IModelManager, tblname, keyword, keywordPlural string) SModelBaseManager {
ots := manager.TableSpec()
var extraOpts sqlchemy.TableExtraOptions
switch consts.DefaultDBDialect() {
case "mysql":
cfg := dbutils.ParseMySQLConnStr(consts.DefaultDBConnStr())
err := cfg.Validate()
if err != nil {
panic(fmt.Sprintf("invalid mysql connection string %s", consts.DefaultDBConnStr()))
}
extraOpts = clickhouse.MySQLExtraOptions(cfg.Hostport, cfg.Database, ots.Name(), cfg.Username, cfg.Password)
default:
panic(fmt.Sprintf("unsupport dialect %s to be backend of clickhouse", consts.DefaultDBDialect()))
}
nts := newClickhouseTableSpecFromMySQL(ots, tblname, ClickhouseDB, extraOpts)
modelMan := SModelBaseManager{
tableSpec: nts,
keyword: keyword,
keywordPlural: keywordPlural,
}
return modelMan
}
func (manager *SModelBaseManager) CreateByInsertOrUpdate() bool {
return true
}
pkg/cloudcommon/db/models.go
+2
View File
@@ -103,6 +103,7 @@ func mustCheckModelManager(modelMan IModelManager) {
func tableSpecId(tableSpec ITableSpec) string {
keys := []string{
string(tableSpec.GetDBName()),
tableSpec.Name(),
}
for _, c := range tableSpec.Columns() {
@@ -151,6 +152,7 @@ func CheckSync(autoSync bool, enableChecksumTables bool, skipInitChecksum bool)
tableSpec := modelMan.TableSpec()
tableKey := tableSpecId(tableSpec)
if _, ok := processedTbl[tableKey]; ok {
log.Warningf("table %s has been synced!", tableKey)
continue
}
processedTbl[tableKey] = true
pkg/cloudcommon/db/tablespec.go
+7
View File
@@ -82,6 +82,13 @@ func newTableSpec(model interface{}, tableName string, indexField string, dateFi
}
}
func newClickhouseTableSpecFromMySQL(spec ITableSpec, name string, dbName sqlchemy.DBName, extraOpts sqlchemy.TableExtraOptions) ITableSpec {
itbl := sqlchemy.NewTableSpecFromISpecWithDBName(spec.(*sTableSpec).ITableSpec, name, dbName, extraOpts)
return &sTableSpec{
ITableSpec: itbl,
}
}
func (ts *sTableSpec) GetSplitTable() *splitable.SSplitTableSpec {
sts, ok := ts.ITableSpec.(*splitable.SSplitTableSpec)
if ok {
pkg/cloudcommon/clickhouse.go → pkg/util/dbutils/clickhouse.go
+5 -5
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package cloudcommon
package dbutils
import (
"fmt"
@@ -27,7 +27,7 @@ import (
// convert clickhouse sqlstr v1 to v2
// v1: tcp://192.168.222.4:9000?database=yunionmeter&read_timeout=10&write_timeout=20
// v2: clickhouse://username:password@host1:9000,host2:9000/database?dial_timeout=200ms&max_execution_time=60
func clickhouseSqlStrV1ToV2(sqlstr string) (string, error) {
func ClickhouseSqlStrV1ToV2(sqlstr string) (string, error) {
if strings.HasPrefix(sqlstr, "clickhouse://") {
// already v2 format
return sqlstr, nil
@@ -57,7 +57,7 @@ func clickhouseSqlStrV1ToV2(sqlstr string) (string, error) {
return fmt.Sprintf("clickhouse://%s/%s?dial_timeout=200ms&max_execution_time=60", hostPart, dbname), nil
}
func clickhouseSqlStrV2ToV1(sqlstr string) (string, error) {
func ClickhouseSqlStrV2ToV1(sqlstr string) (string, error) {
if strings.HasPrefix(sqlstr, "tcp://") {
// already v1 format
return sqlstr, nil
@@ -89,7 +89,7 @@ func clickhouseSqlStrV2ToV1(sqlstr string) (string, error) {
return fmt.Sprintf("tcp://%s?%s&read_timeout=10&write_timeout=20", hostPart, jsonutils.Marshal(qs).QueryString()), nil
}
func validateClickhouseV2Str(sqlstr string) error {
func ValidateClickhouseV2Str(sqlstr string) error {
if !strings.HasPrefix(sqlstr, "clickhouse://") {
return errors.Wrapf(httperrors.ErrInputParameter, "must start with clickhouse://")
}
@@ -108,7 +108,7 @@ func validateClickhouseV2Str(sqlstr string) error {
return nil
}
func validateClickhouseV1Str(sqlstr string) error {
func ValidateClickhouseV1Str(sqlstr string) error {
if !strings.HasPrefix(sqlstr, "tcp://") {
return errors.Wrapf(httperrors.ErrInputParameter, "must start with tcp://")
}
pkg/cloudcommon/clickhouse_test.go → pkg/util/dbutils/clickhouse_test.go
+5 -5
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package cloudcommon
package dbutils
import (
"testing"
@@ -36,7 +36,7 @@ func TestClickhouseSqlStrV1ToV2(t *testing.T) {
want: "clickhouse://admin:pass@192.168.222.4:9000/yunionmeter?dial_timeout=200ms&max_execution_time=60",
},
} {
got, err := clickhouseSqlStrV1ToV2(c.in)
got, err := ClickhouseSqlStrV1ToV2(c.in)
if err != nil {
t.Errorf("%s", err)
} else if got != c.want {
@@ -67,7 +67,7 @@ func TestClickhouseSqlStrV2ToV1(t *testing.T) {
want: "tcp://192.168.222.4:9000?database=yunionmeter&password=pass&username=admin&read_timeout=10&write_timeout=20",
},
} {
got, err := clickhouseSqlStrV2ToV1(c.in)
got, err := ClickhouseSqlStrV2ToV1(c.in)
if err != nil {
t.Errorf("%s", err)
} else if got != c.want {
@@ -102,7 +102,7 @@ func TestValidateClickhouseSqlstrV1(t *testing.T) {
valid: false,
},
} {
err := validateClickhouseV1Str(c.in)
err := ValidateClickhouseV1Str(c.in)
if err != nil && c.valid {
t.Errorf("%s", err)
}
@@ -131,7 +131,7 @@ func TestValidateClickhouseSqlstrV2(t *testing.T) {
valid: true,
},
} {
err := validateClickhouseV2Str(c.in)
err := ValidateClickhouseV2Str(c.in)
if err != nil && c.valid {
t.Errorf("%s", err)
}
pkg/util/dbutils/doc.go
+15
View File
@@ -0,0 +1,15 @@
// Copyright 2019 Yunion
//
// 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 dbutils // import "yunion.io/x/onecloud/pkg/util/dbutils"
pkg/util/dbutils/mysql.go
+72
View File
@@ -0,0 +1,72 @@
// Copyright 2019 Yunion
//
// 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 dbutils
import (
"strings"
"yunion.io/x/pkg/errors"
)
type SDBConfig struct {
Hostport string
Database string
Username string
Password string
}
func (cfg SDBConfig) Validate() error {
errs := make([]error, 0)
if len(cfg.Hostport) == 0 {
errs = append(errs, errors.Error("empty host port"))
}
if len(cfg.Username) == 0 {
errs = append(errs, errors.Error("empty username"))
}
if len(cfg.Database) == 0 {
errs = append(errs, errors.Error("empty database"))
}
return errors.NewAggregate(errs)
}
func ParseMySQLConnStr(connStr string) SDBConfig {
// fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?%s", user, passwd, host, port, dburl, query.Encode())
cfg := SDBConfig{}
index := strings.Index(connStr, "@tcp(")
if index > 0 {
userpass := connStr[:index]
hostdb := connStr[index+len("@tcp("):]
index = strings.Index(userpass, ":")
if index > 0 {
cfg.Username = userpass[:index]
cfg.Password = userpass[index+1:]
} else if index < 0 {
cfg.Username = userpass
}
index = strings.Index(hostdb, ")/")
if index > 0 {
cfg.Hostport = hostdb[:index]
dbstr := hostdb[index+len(")/"):]
index = strings.Index(dbstr, "?")
if index > 0 {
cfg.Database = dbstr[:index]
} else if index < 0 {
cfg.Database = dbstr
}
}
}
return cfg
}
pkg/util/dbutils/mysql_test.go
+44
View File
@@ -0,0 +1,44 @@
// Copyright 2019 Yunion
//
// 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 dbutils
import (
"testing"
"yunion.io/x/jsonutils"
)
func TestParseMySQLConnStr(t *testing.T) {
cases := []struct {
connStr string
cfg SDBConfig
}{
{
connStr: "regionadmin:abcdefg@tcp(192.168.222.22:3306)/yunioncloud?parseTime=True",
cfg: SDBConfig{
Hostport: "192.168.222.22:3306",
Database: "yunioncloud",
Username: "regionadmin",
Password: "abcdefg",
},
},
}
for _, c := range cases {
got := ParseMySQLConnStr(c.connStr)
if jsonutils.Marshal(got).String() != jsonutils.Marshal(c.cfg).String() {
t.Errorf("parse %s got %s want %s", c.connStr, jsonutils.Marshal(got), jsonutils.Marshal(c.cfg))
}
}
}
pkg/util/splitable/splitable.go
+16
View File
@@ -38,6 +38,8 @@ type SSplitTableSpec struct {
lastTableSpec *sqlchemy.STableSpec
lastTableLock *sync.Mutex
lastTableExpire time.Time
extraOpts sqlchemy.TableExtraOptions
}
const (
@@ -140,6 +142,20 @@ func (t *SSplitTableSpec) Drop() error {
return nil
}
func (t *SSplitTableSpec) GetExtraOptions() sqlchemy.TableExtraOptions {
return t.extraOpts
}
func (t *SSplitTableSpec) SetExtraOptions(opts sqlchemy.TableExtraOptions) {
if t.extraOpts == nil {
t.extraOpts = opts
return
}
for k := range opts {
t.extraOpts[k] = opts[k]
}
}
func NewSplitTableSpec(s interface{}, name string, indexField string, dateField string, maxDuration time.Duration, maxSegments int, dbName sqlchemy.DBName) (*SSplitTableSpec, error) {
spec := sqlchemy.NewTableSpecFromStructWithDBName(s, name, dbName)
/*indexCol := spec.ColumnSpec(indexField)
vendor/modules.txt
Vendored
+2 -2
View File
@@ -1458,7 +1458,7 @@ sigs.k8s.io/structured-merge-diff/v4/value
# sigs.k8s.io/yaml v1.2.0
## explicit; go 1.12
sigs.k8s.io/yaml
# yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231204065558-c4edc0938f2e
# yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20231204130443-d54a8b16c7b7
## explicit; go 1.18
yunion.io/x/cloudmux/pkg/apis
yunion.io/x/cloudmux/pkg/apis/billing
@@ -1616,7 +1616,7 @@ yunion.io/x/pkg/utils
# yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e
## explicit; go 1.12
yunion.io/x/s3cli
# yunion.io/x/sqlchemy v1.1.2-0.20231201052514-97026b18ccf0
# yunion.io/x/sqlchemy v1.1.2-0.20231204175132-1eb294922a51
## explicit; go 1.17
yunion.io/x/sqlchemy
yunion.io/x/sqlchemy/backends
vendor/yunion.io/x/cloudmux/pkg/multicloud/apsara/apsara.go
Generated Vendored
+4 -1
View File
@@ -262,6 +262,9 @@ func (self *SApsaraClient) getDefaultClient(regionId string) (*sdk.Client, error
return nil, errors.Wrapf(err, "ParseQuery(%s)", req.URL.RawQuery)
}
action := params.Get("OpenApiAction")
if len(action) == 0 {
action = params.Get("Action")
}
service := strings.ToLower(params.Get("Product"))
respCheck := func(resp *http.Response) error {
if self.cpcfg.UpdatePermission != nil {
@@ -288,7 +291,7 @@ func (self *SApsaraClient) getDefaultClient(regionId string) (*sdk.Client, error
}
return nil
}
if self.cpcfg.ReadOnly {
if self.cpcfg.ReadOnly && len(action) > 0 {
for _, prefix := range []string{"Get", "List", "Describe"} {
if strings.HasPrefix(action, prefix) {
return respCheck, nil
vendor/yunion.io/x/sqlchemy/backends/clickhouse/clickhouse.go
Generated Vendored
+69 -27
View File
@@ -104,6 +104,17 @@ func (click *SClickhouseBackend) UpdateSQLTemplate() string {
return "ALTER TABLE `{{ .Table }}` UPDATE {{ .Columns }} WHERE {{ .Conditions }}"
}
func MySQLExtraOptions(hostport, database, table, user, passwd string) sqlchemy.TableExtraOptions {
return sqlchemy.TableExtraOptions{
EXTRA_OPTION_ENGINE_KEY: EXTRA_OPTION_ENGINE_VALUE_MYSQL,
EXTRA_OPTION_CLICKHOUSE_MYSQL_HOSTPORT_KEY: hostport,
EXTRA_OPTION_CLICKHOUSE_MYSQL_DATABASE_KEY: database,
EXTRA_OPTION_CLICKHOUSE_MYSQL_TABLE_KEY: table,
EXTRA_OPTION_CLICKHOUSE_MYSQL_USERNAME_KEY: user,
EXTRA_OPTION_CLICKHOUSE_MYSQL_PASSWORD_KEY: passwd,
}
}
func (click *SClickhouseBackend) GetCreateSQLs(ts sqlchemy.ITableSpec) []string {
cols := make([]string, 0)
primaries := make([]string, 0)
@@ -129,35 +140,51 @@ func (click *SClickhouseBackend) GetCreateSQLs(ts sqlchemy.ITableSpec) []string
}
}
}
createSql := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` (\n%s\n) ENGINE MergeTree", ts.Name(), strings.Join(cols, ",\n"))
if len(orderbys) == 0 {
orderbys = primaries
}
if len(partitions) > 0 {
createSql += fmt.Sprintf("\nPARTITION BY (%s)", strings.Join(partitions, ", "))
}
if len(primaries) > 0 {
createSql += fmt.Sprintf("\nPRIMARY KEY (%s)", strings.Join(primaries, ", "))
newOrderBys := make([]string, len(primaries))
copy(newOrderBys, primaries)
for _, f := range orderbys {
if !utils.IsInStringArray(f, newOrderBys) {
newOrderBys = append(newOrderBys, f)
}
createSql := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` (\n%s\n) ENGINE = ", ts.Name(), strings.Join(cols, ",\n"))
extraOpts := ts.GetExtraOptions()
engine := extraOpts.Get(EXTRA_OPTION_ENGINE_KEY)
switch engine {
case EXTRA_OPTION_ENGINE_VALUE_MYSQL:
// mysql
createSql += fmt.Sprintf("MySQL('%s', '%s', '%s', '%s', '%s')",
extraOpts.Get(EXTRA_OPTION_CLICKHOUSE_MYSQL_HOSTPORT_KEY),
extraOpts.Get(EXTRA_OPTION_CLICKHOUSE_MYSQL_DATABASE_KEY),
extraOpts.Get(EXTRA_OPTION_CLICKHOUSE_MYSQL_TABLE_KEY),
extraOpts.Get(EXTRA_OPTION_CLICKHOUSE_MYSQL_USERNAME_KEY),
extraOpts.Get(EXTRA_OPTION_CLICKHOUSE_MYSQL_PASSWORD_KEY),
)
default:
// mergetree
createSql += "MergeTree()"
if len(orderbys) == 0 {
orderbys = primaries
}
orderbys = newOrderBys
if len(partitions) > 0 {
createSql += fmt.Sprintf("\nPARTITION BY (%s)", strings.Join(partitions, ", "))
}
if len(primaries) > 0 {
createSql += fmt.Sprintf("\nPRIMARY KEY (%s)", strings.Join(primaries, ", "))
newOrderBys := make([]string, len(primaries))
copy(newOrderBys, primaries)
for _, f := range orderbys {
if !utils.IsInStringArray(f, newOrderBys) {
newOrderBys = append(newOrderBys, f)
}
}
orderbys = newOrderBys
}
if len(orderbys) > 0 {
createSql += fmt.Sprintf("\nORDER BY (%s)", strings.Join(orderbys, ", "))
} else {
createSql += "\nORDER BY tuple()"
}
if ttlCol != nil {
ttlCount, ttlUnit := ttlCol.GetTTL()
createSql += fmt.Sprintf("\nTTL `%s` + INTERVAL %d %s", ttlCol.Name(), ttlCount, ttlUnit)
}
// set default time zone of table to UTC
createSql += "\nSETTINGS index_granularity=8192"
}
if len(orderbys) > 0 {
createSql += fmt.Sprintf("\nORDER BY (%s)", strings.Join(orderbys, ", "))
} else {
createSql += fmt.Sprintf("\nORDER BY tuple()")
}
if ttlCol != nil {
ttlCount, ttlUnit := ttlCol.GetTTL()
createSql += fmt.Sprintf("\nTTL `%s` + INTERVAL %d %s", ttlCol.Name(), ttlCount, ttlUnit)
}
// set default time zone of table to UTC
createSql += "\nSETTINGS index_granularity=8192"
return []string{
createSql,
}
@@ -216,6 +243,21 @@ func (click *SClickhouseBackend) FetchTableColumnSpecs(ts sqlchemy.ITableSpec) (
}
func (click *SClickhouseBackend) GetColumnSpecByFieldType(table *sqlchemy.STableSpec, fieldType reflect.Type, fieldname string, tagmap map[string]string, isPointer bool) sqlchemy.IColumnSpec {
extraOpts := table.GetExtraOptions()
engine := extraOpts.Get(EXTRA_OPTION_ENGINE_KEY)
isMySQLEngine := false
switch engine {
case EXTRA_OPTION_ENGINE_VALUE_MYSQL:
isMySQLEngine = true
}
colSpec := click.getColumnSpecByFieldTypeInternal(table, fieldType, fieldname, tagmap, isPointer)
if isMySQLEngine && colSpec.IsPrimary() {
colSpec.SetPrimary(false)
}
return colSpec
}
func (click *SClickhouseBackend) getColumnSpecByFieldTypeInternal(table *sqlchemy.STableSpec, fieldType reflect.Type, fieldname string, tagmap map[string]string, isPointer bool) sqlchemy.IColumnSpec {
switch fieldType {
case tristate.TriStateType:
col := NewTristateColumn(table.Name(), fieldname, tagmap, isPointer)
vendor/yunion.io/x/sqlchemy/backends/clickhouse/consts.go
Generated Vendored
+11
View File
@@ -23,4 +23,15 @@ const (
// TAG_TTL defines table TTL
TAG_TTL = "clickhouse_ttl"
EXTRA_OPTION_ENGINE_KEY = "clickhouse_engine"
EXTRA_OPTION_ENGINE_VALUE_MERGETRUE = "MergeTree"
EXTRA_OPTION_ENGINE_VALUE_MYSQL = "MySQL"
// 'host:port', 'database', 'table', 'user', 'password'
EXTRA_OPTION_CLICKHOUSE_MYSQL_HOSTPORT_KEY = "clickhouse_mysql_hostport"
EXTRA_OPTION_CLICKHOUSE_MYSQL_DATABASE_KEY = "clickhouse_mysql_database"
EXTRA_OPTION_CLICKHOUSE_MYSQL_TABLE_KEY = "clickhouse_mysql_table"
EXTRA_OPTION_CLICKHOUSE_MYSQL_USERNAME_KEY = "clickhouse_mysql_username"
EXTRA_OPTION_CLICKHOUSE_MYSQL_PASSWORD_KEY = "clickhouse_mysql_password"
)
vendor/yunion.io/x/sqlchemy/extraopts.go
Generated Vendored
+47
View File
@@ -0,0 +1,47 @@
// Copyright 2019 Yunion
//
// 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 sqlchemy
type TableExtraOptions map[string]string
func (opts TableExtraOptions) Get(key string) string {
return opts[key]
}
func (opts TableExtraOptions) Set(key string, val string) TableExtraOptions {
opts[key] = val
return opts
}
func (opts TableExtraOptions) Contains(key string) bool {
if _, ok := opts[key]; ok {
return true
}
return false
}
func (ts *STableSpec) GetExtraOptions() TableExtraOptions {
return ts.extraOptions
}
func (ts *STableSpec) SetExtraOptions(opts TableExtraOptions) {
if ts.extraOptions == nil {
ts.extraOptions = opts
return
}
for k := range opts {
ts.extraOptions[k] = opts[k]
}
}
vendor/yunion.io/x/sqlchemy/table.go
Generated Vendored
+20 -1
View File
@@ -85,6 +85,12 @@ type ITableSpec interface {
// Drop drops table
Drop() error
// getter of Extra Options
GetExtraOptions() TableExtraOptions
// setter of Extra Options
SetExtraOptions(opts TableExtraOptions)
}
// STableSpec defines the table specification, which implements ITableSpec
@@ -95,6 +101,8 @@ type STableSpec struct {
_indexes []STableIndex
_contraints []STableConstraint
extraOptions TableExtraOptions
sDBReferer
}
@@ -129,7 +137,18 @@ func NewTableSpecFromStructWithDBName(s interface{}, name string, dbName DBName)
dbName: dbName,
},
}
// table.struct2TableSpec(val)
return table
}
func NewTableSpecFromISpecWithDBName(spec ITableSpec, name string, dbName DBName, extraOpts TableExtraOptions) *STableSpec {
table := &STableSpec{
name: name,
structType: spec.DataType(),
sDBReferer: sDBReferer{
dbName: dbName,
},
extraOptions: extraOpts,
}
return table
}