null字段添加Nullable配置,否则同步mysql回报nil异常;增加bit类型go处理

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])

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

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
 	}

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)

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
 	}

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) {
@@ -29,6 +28,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,
@@ -76,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)