diff --git a/frontend/src/views/ops/db/SyncTaskEdit.vue b/frontend/src/views/ops/db/SyncTaskEdit.vue
index 7f5a9652..bd585b1c 100644
--- a/frontend/src/views/ops/db/SyncTaskEdit.vue
+++ b/frontend/src/views/ops/db/SyncTaskEdit.vue
@@ -136,7 +136,7 @@
-
+
@@ -155,15 +155,15 @@
-
-
-
-
-
-
+
+
+
+
+
+
diff --git a/server/internal/db/application/db_data_sync.go b/server/internal/db/application/db_data_sync.go
index 1bd25e1b..9b77fcd7 100644
--- a/server/internal/db/application/db_data_sync.go
+++ b/server/internal/db/application/db_data_sync.go
@@ -148,7 +148,24 @@ func (app *dataSyncAppImpl) RunCronJob(ctx context.Context, id uint64) error {
logx.ErrorfContext(ctx, "data source connection unavailable: %s", err.Error())
return
}
- updSql = fmt.Sprintf("and %s > %s", task.UpdField, strings.Trim(task.UpdFieldVal, " "))
+ srcConn, err := app.dbApp.GetDbConn(uint64(task.SrcDbId), task.SrcDbName)
+ if err != nil {
+ logx.ErrorfContext(ctx, "failed to connect to the source database: %s", err.Error())
+ return
+ }
+
+ updFieldDataType := dbi.DefaultDbDataType
+ srcConn.WalkQueryRows(context.Background(), task.DataSql, func(row map[string]any, columns []*dbi.QueryColumn) error {
+ for _, column := range columns {
+ if strings.EqualFold(column.Name, cmp.Or(task.UpdFieldSrc, task.UpdField)) {
+ updFieldDataType = column.DbDataType
+ break
+ }
+ }
+ return errorx.NewBiz("get column data type")
+ })
+
+ updSql = fmt.Sprintf("and %s > %s", task.UpdField, updFieldDataType.DataType.SQLValue(task.UpdFieldVal))
orderSql = "order by " + task.UpdField + " asc "
}
// 正则判断DataSql是否以where .*结尾，如果是则不添加where 1 = 1
@@ -158,9 +175,9 @@ func (app *dataSyncAppImpl) RunCronJob(ctx context.Context, id uint64) error {
}
// 组装查询sql
- sql := fmt.Sprintf("%s %s %s %s", task.DataSql, where, updSql, orderSql)
+ sqlStr := fmt.Sprintf("%s %s %s %s", task.DataSql, where, updSql, orderSql)
- log, err := app.doDataSync(ctx, sql, task)
+ log, err := app.doDataSync(ctx, sqlStr, task)
if err != nil {
log.ErrText = fmt.Sprintf("execution failure: %s", err.Error())
logx.ErrorContext(ctx, log.ErrText)
@@ -335,11 +352,7 @@ func (app *dataSyncAppImpl) srcData2TargetDb(srcRes []map[string]any, fieldMap [
}
// 如果指定了更新字段，则以更新字段取值
- if task.UpdFieldSrc != "" {
- setUpdateFieldVal(task.UpdFieldSrc)
- } else {
- setUpdateFieldVal(updFieldName)
- }
+ setUpdateFieldVal(cmp.Or(task.UpdFieldSrc, updFieldName))
targetDialect := targetDbConn.GetDialect()
diff --git a/server/internal/db/application/db_transfer.go b/server/internal/db/application/db_transfer.go
index d5083cc5..1541c714 100644
--- a/server/internal/db/application/db_transfer.go
+++ b/server/internal/db/application/db_transfer.go
@@ -309,6 +309,7 @@ func (app *dbTransferAppImpl) transfer2Db(ctx context.Context, taskId uint64, lo
err = sqlparser.SQLSplit(pr, func(stmt string) error {
if _, err := targetConn.TxExecContext(ctx, tx, stmt); err != nil {
+ app.EndTransfer(ctx, logId, taskId, fmt.Sprintf("执行sql出错: %s", stmt), err, nil)
pw.CloseWithError(err)
return err
}
diff --git a/server/internal/db/dbm/dbi/metasql/dm_meta.sql b/server/internal/db/dbm/dbi/metasql/dm_meta.sql
index bb7b6448..e9dbd973 100644
--- a/server/internal/db/dbm/dbi/metasql/dm_meta.sql
+++ b/server/internal/db/dbm/dbi/metasql/dm_meta.sql
@@ -1,8 +1,8 @@
--DM_DB_SCHEMAS 库schemas
-select
- distinct owner as SCHEMA_NAME
-from all_objects
-order by owner
+SELECT
+ DISTINCT object_name as SCHEMA_NAME
+FROM ALL_OBJECTS
+WHERE OBJECT_TYPE = 'SCH'
---------------------------------------
--DM_TABLE_INFO 表详细信息
SELECT a.object_name as TABLE_NAME,
diff --git a/server/internal/db/dbm/dm/sqlgen.go b/server/internal/db/dbm/dm/sqlgen.go
index 3fb2c859..02714a30 100644
--- a/server/internal/db/dbm/dm/sqlgen.go
+++ b/server/internal/db/dbm/dm/sqlgen.go
@@ -4,6 +4,7 @@ import (
"fmt"
"mayfly-go/internal/db/dbm/dbi"
"mayfly-go/pkg/utils/collx"
+ "regexp"
"strings"
)
@@ -87,12 +88,16 @@ func (sg *SQLGenerator) GenInsert(tableName string, columns []dbi.Column, values
quote := quoter.Quote
if duplicateStrategy == dbi.DuplicateStrategyNone {
- identityInsert := fmt.Sprintf("set identity_insert %s on;", quote(tableName))
+ identityInsert := ""
+ // 有自增列的才加上这个语句
+ if collx.AnyMatch(columns, func(column dbi.Column) bool { return column.IsIdentity }) {
+ identityInsert = fmt.Sprintf("set identity_insert %s on;", quote(tableName))
+ }
// 达梦数据库只能一条条的执行insert语句，所以这里需要将values拆分成多条insert语句
return collx.ArrayMap(values, func(value []any) string {
- columnStr, valuesStrs := dbi.GenInsertSqlColumnAndValues(sg.Dialect, DbTypeDM, columns, [][]any{value})
- return fmt.Sprintf("%s insert into %s (%s) values %s", identityInsert, quote(tableName), columnStr, strings.Join(valuesStrs, ",\n"))
+ columnStr, valuesStrs := dbi.GenInsertSqlColumnAndValues(sg.Dialect, DbTypeDM, columns, values)
+ return fmt.Sprintf("%s insert into %s %s values %s", identityInsert, quote(tableName), columnStr, strings.Join(valuesStrs, ",\n"))
})
}
@@ -156,9 +161,9 @@ func (sg *SQLGenerator) GenInsert(tableName string, columns []dbi.Column, values
return collx.AsArray(sqlTemp)
}
-func (msg *SQLGenerator) genColumnBasicSql(quoter dbi.Quoter, column dbi.Column) string {
+func (sg *SQLGenerator) genColumnBasicSql(quoter dbi.Quoter, column dbi.Column) string {
colName := quoter.Quote(column.ColumnName)
- dataType := string(column.DataType)
+ dataType := column.DataType
incr := ""
if column.IsIdentity {
@@ -174,7 +179,10 @@ func (msg *SQLGenerator) genColumnBasicSql(quoter dbi.Quoter, column dbi.Column)
if column.ColumnDefault != "" && !strings.Contains(column.ColumnDefault, "(") {
// 哪些字段类型默认值需要加引号
mark := false
- if collx.ArrayAnyMatches([]string{"char", "text", "date", "time", "lob"}, strings.ToLower(dataType)) {
+ if regexp.MustCompile(`'.*'`).MatchString(column.ColumnDefault) {
+ // 字符串默认值
+ mark = false
+ } else if collx.ArrayAnyMatches([]string{"char", "text", "date", "time", "lob"}, strings.ToLower(dataType)) {
// 当数据类型是日期时间，默认值是日期时间函数时，默认值不需要引号
if collx.ArrayAnyMatches([]string{"date", "time"}, strings.ToLower(dataType)) &&
collx.ArrayAnyMatches([]string{"DATE", "TIME"}, strings.ToUpper(column.ColumnDefault)) {
@@ -182,6 +190,10 @@ func (msg *SQLGenerator) genColumnBasicSql(quoter dbi.Quoter, column dbi.Column)
} else {
mark = true
}
+ // 空
+ if column.ColumnDefault == "NULL" {
+ mark = false
+ }
}
if mark {
defVal = fmt.Sprintf(" DEFAULT '%s'", column.ColumnDefault)
diff --git a/server/internal/db/dbm/mssql/dialect.go b/server/internal/db/dbm/mssql/dialect.go
index 75f7eca4..ffe461da 100644
--- a/server/internal/db/dbm/mssql/dialect.go
+++ b/server/internal/db/dbm/mssql/dialect.go
@@ -1,11 +1,9 @@
package mssql
import (
- "database/sql"
"fmt"
"mayfly-go/internal/db/dbm/dbi"
"mayfly-go/pkg/logx"
- "mayfly-go/pkg/utils/collx"
"strings"
"time"
)
@@ -24,176 +22,6 @@ type MssqlDialect struct {
dc *dbi.DbConn
}
-func (md *MssqlDialect) BatchInsert(tx *sql.Tx, tableName string, columns []string, values [][]any, duplicateStrategy int) (int64, error) {
-
- if duplicateStrategy == dbi.DuplicateStrategyUpdate {
- return md.batchInsertMerge(tx, tableName, columns, values, duplicateStrategy)
- }
-
- return md.batchInsertSimple(tx, tableName, columns, values, duplicateStrategy)
-}
-
-func (md *MssqlDialect) batchInsertSimple(tx *sql.Tx, tableName string, columns []string, values [][]any, duplicateStrategy int) (int64, error) {
- // 把二维数组转为一维数组
- var args []any
- var singleSize int // 一条数据的参数个数
- for i, v := range values {
- if i == 0 {
- singleSize = len(v)
- }
- args = append(args, v...)
- }
-
- // 判断如果参数超过2000，则分批次执行，mssql允许最大参数为2100，保险起见，这里限制到2000
- if len(args) > 2000 {
-
- rows := 2000 / singleSize // 每批次最大数据条数
- mp := make(map[any][][]any)
-
- // 把values拆成多份，每份不能超过rows条
- length := len(values)
- for i := 0; i < length; i += rows {
- if i+rows <= length {
- mp[i] = values[i : i+rows]
- } else {
- mp[i] = values[i:length]
- }
- }
-
- var count int64
- for _, v := range mp {
- res, err := md.batchInsertSimple(tx, tableName, columns, v, duplicateStrategy)
- if err != nil {
- return count, err
- }
- count += res
- }
- return count, nil
- }
-
- msMetadata := md.dc.GetMetadata()
- schema := md.dc.Info.CurrentSchema()
- ignoreDupSql := ""
- if duplicateStrategy == dbi.DuplicateStrategyIgnore {
- // ALTER TABLE dbo.TEST ADD CONSTRAINT uniqueRows UNIQUE (ColA, ColB, ColC, ColD) WITH (IGNORE_DUP_KEY = ON)
- indexs, _ := msMetadata.(*MssqlMetadata).getTableIndexWithPK(tableName)
- // 收集唯一索引涉及到的字段
- uniqueColumns := make([]string, 0)
- for _, index := range indexs {
- if index.IsUnique {
- cols := strings.Split(index.ColumnName, ",")
- for _, col := range cols {
- if !collx.ArrayContains(uniqueColumns, col) {
- uniqueColumns = append(uniqueColumns, col)
- }
- }
- }
- }
- if len(uniqueColumns) > 0 {
- // 设置忽略重复键
- ignoreDupSql = fmt.Sprintf("ALTER TABLE %s.%s ADD CONSTRAINT uniqueRows UNIQUE (%s) WITH (IGNORE_DUP_KEY = {sign})", schema, tableName, strings.Join(uniqueColumns, ","))
- _, _ = md.dc.TxExec(tx, strings.ReplaceAll(ignoreDupSql, "{sign}", "ON"))
- }
- }
-
- // 生成占位符字符串：如：(?,?)
- // 重复字符串并用逗号连接
- repeated := strings.Repeat("?,", len(columns))
- // 去除最后一个逗号，占位符由括号包裹
- placeholder := fmt.Sprintf("(%s)", strings.TrimSuffix(repeated, ","))
-
- // 重复占位符字符串n遍
- repeated = strings.Repeat(placeholder+",", len(values))
- // 去除最后一个逗号
- placeholder = strings.TrimSuffix(repeated, ",")
-
- quote := md.Quoter().Quote
- baseTable := fmt.Sprintf("%s.%s", quote(schema), quote(tableName))
-
- sqlStr := fmt.Sprintf("insert into %s (%s) values %s", baseTable, strings.Join(columns, ","), placeholder)
- // 执行批量insert sql
-
- // 设置允许填充自增列之后，显示指定列名可以插入自增列
- identityInsertOn := fmt.Sprintf("SET IDENTITY_INSERT [%s].[%s] ON", schema, tableName)
-
- res, err := md.dc.TxExec(tx, fmt.Sprintf("%s %s", identityInsertOn, sqlStr), args...)
-
- // 执行完之后，设置忽略重复键
- if ignoreDupSql != "" {
- _, _ = md.dc.TxExec(tx, strings.ReplaceAll(ignoreDupSql, "{sign}", "OFF"))
- }
- return res, err
-}
-
-func (md *MssqlDialect) batchInsertMerge(tx *sql.Tx, tableName string, columns []string, values [][]any, duplicateStrategy int) (int64, error) {
- msMetadata := md.dc.GetMetadata()
- schema := md.dc.Info.CurrentSchema()
- quote := md.Quoter().Quote
-
- // 收集MERGE 语句的 ON 子句条件
- caseSqls := make([]string, 0)
- pkCols := make([]string, 0)
-
- // 查询取出自增列字段, merge update不能修改自增列
- identityCols := make([]string, 0)
- cols, err := msMetadata.GetColumns(tableName)
- if err != nil {
- return 0, err
- }
- for _, col := range cols {
- if col.IsIdentity {
- identityCols = append(identityCols, col.ColumnName)
- }
- if col.IsPrimaryKey {
- pkCols = append(pkCols, col.ColumnName)
- name := quote(col.ColumnName)
- caseSqls = append(caseSqls, fmt.Sprintf(" T1.%s = T2.%s ", name, name))
- }
- }
- if len(pkCols) == 0 {
- return md.batchInsertSimple(tx, tableName, columns, values, duplicateStrategy)
- }
- // 重复数据处理策略
- insertVals := make([]string, 0)
- upds := make([]string, 0)
- insertCols := make([]string, 0)
- // 源数据占位sql
- phs := make([]string, 0)
- for _, column := range columns {
- if !collx.ArrayContains(identityCols, md.Quoter().Trim(column)) {
- upds = append(upds, fmt.Sprintf("T1.%s = T2.%s", column, column))
- }
- insertCols = append(insertCols, column)
- insertVals = append(insertVals, fmt.Sprintf("T2.%s", column))
- phs = append(phs, fmt.Sprintf("? %s", column))
- }
-
- // 把二维数组转为一维数组
- var args []any
- tmp := fmt.Sprintf("select %s", strings.Join(phs, ","))
- t2s := make([]string, 0)
- for _, v := range values {
- args = append(args, v...)
- t2s = append(t2s, tmp)
- }
- t2 := strings.Join(t2s, " UNION ALL ")
-
- sqlTemp := "MERGE INTO " + quote(schema) + "." + quote(tableName) + " T1 USING (" + t2 + ") T2 ON " + strings.Join(caseSqls, " AND ")
- sqlTemp += "WHEN NOT MATCHED THEN INSERT (" + strings.Join(insertCols, ",") + ") VALUES (" + strings.Join(insertVals, ",") + ") "
- sqlTemp += "WHEN MATCHED THEN UPDATE SET " + strings.Join(upds, ",")
-
- // 设置允许填充自增列之后，显示指定列名可以插入自增列
- identityInsertOn := fmt.Sprintf("SET IDENTITY_INSERT [%s].[%s] ON", schema, tableName)
-
- // 执行merge sql,必须要以分号结尾
- res, err := md.dc.TxExec(tx, fmt.Sprintf("%s %s;", identityInsertOn, sqlTemp), args...)
-
- if err != nil {
- logx.Errorf("执行sql失败：%s, sql: [ %s ]", err.Error(), sqlTemp)
- }
- return res, err
-}
-
func (md *MssqlDialect) CopyTable(copy *dbi.DbCopyTable) error {
msMetadata := md.dc.GetMetadata()
schema := md.dc.Info.CurrentSchema()
@@ -285,7 +113,7 @@ func (md *MssqlDialect) CopyTableDDL(tableName string, newTableName string) (str
return strings.Join(sqlArr, ";"), nil
}
-func (dx *MssqlDialect) Quoter() dbi.Quoter {
+func (md *MssqlDialect) Quoter() dbi.Quoter {
return mssqlQuoter
}
diff --git a/server/internal/db/dbm/mssql/sqlgen.go b/server/internal/db/dbm/mssql/sqlgen.go
index 1573a814..2eb001a6 100644
--- a/server/internal/db/dbm/mssql/sqlgen.go
+++ b/server/internal/db/dbm/mssql/sqlgen.go
@@ -103,12 +103,187 @@ func (sg *SQLGenerator) GenIndexDDL(table dbi.Table, indexs []dbi.Index) []strin
}
func (sg *SQLGenerator) GenInsert(tableName string, columns []dbi.Column, values [][]any, duplicateStrategy int) []string {
- return collx.AsArray("")
+
+ if duplicateStrategy == dbi.DuplicateStrategyUpdate {
+ return sg.batchInsertMerge(tableName, columns, values, duplicateStrategy)
+ }
+
+ return sg.batchInsertSimple(tableName, columns, values, duplicateStrategy)
}
-func (msg *SQLGenerator) genColumnBasicSql(quoter dbi.Quoter, column dbi.Column) string {
+func (sg *SQLGenerator) batchInsertSimple(tableName string, columns []dbi.Column, values [][]any, duplicateStrategy int) []string {
+ var res []string
+
+ // 把二维数组转为一维数组
+ var args []any
+ var singleSize int // 一条数据的参数个数
+ for i, v := range values {
+ if i == 0 {
+ singleSize = len(v)
+ }
+ args = append(args, v...)
+ }
+
+ // 判断如果参数超过2000，则分批次执行，mssql允许最大参数为2100，保险起见，这里限制到2000
+ if len(args) > 2000 {
+
+ rows := 2000 / singleSize // 每批次最大数据条数
+ mp := make(map[any][][]any)
+
+ // 把values拆成多份，每份不能超过rows条
+ length := len(values)
+ for i := 0; i < length; i += rows {
+ if i+rows <= length {
+ mp[i] = values[i : i+rows]
+ } else {
+ mp[i] = values[i:length]
+ }
+ }
+
+ var strs []string
+ for _, v := range mp {
+ res := sg.batchInsertSimple(tableName, columns, v, duplicateStrategy)
+ strs = append(strs, res...)
+ }
+ return strs
+ }
+
+ msMetadata := sg.dc.GetMetadata()
+ schema := sg.dc.Info.CurrentSchema()
+ ignoreDupSql := ""
+ if duplicateStrategy == dbi.DuplicateStrategyIgnore {
+ // ALTER TABLE dbo.TEST ADD CONSTRAINT uniqueRows UNIQUE (ColA, ColB, ColC, ColD) WITH (IGNORE_DUP_KEY = ON)
+ indexs, _ := msMetadata.(*MssqlMetadata).getTableIndexWithPK(tableName)
+ // 收集唯一索引涉及到的字段
+ uniqueColumns := make([]string, 0)
+ for _, index := range indexs {
+ if index.IsUnique {
+ cols := strings.Split(index.ColumnName, ",")
+ for _, col := range cols {
+ if !collx.ArrayContains(uniqueColumns, col) {
+ uniqueColumns = append(uniqueColumns, col)
+ }
+ }
+ }
+ }
+ if len(uniqueColumns) > 0 {
+ // 设置忽略重复键
+ ignoreDupSql = fmt.Sprintf("ALTER TABLE %s.%s ADD CONSTRAINT uniqueRows UNIQUE (%s) WITH (IGNORE_DUP_KEY = {sign})", schema, tableName, strings.Join(uniqueColumns, ","))
+ res = append(res, strings.ReplaceAll(ignoreDupSql, "{sign}", "ON"))
+ }
+ }
+
+ // 生成占位符字符串：如：(?,?)
+ // 重复字符串并用逗号连接
+ repeated := strings.Repeat("?,", len(columns))
+ // 去除最后一个逗号，占位符由括号包裹
+ placeholder := fmt.Sprintf("(%s)", strings.TrimSuffix(repeated, ","))
+
+ // 重复占位符字符串n遍
+ repeated = strings.Repeat(placeholder+",", len(values))
+ // 去除最后一个逗号
+ placeholder = strings.TrimSuffix(repeated, ",")
+
+ quote := sg.dc.GetDialect().Quoter().Quote
+ baseTable := fmt.Sprintf("%s.%s", quote(schema), quote(tableName))
+
+ // 设置允许填充自增列之后，显示指定列名可以插入自增列
+ identityInsertOn := ""
+ for _, column := range columns {
+ if column.IsIdentity {
+ identityInsertOn = fmt.Sprintf("SET IDENTITY_INSERT [%s].[%s] ON", schema, tableName)
+ }
+ }
+
+ columnStr, valuesStrs := dbi.GenInsertSqlColumnAndValues(sg.dc.GetDialect(), DbTypeMssql, columns, values)
+ insertSql := fmt.Sprintf("%s insert into %s %s VALUES \n%s", identityInsertOn, baseTable, columnStr, strings.Join(valuesStrs, ",\n"))
+ res = append(res, insertSql)
+
+ // 执行完之后，设置忽略重复键
+ if ignoreDupSql != "" {
+ res = append(res, strings.ReplaceAll(ignoreDupSql, "{sign}", "OFF"))
+ }
+ return res
+}
+
+func (sg *SQLGenerator) batchInsertMerge(tableName string, columns []dbi.Column, values [][]any, duplicateStrategy int) []string {
+ var res []string
+ schema := sg.dc.Info.CurrentSchema()
+ quote := sg.dc.GetDialect().Quoter().Quote
+
+ // 收集MERGE 语句的 ON 子句条件
+ caseArr := make([]string, 0)
+ // 搜集主键字段
+ pkCols := make([]string, 0)
+ // 查询取出自增列字段, merge update不能修改自增列
+ identityCols := make([]string, 0)
+ // 标记是否有自增字段
+ hashIdentity := false
+
+ for _, col := range columns {
+ if col.IsIdentity {
+ hashIdentity = true
+ identityCols = append(identityCols, col.ColumnName)
+ }
+ if col.IsPrimaryKey {
+ pkCols = append(pkCols, col.ColumnName)
+ name := quote(col.ColumnName)
+ caseArr = append(caseArr, fmt.Sprintf(" T1.%s = T2.%s ", name, name))
+ }
+ }
+ if len(pkCols) == 0 {
+ return sg.batchInsertSimple(tableName, columns, values, duplicateStrategy)
+ }
+ // 重复数据处理策略
+ updSqls := make([]string, 0)
+ insertVals := make([]string, 0)
+ insertCols := make([]string, 0)
+
+ for _, column := range columns {
+ columnName := column.ColumnName
+ quoteName := quote(columnName)
+ if !collx.ArrayContains(identityCols, sg.dc.GetDialect().Quoter().Trim(columnName)) {
+ updSqls = append(updSqls, fmt.Sprintf("T1.%s = T2.%s", columnName, columnName))
+ }
+ insertCols = append(insertCols, quoteName)
+ insertVals = append(insertVals, fmt.Sprintf("T2.%s", columnName))
+ }
+
+ // 把values二维数组转为一维数组
+ valArr := make([]string, 0)
+
+ valueSql := make([]string, 0)
+ for _, value := range values {
+ for j, column := range columns {
+ val := dbi.GetDbDataType(DbTypeMssql, column.DataType).DataType.SQLValue(value[j])
+ valArr = append(valArr, fmt.Sprintf("%s %s", val, column.ColumnName))
+ }
+ valueSql = append(valueSql, fmt.Sprintf("select %s", strings.Join(valArr, ", ")))
+ }
+
+ quoteTable := fmt.Sprintf("%s.%s", quote(schema), quote(tableName))
+ unionSql := strings.Join(valueSql, " UNION ALL ")
+ caseSql := strings.Join(caseArr, " AND ")
+
+ sqlTemp := "MERGE INTO " + quoteTable + " T1 USING (" + unionSql + ") T2 ON " + caseSql
+ sqlTemp += "WHEN NOT MATCHED THEN INSERT (" + strings.Join(insertCols, ",") + ") (" + strings.Join(insertVals, ",") + ") "
+ sqlTemp += "WHEN MATCHED THEN UPDATE SET " + strings.Join(updSqls, ",")
+
+ identityInsertOn := ""
+ if hashIdentity {
+ // 设置允许填充自增列之后，显示指定列名可以插入自增列
+ identityInsertOn = fmt.Sprintf("SET IDENTITY_INSERT [%s].[%s] ON", schema, tableName)
+
+ }
+ // 执行merge sql,必须要以分号结尾
+ res = append(res, fmt.Sprintf("%s %s", identityInsertOn, sqlTemp))
+
+ return res
+}
+
+func (sg *SQLGenerator) genColumnBasicSql(quoter dbi.Quoter, column dbi.Column) string {
colName := quoter.Quote(column.ColumnName)
- dataType := string(column.DataType)
+ dataType := column.DataType
incr := ""
if column.IsIdentity {
diff --git a/server/internal/db/dbm/mysql/sqlgen.go b/server/internal/db/dbm/mysql/sqlgen.go
index b2e06f7c..bf7d674f 100644
--- a/server/internal/db/dbm/mysql/sqlgen.go
+++ b/server/internal/db/dbm/mysql/sqlgen.go
@@ -96,7 +96,7 @@ func (msg *SQLGenerator) GenInsert(tableName string, columns []dbi.Column, value
}
func (msg *SQLGenerator) genColumnBasicSql(quoter dbi.Quoter, column dbi.Column) string {
- dataType := string(column.DataType)
+ dataType := column.DataType
incr := ""
if column.IsIdentity {
diff --git a/server/pkg/utils/collx/array.go b/server/pkg/utils/collx/array.go
index 4c62fe98..8b44c02e 100644
--- a/server/pkg/utils/collx/array.go
+++ b/server/pkg/utils/collx/array.go
@@ -185,3 +185,12 @@ func ArrayFilter[T any](array []T, fn func(T) bool) []T {
}
return filtered
}
+
+func AnyMatch[T any](array []T, fn func(T) bool) bool {
+ for _, val := range array {
+ if fn(val) {
+ return true
+ }
+ }
+ return false
+}