// newMySQLSink creates a new MySQL sink using schema storage
newMySQLSink creates a new MySQL sink using schema storage
// newMySQLSink creates a new MySQL sink using schema storage
func newMySQLSink(
	ctx context.Context,
	changefeedID model.ChangeFeedID,
	sinkURI *url.URL,
	filter *tifilter.Filter,
	replicaConfig *config.ReplicaConfig,
	opts map[string]string,
) (Sink, error) {
	opts[OptChangefeedID] = changefeedID
	params, err := parseSinkURIToParams(ctx, sinkURI, opts)
	if err != nil {
		return nil, err
	}
	params.enableOldValue = replicaConfig.EnableOldValue
	username := sinkURI.User.Username()
	password, _ := sinkURI.User.Password()
	port := sinkURI.Port()
	if username == "" {
		username = "root"
	}
	if port == "" {
		port = "4000"
	}
	dsnStr := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", username, password, sinkURI.Hostname(), port, params.tls)
	dsn, err := dmysql.ParseDSN(dsnStr)
	if err != nil {
		return nil, cerror.WrapError(cerror.ErrMySQLInvalidConfig, err)
	}
	if dsn.Params == nil {
		dsn.Params = make(map[string]string, 1)
	}
	if params.timezone != "" {
		dsn.Params["time_zone"] = params.timezone
	}
	dsn.Params["readTimeout"] = params.readTimeout
	dsn.Params["writeTimeout"] = params.writeTimeout
	dsn.Params["timeout"] = params.dialTimeout
	testDB, err := GetDBConnImpl(ctx, dsn.FormatDSN())
	if err != nil {
		return nil, err
	}
	defer testDB.Close()
	dsn.Params["sql_mode"], err = querySQLMode(ctx, testDB)
	if err != nil {
		return nil, errors.Trace(err)
	}
	dsn.Params["sql_mode"], err = dmutils.AdjustSQLModeCompatible(dsn.Params["sql_mode"])
	if err != nil {
		return nil, errors.Trace(err)
	}
	var sinkCyclic *cyclic.Cyclic = nil
	if val, ok := opts[mark.OptCyclicConfig]; ok {
		cfg := new(config.CyclicConfig)
		err := cfg.Unmarshal([]byte(val))
		if err != nil {
			return nil, cerror.WrapError(cerror.ErrMySQLInvalidConfig, err)
		}
		sinkCyclic = cyclic.NewCyclic(cfg)
		dsn.Params["sql_mode"] = cyclic.RelaxSQLMode(dsn.Params["sql_mode"])
	}
	dsn.Params["sql_mode"] = strconv.Quote(dsn.Params["sql_mode"])
	dsnStr, err = generateDSNByParams(ctx, dsn, params, testDB)
	if err != nil {
		return nil, errors.Trace(err)
	}
	db, err := GetDBConnImpl(ctx, dsnStr)
	if err != nil {
		return nil, err
	}
	log.Info("Start mysql sink")
	db.SetMaxIdleConns(params.workerCount)
	db.SetMaxOpenConns(params.workerCount)
	metricConflictDetectDurationHis :=
		conflictDetectDurationHis.WithLabelValues(params.changefeedID)
	metricBucketSizeCounters := make([]prometheus.Counter, params.workerCount)
	for i := 0; i < params.workerCount; i++ {
		metricBucketSizeCounters[i] =
			bucketSizeCounter.WithLabelValues(params.changefeedID, strconv.Itoa(i))
	}
	ctx, cancel := context.WithCancel(ctx)
	sink := &mysqlSink{
		db:                              db,
		params:                          params,
		filter:                          filter,
		cyclic:                          sinkCyclic,
		txnCache:                        common.NewUnresolvedTxnCache(),
		statistics:                      NewStatistics(ctx, "mysql"),
		metricConflictDetectDurationHis: metricConflictDetectDurationHis,
		metricBucketSizeCounters:        metricBucketSizeCounters,
		errCh:                           make(chan error, 1),
		forceReplicate:                  replicaConfig.ForceReplicate,
		cancel:                          cancel,
	}
	sink.execWaitNotifier = new(notify.Notifier)
	sink.resolvedNotifier = new(notify.Notifier)
	err = sink.createSinkWorkers(ctx)
	if err != nil {
		return nil, err
	}
	receiver, err := sink.resolvedNotifier.NewReceiver(50 * time.Millisecond)
	if err != nil {
		return nil, err
	}
	go sink.flushRowChangedEvents(ctx, receiver)
	return sink, nil
}
func newMySQLSink ( ctx context . Context , changefeedID model . ChangeFeedID , sinkURI * url . URL , filter * tifilter . Filter , replicaConfig * config . ReplicaConfig , opts map [ string ] string , ) ( Sink , error ) { opts [ OptChangefeedID ] = changefeedID params , err := parseSinkURIToParams ( ctx , sinkURI , opts ) if err != nil { return nil , err } params . enableOldValue = replicaConfig . EnableOldValue username := sinkURI . User . Username ( ) password , _ := sinkURI . User . Password ( ) port := sinkURI . Port ( ) if username == " " { username = " " } if port == " " { port = " " } dsnStr := fmt . Sprintf ( " " , username , password , sinkURI . Hostname ( ) , port , params . tls ) dsn , err := dmysql . ParseDSN ( dsnStr ) if err != nil { return nil , cerror . WrapError ( cerror . ErrMySQLInvalidConfig , err ) } if dsn . Params == nil { dsn . Params = make ( map [ string ] string , 1 ) } if params . timezone != " " { dsn . Params [ " " ] = params . timezone } dsn . Params [ " " ] = params . readTimeout dsn . Params [ " " ] = params . writeTimeout dsn . Params [ " " ] = params . dialTimeout testDB , err := GetDBConnImpl ( ctx , dsn . FormatDSN ( ) ) if err != nil { return nil , err } defer testDB . Close ( ) dsn . Params [ " " ] , err = querySQLMode ( ctx , testDB ) if err != nil { return nil , errors . Trace ( err ) } dsn . Params [ " " ] , err = dmutils . AdjustSQLModeCompatible ( dsn . Params [ " " ] ) if err != nil { return nil , errors . Trace ( err ) } var sinkCyclic * cyclic . Cyclic = nil if val , ok := opts [ mark . OptCyclicConfig ] ; ok { cfg := new ( config . CyclicConfig ) err := cfg . Unmarshal ( [ ] byte ( val ) ) if err != nil { return nil , cerror . WrapError ( cerror . ErrMySQLInvalidConfig , err ) } sinkCyclic = cyclic . NewCyclic ( cfg ) dsn . Params [ " " ] = cyclic . RelaxSQLMode ( dsn . Params [ " " ] ) } dsn . Params [ " " ] = strconv . Quote ( dsn . Params [ " " ] ) dsnStr , err = generateDSNByParams ( ctx , dsn , params , testDB ) if err != nil { return nil , errors . Trace ( err ) } db , err := GetDBConnImpl ( ctx , dsnStr ) if err != nil { return nil , err } log . Info ( " " ) db . SetMaxIdleConns ( params . workerCount ) db . SetMaxOpenConns ( params . workerCount ) metricConflictDetectDurationHis := conflictDetectDurationHis . WithLabelValues ( params . changefeedID ) metricBucketSizeCounters := make ( [ ] prometheus . Counter , params . workerCount ) for i := 0 ; i < params . workerCount ; i ++ { metricBucketSizeCounters [ i ] = bucketSizeCounter . WithLabelValues ( params . changefeedID , strconv . Itoa ( i ) ) } ctx , cancel := context . WithCancel ( ctx ) sink := & mysqlSink { db : db , params : params , filter : filter , cyclic : sinkCyclic , txnCache : common . NewUnresolvedTxnCache ( ) , statistics : NewStatistics ( ctx , " " ) , metricConflictDetectDurationHis : metricConflictDetectDurationHis , metricBucketSizeCounters : metricBucketSizeCounters , errCh : make ( chan error , 1 ) , forceReplicate : replicaConfig . ForceReplicate , cancel : cancel , } sink . execWaitNotifier = new ( notify . Notifier ) sink . resolvedNotifier = new ( notify . Notifier ) err = sink . createSinkWorkers ( ctx ) if err != nil { return nil , err } receiver , err := sink . resolvedNotifier . NewReceiver ( 50 * time . Millisecond ) if err != nil { return nil , err } go sink . flushRowChangedEvents ( ctx , receiver ) return sink , nil }
newMySQLSink creates a new MySQL sink using schema storage
newMySQLSink creates a new MySQL sink using schema storage
// dsn format of the driver: // [username[:password]@][protocol[(address)]]/dbname[?param1=value1&...&paramN=valueN] // create test db used for parameter detection // Refer https://github.com/go-sql-driver/mysql#parameters // Adjust sql_mode for compatibility. // Adjust sql_mode for cyclic replication. // NOTE: quote the string is necessary to avoid ambiguities.