From 3f6fb5afeffd9c6bb719ee3f214c0f11f449a56b Mon Sep 17 00:00:00 2001
From: zongyangleo <zy12250605@163.com>
Date: Thu, 26 Dec 2024 04:11:28 +0000
Subject: [PATCH] =?UTF-8?q?!129=20fix:=20db=20=E7=9B=B8=E5=85=B3bug=20*=20?=
 =?UTF-8?q?fix:=20db=20=E7=9B=B8=E5=85=B3bug?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 frontend/src/views/ops/db/SyncTaskEdit.vue    |  14 +-
 .../internal/db/application/db_data_sync.go   |  29 ++-
 server/internal/db/application/db_transfer.go |   1 +
 .../internal/db/dbm/dbi/metasql/dm_meta.sql   |   8 +-
 server/internal/db/dbm/dm/sqlgen.go           |  24 ++-
 server/internal/db/dbm/mssql/dialect.go       | 174 +----------------
 server/internal/db/dbm/mssql/sqlgen.go        | 181 +++++++++++++++++-
 server/internal/db/dbm/mysql/sqlgen.go        |   2 +-
 server/pkg/utils/collx/array.go               |   9 +
 9 files changed, 240 insertions(+), 202 deletions(-)

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 @@
 
                     <el-tab-pane :label="$t('db.fieldMap')" :name="fieldTab" :disabled="!baseFieldCompleted">
                         <el-form-item prop="fieldMap" :label="$t('db.fieldMap')" required>
-                            <el-table :data="form.fieldMap" :max-height="400" size="small">
+                            <el-table :data="form.fieldMap" :max-height="650" size="small">
                                 <el-table-column prop="src" :label="$t('db.srcField')" :width="200" />
                                 <el-table-column prop="target" :label="$t('db.targetField')">
                                     <template #default="scope">
@@ -155,15 +155,15 @@
                     </el-tab-pane>
 
                     <el-tab-pane :label="$t('db.sqlPreview')" :name="sqlPreviewTab" :disabled="!baseFieldCompleted">
-                        <el-form-item prop="fieldMap" :label="$t('db.selectSql')">
-                            <el-input type="textarea" v-model="state.previewDataSql" readonly :input-style="{ height: '170px' }" />
-                        </el-form-item>
-                        <el-form-item prop="fieldMap" :label="$t('db.insertSql')">
-                            <el-input type="textarea" v-model="state.previewInsertSql" readonly :input-style="{ height: '170px' }" />
-                        </el-form-item>
                         <el-form-item prop="isReplace" v-if="compatibleDuplicateStrategy(form.targetDbType!)" :label="$t('db.keyDuplicateStrategy')">
                             <EnumSelect :enums="DbDataSyncDuplicateStrategyEnum" v-model="form.duplicateStrategy" @change="handleDuplicateStrategy" />
                         </el-form-item>
+                        <el-form-item prop="fieldMap" :label="$t('db.selectSql')">
+                            <el-input type="textarea" v-model="state.previewDataSql" readonly :input-style="{ height: '300px' }" />
+                        </el-form-item>
+                        <el-form-item prop="fieldMap" :label="$t('db.insertSql')">
+                            <el-input type="textarea" v-model="state.previewInsertSql" readonly :input-style="{ height: '300px' }" />
+                        </el-form-item>
                     </el-tab-pane>
                 </el-tabs>
             </el-form>
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
+}