From 4a237dc474065ead6a7a6fc8572d418a7136b5ce Mon Sep 17 00:00:00 2001 From: GoogleFan Date: Tue, 17 May 2022 11:16:04 +0800 Subject: [PATCH 1/4] =?UTF-8?q?=E5=A6=82=E6=9E=9C=E5=AD=97=E6=AE=B5?= =?UTF-8?q?=E7=B1=BB=E5=9E=8B=E5=8F=AF=E4=BB=A5=E4=B8=BAnull=20=E5=88=99?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0=20Nullable,=E5=90=A6=E5=88=99=E5=90=8C?= =?UTF-8?q?=E6=AD=A5mysql=E5=9B=9E=E6=8A=A5nil=E5=BC=82=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/mysqlx/clickhouse.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pkg/mysqlx/clickhouse.go b/pkg/mysqlx/clickhouse.go index a88cd5c..488dc69 100644 --- a/pkg/mysqlx/clickhouse.go +++ b/pkg/mysqlx/clickhouse.go @@ -29,6 +29,10 @@ func ToClickhouseTable(dsn, db, table, indexes string, withTime bool) ([]string, } // type converter columns[i].Type = toClickhouseType(c.Type) + // 如果字段类型可以为null 则添加 Nullable,否则同步mysql回报nil异常 + if c.Null == "YES" { + columns[i].Type = "Nullable(" + columns[i].Type + ")" + } newColumns = append(newColumns, table2.Column{ Name: columns[i].Field, Type: columns[i].Type, From c8709393fd20e97fa0480182c924480ef0f444e8 Mon Sep 17 00:00:00 2001 From: GoogleFan Date: Wed, 25 May 2022 19:12:12 +0800 Subject: [PATCH 2/4] =?UTF-8?q?bug=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=BD=93Mysq?= =?UTF-8?q?l=E5=AD=97=E6=AE=B5=E7=B1=BB=E5=9E=8B=E6=98=AFbit=E6=97=B6?= =?UTF-8?q?=EF=BC=8C=E6=B7=BB=E5=8A=A0=E5=88=A4=E6=96=AD=E8=BD=AC=E6=8D=A2?= =?UTF-8?q?=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/dm/choperator/chproxyoperator.go | 11 ++++++++++- cmd/dm/choperator/ckgroupoperator.go | 11 ++++++++++- cmd/dm/util/mysqltypeconv.go | 3 +++ cmd/rtu/model/debeziumsyncdatatype.go | 3 +++ 4 files changed, 26 insertions(+), 2 deletions(-) diff --git a/cmd/dm/choperator/chproxyoperator.go b/cmd/dm/choperator/chproxyoperator.go index 0d963ff..eb0d2c9 100644 --- a/cmd/dm/choperator/chproxyoperator.go +++ b/cmd/dm/choperator/chproxyoperator.go @@ -41,7 +41,16 @@ func (cpo *ChProxyOperator) BatchInsert(insertData [][]interface{}, insertQuery if uar, ok := val.(time.Time); ok { return uar.In(ShangHaiLocation).Format("2006-01-02 15:04:05") } else if uar, ok := val.([]uint8); ok { - return string(uar) + // 当Mysql字段类型是bit时，添加转换 + sv := string(uar) + switch sv { + case "\x00": + return 0 + case "\x01": + return 1 + default: + return sv + } } return val }(), arr[key]) diff --git a/cmd/dm/choperator/ckgroupoperator.go b/cmd/dm/choperator/ckgroupoperator.go index bfb82d4..0644665 100644 --- a/cmd/dm/choperator/ckgroupoperator.go +++ b/cmd/dm/choperator/ckgroupoperator.go @@ -36,7 +36,16 @@ func (cgo *CkGroupOperator) MysqlBatchInsert(insertData [][]interface{}, insertQ if uar, ok := val.(time.Time); ok { return uar.In(ShangHaiLocation).Format("2006-01-02 15:04:05") } else if uar, ok := val.([]uint8); ok { - return string(uar) + // 当Mysql字段类型是bit时，添加转换 + sv := string(uar) + switch sv { + case "\x00": + return 0 + case "\x01": + return 1 + default: + return sv + } } return val diff --git a/cmd/dm/util/mysqltypeconv.go b/cmd/dm/util/mysqltypeconv.go index 1261ecd..00dac88 100644 --- a/cmd/dm/util/mysqltypeconv.go +++ b/cmd/dm/util/mysqltypeconv.go @@ -128,6 +128,9 @@ func ParseValueByType(vv interface{}, t DataType) (interface{}, error) { // ParseTypeByMysqlType 将MySQL的数据类型转换为Go语言内部转换用的DataType func ParseTypeByMysqlType(sqlType string) DataType { sqlType = strings.ToLower(sqlType) + if strings.Contains(sqlType, "bit") { + return DataTypeInt + } if strings.Contains(sqlType, "int") { return DataTypeInt } diff --git a/cmd/rtu/model/debeziumsyncdatatype.go b/cmd/rtu/model/debeziumsyncdatatype.go index 073da75..22f2de5 100644 --- a/cmd/rtu/model/debeziumsyncdatatype.go +++ b/cmd/rtu/model/debeziumsyncdatatype.go @@ -30,6 +30,9 @@ var NullValMap = map[DataType]interface{}{ // ParseTypeByMysqlType 将MySQL的数据类型转换为Go语言内部转换用的DataType func ParseTypeByMysqlType(sqlType string) DataType { sqlType = strings.ToLower(sqlType) + if strings.Contains(sqlType, "bit") { + return DataTypeInt + } if strings.Contains(sqlType, "int") { return DataTypeInt } From ebf0c480be7fce30518eb273c20e59cd9c1b7875 Mon Sep 17 00:00:00 2001 From: GoogleFan Date: Thu, 26 May 2022 16:42:13 +0800 Subject: [PATCH 3/4] =?UTF-8?q?bug=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=BD=93Mysq?= =?UTF-8?q?l=20=E7=B1=BB=E5=9E=8B=E4=BF=AE=E9=A5=B0=E7=AC=A6=20=E6=98=AF?= =?UTF-8?q?=E7=A9=BA=E6=A0=BC=20=E8=80=8C=E9=9D=9E=EF=BC=88=EF=BC=89?= =?UTF-8?q?=E6=97=B6=EF=BC=8C=E5=BC=82=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/mysqlx/clickhouse.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/pkg/mysqlx/clickhouse.go b/pkg/mysqlx/clickhouse.go index 488dc69..6c9914b 100644 --- a/pkg/mysqlx/clickhouse.go +++ b/pkg/mysqlx/clickhouse.go @@ -2,11 +2,10 @@ package mysqlx import ( "errors" - "strings" - "github.com/zeromicro/cds/pkg/strx" table2 "github.com/zeromicro/cds/pkg/table" "github.com/zeromicro/go-zero/core/logx" + "strings" ) func ToClickhouseTable(dsn, db, table, indexes string, withTime bool) ([]string, string, error) { @@ -80,9 +79,15 @@ func ToClickhouseTable(dsn, db, table, indexes string, withTime bool) ([]string, } func toClickhouseType(typ string) string { - after := strx.SubAfterLast(typ, ")", "") - typ = strx.SubBeforeLast(typ, "(", typ) typ = strings.ToLower(typ) + var after string + if strings.Contains(typ, "(") { + after = strx.SubAfterLast(typ, ")", "") + typ = strx.SubBeforeLast(typ, "(", typ) + } else { + after = strx.SubAfterLast(typ, " ", "") + typ = strx.SubBeforeLast(typ, " ", typ) + } switch typ { case "bool", "boolean", "tinyint": return withUnsigned("Int8", after) From 4d95f91bf7f8ebf5ecb3ad435ea594a8b631f28a Mon Sep 17 00:00:00 2001 From: GoogleFan Date: Thu, 26 May 2022 20:08:06 +0800 Subject: [PATCH 4/4] =?UTF-8?q?bug=E4=BF=AE=E5=A4=8D=EF=BC=9A=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=E5=BD=93=E9=80=89=E6=8B=A9=E5=A4=9A=E4=B8=AA=E8=A1=A8?= =?UTF-8?q?=E6=97=B6=E5=87=BA=E9=94=99=E7=9A=84=E6=83=85=E5=86=B5=EF=BC=8C?= =?UTF-8?q?=E7=94=A8SelectedTable=E6=95=B0=E7=BB=84=E4=B8=AA=E6=95=B0?= =?UTF-8?q?=E8=B5=8B=E5=80=BC=E7=BB=99QueryKey?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/galaxy/internal/logic/rtuaddlogic.go | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/cmd/galaxy/internal/logic/rtuaddlogic.go b/cmd/galaxy/internal/logic/rtuaddlogic.go index ad42120..7a6a9ba 100644 --- a/cmd/galaxy/internal/logic/rtuaddlogic.go +++ b/cmd/galaxy/internal/logic/rtuaddlogic.go @@ -35,11 +35,15 @@ func (l *RtuAddLogic) RtuAdd(req types.RtuModel) (*types.RtuModel, error) { sourceType := "canal-" + config.TYPE_MYSQL if strings.HasPrefix(req.Source.Dsn, "mongodb://") { req.Source.QueryKey = []string{"_id"} - sourceType = "connector-" + config.TYPE_MONGODB - } else { - req.Source.QueryKey = []string{""} } + // 修复当选择多个表时出错的情况，用SelectedTable数组个数赋值给QueryKey + if len(req.Source.QueryKey) == 0 { + req.Source.QueryKey = req.Source.SelectedTable + } + //else { + // req.Source.QueryKey = []string{""} + //} shards, e := json.Marshal(l.svcCtx.Config.CkDataNodes[1:]) if e != nil { logx.Error(e)