redo: make row flush batch size configurable

api/v2/changefeed_toml_test.go

@@ -102,6 +102,7 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) {
 				Level:             util.AddressOf("eventual"),
 				MaxLogSize:        util.AddressOf(int64(128)),
 				FlushIntervalInMs: util.AddressOf(int64(2000)),
+				FlushBatchSize:    util.AddressOf(2048),
 				Storage:           util.AddressOf("s3://test"),
 			},
 		},
@@ -131,6 +132,7 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) {
 	require.Equal(t, 10*time.Minute, *wrapper.Config.SyncPointInterval)
 	require.Equal(t, "correctness", util.GetOrZero(wrapper.Config.Integrity.IntegrityCheckLevel))
 	require.Equal(t, "eventual", util.GetOrZero(wrapper.Config.Consistent.Level))
+	require.Equal(t, 2048, util.GetOrZero(wrapper.Config.Consistent.FlushBatchSize))
 }
 
 // TestDefaultConfigTOMLRoundTripToInternal encodes the full default replica

api/v2/model.go

@@ -328,6 +328,9 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 		if c.Consistent.FlushIntervalInMs != nil {
 			res.Consistent.FlushIntervalInMs = c.Consistent.FlushIntervalInMs
 		}
+		if c.Consistent.FlushBatchSize != nil {
+			res.Consistent.FlushBatchSize = c.Consistent.FlushBatchSize
+		}
 		if c.Consistent.MetaFlushIntervalInMs != nil {
 			res.Consistent.MetaFlushIntervalInMs = c.Consistent.MetaFlushIntervalInMs
 		}
@@ -981,6 +984,9 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 		if cloned.Consistent.FlushIntervalInMs != nil {
 			res.Consistent.FlushIntervalInMs = cloned.Consistent.FlushIntervalInMs
 		}
+		if cloned.Consistent.FlushBatchSize != nil {
+			res.Consistent.FlushBatchSize = cloned.Consistent.FlushBatchSize
+		}
 		if cloned.Consistent.MetaFlushIntervalInMs != nil {
 			res.Consistent.MetaFlushIntervalInMs = cloned.Consistent.MetaFlushIntervalInMs
 		}
@@ -1266,9 +1272,11 @@ type ColumnSelector struct {
 // ConsistentConfig represents replication consistency config for a changefeed
 // This is a duplicate of config.ConsistentConfig
 type ConsistentConfig struct {
-	Level                 *string                `json:"level,omitempty" toml:"level,omitempty"`
-	MaxLogSize            *int64                 `json:"max_log_size,omitempty" toml:"max-log-size,omitempty"`
-	FlushIntervalInMs     *int64                 `json:"flush_interval,omitempty" toml:"flush-interval,omitempty"`
+	Level             *string `json:"level,omitempty" toml:"level,omitempty"`
+	MaxLogSize        *int64  `json:"max_log_size,omitempty" toml:"max-log-size,omitempty"`
+	FlushIntervalInMs *int64  `json:"flush_interval,omitempty" toml:"flush-interval,omitempty"`
+	// FlushBatchSize is the row-count flush threshold. Zero disables it.
+	FlushBatchSize        *int                   `json:"flush_batch_size,omitempty" toml:"flush-batch-size,omitempty"`
 	MetaFlushIntervalInMs *int64                 `json:"meta_flush_interval,omitempty" toml:"meta-flush-interval,omitempty"`
 	EncodingWorkerNum     *int                   `json:"encoding_worker_num,omitempty" toml:"encoding-worker-num,omitempty"`
 	FlushWorkerNum        *int                   `json:"flush_worker_num,omitempty" toml:"flush-worker-num,omitempty"`

api/v2/model_test.go

@@ -192,23 +192,29 @@ func TestReplicaConfigConversionBatchFields(t *testing.T) {
 	require.Nil(t, apiNoBatch.EventCollectorBatchBytes)
 }
 
-func TestReplicaConfigConversionRedoBatchField(t *testing.T) {
+// TestReplicaConfigConversionRedoBatchFields verifies redo-specific batch
+// settings survive API-to-internal and internal-to-API conversion unchanged.
+func TestReplicaConfigConversionRedoBatchFields(t *testing.T) {
 	t.Parallel()
 
 	apiCfg := &ReplicaConfig{
 		Consistent: &ConsistentConfig{
 			EventCollectorBatchCount: util.AddressOf(4096),
+			FlushBatchSize:           util.AddressOf(2048),
 		},
 	}
 
 	internalCfg := apiCfg.ToInternalReplicaConfig()
 	require.NotNil(t, internalCfg.Consistent)
 	require.Equal(t, 4096, util.GetOrZero(internalCfg.Consistent.EventCollectorBatchCount))
+	require.Equal(t, 2048, util.GetOrZero(internalCfg.Consistent.FlushBatchSize))
 
 	apiCfgBack := ToAPIReplicaConfig(internalCfg)
 	require.NotNil(t, apiCfgBack.Consistent)
 	require.NotNil(t, apiCfgBack.Consistent.EventCollectorBatchCount)
 	require.Equal(t, 4096, *apiCfgBack.Consistent.EventCollectorBatchCount)
+	require.NotNil(t, apiCfgBack.Consistent.FlushBatchSize)
+	require.Equal(t, 2048, *apiCfgBack.Consistent.FlushBatchSize)
 }
 
 func TestReplicaConfigConversionMySQLAsyncDDLTimeout(t *testing.T) {

downstreamadapter/sink/redo/sink.go

@@ -33,6 +33,10 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+// dmlWriterInputBatchSize bounds each transfer from the unlimited sink buffer
+// into the writer. It is independent of the writer's persistent flush policy.
+const dmlWriterInputBatchSize = 1024
+
 // Sink manages redo log writer, buffers un-persistent redo logs, calculates
 // redo log resolved ts. It implements Sink interface.
 type Sink struct {
@@ -237,7 +241,7 @@ func (s *Sink) Close() {
 }
 
 func (s *Sink) sendMessages(ctx context.Context) error {
-	buffer := make([]*commonEvent.RedoRowEvent, 0, redo.DefaultFlushBatchSize)
+	buffer := make([]*commonEvent.RedoRowEvent, 0, dmlWriterInputBatchSize)
 	for {
 		select {
 		case <-ctx.Done():

downstreamadapter/sink/redo/sink_test.go

@@ -340,6 +340,9 @@ func runBenchTest(b *testing.B, storage string, useFileBackend bool) {
 	require.ErrorIs(b, eg.Wait(), context.Canceled)
 }
 
+// TestRedoSinkSendMessagesInBatch fills the unlimited buffer, drains it through
+// the bounded writer-input batches, and verifies this transport batch remains
+// independent of the configurable persistent flush threshold.
 func TestRedoSinkSendMessagesInBatch(t *testing.T) {
 	t.Parallel()
 
@@ -365,8 +368,8 @@ func TestRedoSinkSendMessagesInBatch(t *testing.T) {
 	}
 
 	gomock.InOrder(
-		expectWriteBatch(redo.DefaultFlushBatchSize),
-		expectWriteBatch(redo.DefaultFlushBatchSize),
+		expectWriteBatch(dmlWriterInputBatchSize),
+		expectWriteBatch(dmlWriterInputBatchSize),
 		expectWriteBatch(17),
 	)
 
@@ -380,7 +383,7 @@ func TestRedoSinkSendMessagesInBatch(t *testing.T) {
 		doneCh <- s.sendMessages(ctx)
 	}()
 
-	totalEvents := redo.DefaultFlushBatchSize*2 + 17
+	totalEvents := dmlWriterInputBatchSize*2 + 17
 	events := make([]*commonEvent.RedoRowEvent, 0, totalEvents)
 	for range totalEvents {
 		events = append(events, &commonEvent.RedoRowEvent{})

pkg/config/consistent.go

@@ -37,6 +37,9 @@ type ConsistentConfig struct {
 	// FlushIntervalInMs is the flush interval(ms) of redo log to flush log to storage.
 	// Default is 2000ms.
 	FlushIntervalInMs *int64 `toml:"flush-interval" json:"flush-interval,omitempty"`
+	// FlushBatchSize is the number of row events that triggers a redo log flush.
+	// A value of 0 disables count-based flushing. Default is 0.
+	FlushBatchSize *int `toml:"flush-batch-size" json:"flush-batch-size,omitempty"`
 	// MetaFlushIntervalInMs is the flush interval(ms) of redo log to
 	// flush meta(resolvedTs and checkpointTs) to storage.
 	// Default is 200ms.
@@ -101,6 +104,13 @@ func (c *ConsistentConfig) validateAndAdjust(enableIOCheck bool) error {
 			fmt.Sprintf("The consistent.flush-interval:%d must be equal or greater than %d",
 				util.GetOrZero(c.FlushIntervalInMs), redo.MinFlushIntervalInMs))
 	}
+	if c.FlushBatchSize == nil {
+		c.FlushBatchSize = util.AddressOf(redo.DefaultFlushBatchSize)
+	}
+	if *c.FlushBatchSize < 0 {
+		return errors.ErrInvalidReplicaConfig.FastGenByArgs(
+			"consistent.flush-batch-size must be set not smaller than 0")
+	}
 
 	if util.GetOrZero(c.MetaFlushIntervalInMs) == 0 {
 		c.MetaFlushIntervalInMs = util.AddressOf(int64(redo.DefaultMetaFlushIntervalInMs))

pkg/config/replica_config.go

@@ -92,6 +92,7 @@ var defaultReplicaConfig = &ReplicaConfig{
 		Level:                 util.AddressOf("none"),
 		MaxLogSize:            util.AddressOf(redo.DefaultMaxLogSize),
 		FlushIntervalInMs:     util.AddressOf(int64(redo.DefaultFlushIntervalInMs)),
+		FlushBatchSize:        util.AddressOf(redo.DefaultFlushBatchSize),
 		MetaFlushIntervalInMs: util.AddressOf(int64(redo.DefaultMetaFlushIntervalInMs)),
 		EncodingWorkerNum:     util.AddressOf(redo.DefaultEncodingWorkerNum),
 		FlushWorkerNum:        util.AddressOf(redo.DefaultFlushWorkerNum),

pkg/config/replica_config_test.go

@@ -242,6 +242,44 @@ func TestReplicaConfigValidateBatchConfig(t *testing.T) {
 	assertBatchConfig(nil, util.AddressOf(-1), "event-collector-batch-bytes")
 }
 
+// TestConsistentFlushBatchSizeValidation verifies that adjustment supplies the
+// disabled-by-default value, preserves non-negative overrides, and rejects a
+// negative row-count threshold before the redo writer is created.
+func TestConsistentFlushBatchSizeValidation(t *testing.T) {
+	require.Equal(t, 0, util.GetOrZero(GetDefaultReplicaConfig().Consistent.FlushBatchSize))
+
+	tests := []struct {
+		name      string
+		value     *int
+		wantValue int
+		wantErr   bool
+	}{
+		{name: "unset uses disabled default", wantValue: 0},
+		{name: "zero disables count based flush", value: util.AddressOf(0), wantValue: 0},
+		{name: "positive value enables count based flush", value: util.AddressOf(2048), wantValue: 2048},
+		{name: "negative value is rejected", value: util.AddressOf(-1), wantErr: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := &ConsistentConfig{
+				Level:          util.AddressOf("eventual"),
+				FlushBatchSize: tt.value,
+				Storage:        util.AddressOf("blackhole://"),
+			}
+
+			err := cfg.validateAndAdjust(false)
+			if tt.wantErr {
+				require.ErrorContains(t, err, "consistent.flush-batch-size")
+				return
+			}
+			require.NoError(t, err)
+			require.NotNil(t, cfg.FlushBatchSize)
+			require.Equal(t, tt.wantValue, *cfg.FlushBatchSize)
+		})
+	}
+}
+
 func TestReplicaConfig_EnableRedoIOCheck_DefaultValue(t *testing.T) {
 	config := GetDefaultReplicaConfig()
 	require.True(t, util.GetOrZero(config.EnableRedoIOCheck))

pkg/redo/config.go

@@ -49,8 +49,8 @@ const (
 	DefaultMetaFlushIntervalInMs = 200
 	// MinFlushIntervalInMs is the minimum flush interval for redo log.
 	MinFlushIntervalInMs = 50
-	// DefaultFlushBatchSize is the default flush batch size for redo log.
-	DefaultFlushBatchSize = 1024
+	// DefaultFlushBatchSize disables count-based redo log flushing by default.
+	DefaultFlushBatchSize = 0
 
 	// DefaultEncodingWorkerNum is the default number of encoding workers.
 	DefaultEncodingWorkerNum = 16

pkg/redo/testutil/config.go

@@ -24,6 +24,7 @@ func NewConsistentConfig(storage string) *config.ConsistentConfig {
 	level := string(redo.ConsistentLevelEventual)
 	maxLogSize := int64(redo.DefaultMaxLogSize)
 	flushIntervalInMs := int64(redo.DefaultFlushIntervalInMs)
+	flushBatchSize := redo.DefaultFlushBatchSize
 	metaFlushIntervalInMs := int64(redo.MinFlushIntervalInMs)
 	encodingWorkerNum := redo.DefaultEncodingWorkerNum
 	flushWorkerNum := redo.DefaultFlushWorkerNum
@@ -35,6 +36,7 @@ func NewConsistentConfig(storage string) *config.ConsistentConfig {
 		MaxLogSize:            util.AddressOf(maxLogSize),
 		Storage:               util.AddressOf(storage),
 		FlushIntervalInMs:     util.AddressOf(flushIntervalInMs),
+		FlushBatchSize:        util.AddressOf(flushBatchSize),
 		MetaFlushIntervalInMs: util.AddressOf(metaFlushIntervalInMs),
 		EncodingWorkerNum:     util.AddressOf(encodingWorkerNum),
 		FlushWorkerNum:        util.AddressOf(flushWorkerNum),

pkg/redo/writer/config.go

@@ -43,6 +43,8 @@ type Config struct {
 
 	// Shared by file and memory backends as the flush ticker interval.
 	flushIntervalInMs int64
+	// Shared by file and memory backends as the optional row-count flush threshold.
+	flushBatchSize int
 
 	// Used only by the memory backend for encoding workers.
 	encodingWorkerNum int
@@ -77,6 +79,7 @@ func NewConfig(changefeedID common.ChangeFeedID, consistentCfg *config.Consisten
 		maxLogSizeInBytes: util.GetOrZero(consistentCfg.MaxLogSize) * redo.Megabyte,
 		useFileBackend:    util.GetOrZero(consistentCfg.UseFileBackend),
 		flushIntervalInMs: util.GetOrZero(consistentCfg.FlushIntervalInMs),
+		flushBatchSize:    util.GetOrZero(consistentCfg.FlushBatchSize),
 		encodingWorkerNum: util.GetOrZero(consistentCfg.EncodingWorkerNum),
 		flushWorkerNum:    util.GetOrZero(consistentCfg.FlushWorkerNum),
 		compression:       util.GetOrZero(consistentCfg.Compression),
@@ -153,6 +156,11 @@ func (cfg *Config) FlushIntervalInMs() int64 {
 	return cfg.flushIntervalInMs
 }
 
+// FlushBatchSize returns the row-count flush threshold. Zero disables it.
+func (cfg *Config) FlushBatchSize() int {
+	return cfg.flushBatchSize
+}
+
 func (cfg *Config) EncodingWorkerNum() int {
 	return cfg.encodingWorkerNum
 }