TRKJ/mayfly-go
2
0
Fork 0
mirror of https://gitee.com/dromara/mayfly-go synced 2025-11-02 15:30:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

!81 fix: 数据同步相关bug修复

Browse Source 
Merge pull request !81 from zongyangleo/dev_0109
This commit is contained in:
Coder慌
2024-01-09 04:12:53 +00:00
committed by Gitee
parent af454f7d5d a764c4f974
commit d3d26c85c3
7 changed files with 37 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
mayfly_go_web/src/views/ops/db/SyncTaskEdit.vue
View File

@@ -278,7 +278,6 @@ watch(tabActiveName, async (newValue: string) => {
await handleGetTargetFields();
break;
case 'targetDb':
await handleGetSrcFields();
await handleGetTargetFields();
if (state.form.targetDbId && state.form.targetDbName) {
await loadDbTables(state.form.targetDbId, state.form.targetDbName);

16
mayfly_go_web/src/views/ops/db/db.ts
View File

@@ -6,7 +6,7 @@ import * as monaco from 'monaco-editor/esm/vs/editor/editor.api';
import { editor, languages, Position } from 'monaco-editor';
import { registerCompletionItemProvider } from '@/components/monaco/completionItemProvider';
import { EditorCompletionItem, getDbDialect } from './dialect';
import {DbDialect, EditorCompletionItem, getDbDialect} from './dialect'
const dbInstCache: Map<number, DbInst> = new Map();
@@ -91,7 +91,7 @@ export class DbInst {
return tables;
}
async loadTableSuggestions(dbName: string, range: any, reload?: boolean) {
async loadTableSuggestions(dbDialect: DbDialect, dbName: string, range: any, reload?: boolean) {
const tables = await this.loadTables(dbName, reload);
// 表名联想
let suggestions: languages.CompletionItem[] = [];
@@ -104,7 +104,7 @@ export class DbInst {
},
kind: monaco.languages.CompletionItemKind.File,
detail: tableComment,
insertText: tableName + ' ',
insertText: dbDialect.wrapName(tableName) + ' ',
range,
sortText: 300 + index + '',
});
@@ -113,7 +113,7 @@ export class DbInst {
}
/** 加载列信息提示 */
async loadTableColumnSuggestions(db: string, tableName: string, range: any) {
async loadTableColumnSuggestions(dbDialect: DbDialect,db: string, tableName: string, range: any) {
let dbHits = await this.loadDbHints(db);
let columns = dbHits[tableName];
let suggestions: languages.CompletionItem[] = [];
@@ -128,7 +128,7 @@ export class DbInst {
},
kind: monaco.languages.CompletionItemKind.Property,
detail: '', // 不显示detail, 否则选中时备注等会被遮挡
insertText: fieldName, // create_time
insertText: dbDialect.wrapName(fieldName)+ ' ', // create_time
range,
sortText: 100 + index + '', // 使用表字段声明顺序排序,排序需为字符串类型
});
@@ -651,20 +651,20 @@ export function registerDbCompletionItemProvider(dbId: number, db: string, dbs:
if (db.indexOf('/') > 0) {
dbName = db.substring(0, db.indexOf('/') + 1) + alias;
}
return await dbInst.loadTableSuggestions(dbName, range);
return await dbInst.loadTableSuggestions(dbDialect, dbName, range);
}
// 表下列名联想 .前的字符串是表名或表别名
const sqlInfo = getTableName4SqlCtx(sqlStatement, alias, db);
// 提出到表名，则将表对应的字段也添加进提示建议
if (sqlInfo) {
return await dbInst.loadTableColumnSuggestions(sqlInfo.db, sqlInfo.tableName, range);
return await dbInst.loadTableColumnSuggestions(dbDialect, sqlInfo.db, sqlInfo.tableName, range);
}
}
// 空格触发也会提示字段信息
const sqlInfo = getTableName4SqlCtx(sqlStatement, alias, db);
if (sqlInfo) {
const columnSuggestions = await dbInst.loadTableColumnSuggestions(sqlInfo.db, sqlInfo.tableName, range);
const columnSuggestions = await dbInst.loadTableColumnSuggestions(dbDialect, sqlInfo.db, sqlInfo.tableName, range);
suggestions.push(...columnSuggestions.suggestions);
}

2
mayfly_go_web/src/views/ops/db/dialect/postgres_dialect.ts
View File

@@ -203,7 +203,7 @@ class PostgresqlDialect implements DbDialect {
}
wrapName = (name: string) => {
return name;
return `"${name}"`;
};
matchType(text: string, arr: string[]): boolean {

12
server/internal/db/dbm/dialect_dm.go
View File

@@ -278,11 +278,11 @@ func (dd *DMDialect) GetDbProgram() DbProgram {
panic("implement me")
}
func (pd *DMDialect) WrapName(name string) string {
func (dd *DMDialect) WrapName(name string) string {
return "\"" + name + "\""
}
func (pd *DMDialect) GetDataType(dbColumnType string) DataType {
func (dd *DMDialect) GetDataType(dbColumnType string) DataType {
if regexp.MustCompile(`(?i)int|double|float|number|decimal|byte|bit`).MatchString(dbColumnType) {
return DataTypeNumber
}
@@ -301,7 +301,7 @@ func (pd *DMDialect) GetDataType(dbColumnType string) DataType {
return DataTypeString
}
func (pd *DMDialect) BatchInsert(tx *sql.Tx, tableName string, columns []string, values [][]any) (int64, error) {
func (dd *DMDialect) BatchInsert(tx *sql.Tx, tableName string, columns []string, values [][]any) (int64, error) {
// 执行批量insert sql
// insert into "table_name" ("column1", "column2", ...) values (value1, value2, ...)
@@ -311,11 +311,11 @@ func (pd *DMDialect) BatchInsert(tx *sql.Tx, tableName string, columns []string,
// 去除最后一个逗号，占位符由括号包裹
placeholder := fmt.Sprintf("(%s)", strings.TrimSuffix(repeated, ","))
sqlTemp := fmt.Sprintf("insert into %s (%s) values %s", pd.WrapName(tableName), strings.Join(columns, ","), placeholder)
sqlTemp := fmt.Sprintf("insert into %s (%s) values %s", dd.WrapName(tableName), strings.Join(columns, ","), placeholder)
effRows := 0
for _, value := range values {
// 达梦数据库只能一条条的执行insert
er, err := pd.dc.TxExec(tx, sqlTemp, value...)
er, err := dd.dc.TxExec(tx, sqlTemp, value...)
if err != nil {
logx.Errorf("执行sql失败：%s", err.Error())
return int64(effRows), err
@@ -326,7 +326,7 @@ func (pd *DMDialect) BatchInsert(tx *sql.Tx, tableName string, columns []string,
return int64(effRows), nil
}
func (pd *DMDialect) FormatStrData(dbColumnValue string, dataType DataType) string {
func (dd *DMDialect) FormatStrData(dbColumnValue string, dataType DataType) string {
switch dataType {
case DataTypeDateTime: // "2024-01-02T22:08:22.275697+08:00"
res, _ := time.Parse(time.RFC3339, dbColumnValue)

12
server/internal/db/dbm/dialect_mysql.go
View File

@@ -202,11 +202,11 @@ func (md *MysqlDialect) GetDbProgram() DbProgram {
return NewDbProgramMysql(md.dc)
}
func (pd *MysqlDialect) WrapName(name string) string {
func (md *MysqlDialect) WrapName(name string) string {
return "`" + name + "`"
}
func (pd *MysqlDialect) GetDataType(dbColumnType string) DataType {
func (md *MysqlDialect) GetDataType(dbColumnType string) DataType {
if regexp.MustCompile(`(?i)int|double|float|number|decimal|byte|bit`).MatchString(dbColumnType) {
return DataTypeNumber
}
@@ -225,7 +225,7 @@ func (pd *MysqlDialect) GetDataType(dbColumnType string) DataType {
return DataTypeString
}
func (pd *MysqlDialect) BatchInsert(tx *sql.Tx, tableName string, columns []string, values [][]any) (int64, error) {
func (md *MysqlDialect) BatchInsert(tx *sql.Tx, tableName string, columns []string, values [][]any) (int64, error) {
// 生成占位符字符串：如：(?,?)
// 重复字符串并用逗号连接
repeated := strings.Repeat("?,", len(columns))
@@ -240,17 +240,17 @@ func (pd *MysqlDialect) BatchInsert(tx *sql.Tx, tableName string, columns []stri
// 去除最后一个逗号
placeholder = strings.TrimSuffix(repeated, ",")
sqlStr := fmt.Sprintf("insert into %s (%s) values %s", pd.WrapName(tableName), strings.Join(columns, ","), placeholder)
sqlStr := fmt.Sprintf("insert into %s (%s) values %s", md.WrapName(tableName), strings.Join(columns, ","), placeholder)
// 执行批量insert sql
// 把二维数组转为一维数组
var args []any
for _, v := range values {
args = append(args, v...)
}
return pd.dc.TxExec(tx, sqlStr, args...)
return md.dc.TxExec(tx, sqlStr, args...)
}
func (pd *MysqlDialect) FormatStrData(dbColumnValue string, dataType DataType) string {
func (md *MysqlDialect) FormatStrData(dbColumnValue string, dataType DataType) string {
// mysql不需要格式化时间日期等
return dbColumnValue
}

31
server/internal/db/dbm/dialect_pgsql.go
View File

@@ -281,7 +281,7 @@ func (pd *PgsqlDialect) GetDbProgram() DbProgram {
}
func (pd *PgsqlDialect) WrapName(name string) string {
return name
return fmt.Sprintf(`"%s"`, name)
}
func (pd *PgsqlDialect) GetDataType(dbColumnType string) DataType {
@@ -307,28 +307,25 @@ func (pd *PgsqlDialect) BatchInsert(tx *sql.Tx, tableName string, columns []stri
// 执行批量insert sql，跟mysql一样 pg或高斯支持批量insert语法
// insert into table_name (column1, column2, ...) values (value1, value2, ...), (value1, value2, ...), ...
// 生成占位符字符串：如：(?,?)
// 重复字符串并用逗号连接
repeated := strings.Repeat("?,", len(columns))
// 去除最后一个逗号，占位符由括号包裹
placeholder := fmt.Sprintf("(%s)", strings.TrimSuffix(repeated, ","))
// 执行批量insert sql，mysql支持批量insert语法
// insert into table_name (column1, column2, ...) values (value1, value2, ...), (value1, value2, ...), ...
// 重复占位符字符串n遍
repeated = strings.Repeat(placeholder+",", len(values))
// 去除最后一个逗号
placeholder = strings.TrimSuffix(repeated, ",")
sqlStr := fmt.Sprintf("insert into %s (%s) values %s", pd.WrapName(tableName), strings.Join(columns, ","), placeholder)
// 执行批量insert sql
// 把二维数组转为一维数组
var args []any
for _, v := range values {
args = append(args, v...)
}
// 构建占位符字符串 "($1, $2, $3), ($4, $5, $6), ..." 用于指定参数
var placeholders []string
for i := 0; i < len(args); i += len(columns) {
var placeholder []string
for j := 0; j < len(columns); j++ {
placeholder = append(placeholder, fmt.Sprintf("$%d", i+j+1))
}
placeholders = append(placeholders, "("+strings.Join(placeholder, ", ")+")")
}
sqlStr := fmt.Sprintf("insert into %s (%s) values %s", pd.WrapName(tableName), strings.Join(columns, ","), strings.Join(placeholders, ", "))
// 执行批量insert sql
return pd.dc.TxExec(tx, sqlStr, args...)
}

3
server/internal/db/infrastructure/persistence/db_data_sync.go
View File

@@ -19,7 +19,8 @@ func newDataSyncTaskRepo() repository.DataSyncTask {
// 分页获取数据库信息列表
func (d *dataSyncTaskRepoImpl) GetTaskList(condition *entity.DataSyncTaskQuery, pageParam *model.PageParam, toEntity any, orderBy ...string) (*model.PageResult[any], error) {
qd := gormx.NewQuery(new(entity.DataSyncTask)).
Like("task_name", condition.Name)
Like("task_name", condition.Name).
Eq("status", condition.Status)
return gormx.PageQuery(qd, pageParam, toEntity)
}