// prepareDMLs converts model.RowChangedEvent list to query string list and args list
prepareDMLs converts model.RowChangedEvent list to query string list and args list
// prepareDMLs converts model . RowChangedEvent list to query string list and args list
func (s *mysqlSink) prepareDMLs(rows []*model.RowChangedEvent, replicaID uint64, bucket int) *preparedDMLs {
	sqls := make([]string, 0, len(rows))
	values := make([][]interface{}, 0, len(rows))
	replaces := make(map[string][][]interface{})
	rowCount := 0
	translateToInsert := s.params.enableOldValue && !s.params.safeMode
	flushCacheDMLs := func() {
		if s.params.batchReplaceEnabled && len(replaces) > 0 {
			replaceSqls, replaceValues := reduceReplace(replaces, s.params.batchReplaceSize)
			sqls = append(sqls, replaceSqls...)
			values = append(values, replaceValues...)
			replaces = make(map[string][][]interface{})
		}
	}
	for _, row := range rows {
		var query string
		var args []interface{}
		quoteTable := quotes.QuoteSchema(row.Table.Schema, row.Table.Table)
		if translateToInsert && len(row.PreColumns) != 0 && len(row.Columns) != 0 {
			flushCacheDMLs()
			query, args = prepareUpdate(quoteTable, row.PreColumns, row.Columns, s.forceReplicate)
			if query != "" {
				sqls = append(sqls, query)
				values = append(values, args)
				rowCount++
			}
			continue
		}
		if len(row.PreColumns) != 0 {
			flushCacheDMLs()
			query, args = prepareDelete(quoteTable, row.PreColumns, s.forceReplicate)
			if query != "" {
				sqls = append(sqls, query)
				values = append(values, args)
				rowCount++
			}
		}
		if len(row.Columns) != 0 {
			if s.params.batchReplaceEnabled {
				query, args = prepareReplace(quoteTable, row.Columns, false , translateToInsert)
				if query != "" {
					if _, ok := replaces[query]; !ok {
						replaces[query] = make([][]interface{}, 0)
					}
					replaces[query] = append(replaces[query], args)
					rowCount++
				}
			} else {
				query, args = prepareReplace(quoteTable, row.Columns, true , translateToInsert)
				if query != "" {
					sqls = append(sqls, query)
					values = append(values, args)
					rowCount++
				}
			}
		}
	}
	flushCacheDMLs()
	dmls := &preparedDMLs{
		sqls:   sqls,
		values: values,
	}
	if s.cyclic != nil && len(rows) > 0 {
		row := rows[0]
		updateMark := s.cyclic.UdpateSourceTableCyclicMark(
			row.Table.Schema, row.Table.Table, uint64(bucket), replicaID, row.StartTs)
		dmls.markSQL = updateMark
	}
	dmls.rowCount = rowCount
	return dmls
}
func ( s * mysqlSink ) prepareDMLs ( rows [ ] * model . RowChangedEvent , replicaID uint64 , bucket int ) * preparedDMLs { sqls := make ( [ ] string , 0 , len ( rows ) ) values := make ( [ ] [ ] interface { } , 0 , len ( rows ) ) replaces := make ( map [ string ] [ ] [ ] interface { } ) rowCount := 0 translateToInsert := s . params . enableOldValue && ! s . params . safeMode flushCacheDMLs := func ( ) { if s . params . batchReplaceEnabled && len ( replaces ) > 0 { replaceSqls , replaceValues := reduceReplace ( replaces , s . params . batchReplaceSize ) sqls = append ( sqls , replaceSqls ... ) values = append ( values , replaceValues ... ) replaces = make ( map [ string ] [ ] [ ] interface { } ) } } for _ , row := range rows { var query string var args [ ] interface { } quoteTable := quotes . QuoteSchema ( row . Table . Schema , row . Table . Table ) if translateToInsert && len ( row . PreColumns ) != 0 && len ( row . Columns ) != 0 { flushCacheDMLs ( ) query , args = prepareUpdate ( quoteTable , row . PreColumns , row . Columns , s . forceReplicate ) if query != " " { sqls = append ( sqls , query ) values = append ( values , args ) rowCount ++ } continue } if len ( row . PreColumns ) != 0 { flushCacheDMLs ( ) query , args = prepareDelete ( quoteTable , row . PreColumns , s . forceReplicate ) if query != " " { sqls = append ( sqls , query ) values = append ( values , args ) rowCount ++ } } if len ( row . Columns ) != 0 { if s . params . batchReplaceEnabled { query , args = prepareReplace ( quoteTable , row . Columns , false , translateToInsert ) if query != " " { if _ , ok := replaces [ query ] ; ! ok { replaces [ query ] = make ( [ ] [ ] interface { } , 0 ) } replaces [ query ] = append ( replaces [ query ] , args ) rowCount ++ } } else { query , args = prepareReplace ( quoteTable , row . Columns , true , translateToInsert ) if query != " " { sqls = append ( sqls , query ) values = append ( values , args ) rowCount ++ } } } } flushCacheDMLs ( ) dmls := & preparedDMLs { sqls : sqls , values : values , } if s . cyclic != nil && len ( rows ) > 0 { row := rows [ 0 ] updateMark := s . cyclic . UdpateSourceTableCyclicMark ( row . Table . Schema , row . Table . Table , uint64 ( bucket ) , replicaID , row . StartTs ) dmls . markSQL = updateMark } dmls . rowCount = rowCount return dmls }
prepareDMLs converts model.RowChangedEvent list to query string list and args list
prepareDMLs converts model . RowChangedEvent list to query string list and args list
// translateToInsert control the update and insert behavior // flush cached batch replace or insert, to keep the sequence of DMLs // If the old value is enabled, is not in safe mode and is an update event, then translate to UPDATE. // NOTICE: Only update events with the old value feature enabled will have both columns and preColumns. // Case for update event or delete event. // For update event: // If old value is disabled or in safe mode, update will be translated to DELETE + REPLACE SQL. // So we will prepare a DELETE SQL here. // For delete event: // It will be translated directly into a DELETE SQL. // Case for update event or insert event. // For update event: // If old value is disabled or in safe mode, update will be translated to DELETE + REPLACE SQL. // So we will prepare a REPLACE SQL here. // For insert event: // It will be translated directly into a // INSERT(old value is enabled and not in safe mode) // or REPLACE(old value is disabled or in safe mode) SQL. /* appendPlaceHolder */ /* appendPlaceHolder */ // Write mark table with the current replica ID. // rowCount is used in statistics, and for simplicity, // we do not count mark table rows in rowCount.