func()

in pkg/datasource/sql/undo/builder/mysql_insertonduplicate_update_undo_log_builder.go [96:162]

• 64 lines of code
• 16 McCabe index (conditional complexity)

func (u *MySQLInsertOnDuplicateUndoLogBuilder) buildBeforeImageSQL(insertStmt *ast.InsertStmt, metaData types.TableMeta, args []driver.Value) (string, []driver.Value, error) {
	if err := checkDuplicateKeyUpdate(insertStmt, metaData); err != nil {
		return "", nil, err
	}
	var selectArgs []driver.Value
	pkIndexMap := u.getPkIndex(insertStmt, metaData)
	var pkIndexArray []int
	for _, val := range pkIndexMap {
		tmpVal := val
		pkIndexArray = append(pkIndexArray, tmpVal)
	}
	insertRows, err := getInsertRows(insertStmt, pkIndexArray)
	if err != nil {
		return "", nil, err
	}
	insertNum := len(insertRows)
	paramMap, err := u.buildImageParameters(insertStmt, args, insertRows)
	if err != nil {
		return "", nil, err
	}

	sql := strings.Builder{}
	sql.WriteString("SELECT * FROM " + metaData.TableName + " ")
	isContainWhere := false
	for i := 0; i < insertNum; i++ {
		finalI := i
		paramAppenderTempList := make([]driver.Value, 0)
		for _, index := range metaData.Indexs {
			//unique index
			if index.NonUnique || isIndexValueNotNull(index, paramMap, finalI) == false {
				continue
			}
			columnIsNull := true
			uniqueList := make([]string, 0)
			for _, columnMeta := range index.Columns {
				columnName := strings.ToLower(columnMeta.ColumnName)
				imageParameters, ok := paramMap[columnName]
				if !ok && columnMeta.ColumnDef != nil {
					if strings.EqualFold("PRIMARY", index.Name) {
						u.BeforeImageSqlPrimaryKeys[columnName] = true
					}
					uniqueList = append(uniqueList, columnName+" = DEFAULT("+columnName+") ")
					columnIsNull = false
					continue
				}
				if strings.EqualFold("PRIMARY", index.Name) {
					u.BeforeImageSqlPrimaryKeys[columnName] = true
				}
				columnIsNull = false
				uniqueList = append(uniqueList, columnName+" = ? ")
				paramAppenderTempList = append(paramAppenderTempList, imageParameters[finalI])
			}

			if !columnIsNull {
				if isContainWhere {
					sql.WriteString(" OR (" + strings.Join(uniqueList, " and ") + ") ")
				} else {
					sql.WriteString(" WHERE (" + strings.Join(uniqueList, " and ") + ") ")
					isContainWhere = true
				}
			}
		}
		selectArgs = append(selectArgs, paramAppenderTempList...)
	}
	log.Infof("build select sql by insert on duplicate sourceQuery, sql {}", sql.String())
	return sql.String(), selectArgs, nil
}