sole add switch for scanwindow

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -52,6 +52,7 @@ type DispatcherService interface {
 	GetFilterConfig() *eventpb.FilterConfig
 	EnableSyncPoint() bool
 	GetSyncPointInterval() time.Duration
+	GetEnableScanWindow() bool
 	GetSkipSyncpointAtStartTs() bool
 	GetTxnAtomicity() config.AtomicityLevel
 	GetResolvedTs() uint64

downstreamadapter/dispatcher/basic_dispatcher_info.go

@@ -56,6 +56,10 @@ type SharedInfo struct {
 	// will break the splittability of this table.
 	enableSplittableCheck bool
 
+	// enableScanWindow controls whether the event service applies the adaptive scan
+	// window (memory control + adaptive scan interval) for this changefeed.
+	enableScanWindow bool
+
 	// router is used to route source schema/table names to target schema/table names.
 	// It is used to apply routing to TableInfo before storing it.
 	router routing.Router
@@ -96,6 +100,7 @@ func NewSharedInfo(
 	syncPointConfig *syncpoint.SyncPointConfig,
 	txnAtomicity *config.AtomicityLevel,
 	enableSplittableCheck bool,
+	enableScanWindow bool,
 	router routing.Router,
 	eventCollectorBatchCount int,
 	eventCollectorBatchBytes int,
@@ -113,6 +118,7 @@ func NewSharedInfo(
 		filterConfig:             filterConfig,
 		syncPointConfig:          syncPointConfig,
 		enableSplittableCheck:    enableSplittableCheck,
+		enableScanWindow:         enableScanWindow,
 		router:                   router,
 		eventCollectorBatchCount: eventCollectorBatchCount,
 		eventCollectorBatchBytes: eventCollectorBatchBytes,
@@ -234,6 +240,10 @@ func (d *BasicDispatcher) GetSyncPointInterval() time.Duration {
 	return time.Duration(0)
 }
 
+func (d *BasicDispatcher) GetEnableScanWindow() bool {
+	return d.sharedInfo.enableScanWindow
+}
+
 func (d *BasicDispatcher) GetTableSpan() *heartbeatpb.TableSpan {
 	return d.tableSpan
 }

downstreamadapter/dispatcher/event_dispatcher_test.go

@@ -76,6 +76,7 @@ func newTestSharedInfo(
 		syncPointConfig,
 		&defaultAtomicity,
 		enableSplittableCheck,
+		false, // enableScanWindow
 		routing.Router{},
 		0,
 		0,

downstreamadapter/dispatchermanager/dispatcher_manager.go

@@ -324,6 +324,7 @@ func NewDispatcherManager(
 		syncPointConfig,
 		manager.config.SinkConfig.TxnAtomicity,
 		manager.config.EnableSplittableCheck,
+		manager.config.EnableScanWindow,
 		router,
 		batchCounts,
 		batchBytes,

downstreamadapter/dispatchermanager/dispatcher_manager_test.go

@@ -133,6 +133,7 @@ func createTestManager(t *testing.T) *DispatcherManager {
 		nil,   // syncPointConfig
 		&defaultAtomicity,
 		false,
+		false, // enableScanWindow
 		routing.Router{},
 		0,
 		0,

downstreamadapter/eventcollector/dispatcher_session.go

@@ -533,6 +533,7 @@ func (s *dispatcherSession) newDispatcherRegisterRequest(serverID string, onlyRe
 			Integrity:            s.target.GetIntegrityConfig(),
 			OutputRawChangeEvent: s.target.IsOutputRawChangeEvent(),
 			TxnAtomicity:         string(s.target.GetTxnAtomicity()),
+			EnableScanWindow:     s.target.GetEnableScanWindow(),
 		},
 	}
 }
@@ -568,6 +569,7 @@ func (s *dispatcherSession) newDispatcherResetRequest(serverID string, resetTs u
 			Integrity:            s.target.GetIntegrityConfig(),
 			OutputRawChangeEvent: s.target.IsOutputRawChangeEvent(),
 			TxnAtomicity:         string(s.target.GetTxnAtomicity()),
+			EnableScanWindow:     s.target.GetEnableScanWindow(),
 		},
 	}
 }

downstreamadapter/eventcollector/dispatcher_stat_test.go

@@ -118,6 +118,10 @@ func (m *mockDispatcher) GetSyncPointInterval() time.Duration {
 	return time.Second * 10
 }
 
+func (m *mockDispatcher) GetEnableScanWindow() bool {
+	return false
+}
+
 func (m *mockDispatcher) GetSkipSyncpointAtStartTs() bool {
 	return m.skipSyncpointAtStartTs
 }

downstreamadapter/eventcollector/event_collector.go

@@ -40,6 +40,11 @@ const (
 	receiveChanSize               = 1024 * 8
 	commonMsgRetryQuota           = 3 // The number of retries for most droppable dispatcher requests.
 	eventServiceHeartbeatInterval = 10 * time.Second
+
+	// scanWindowReleaseMemoryRatio is the fraction of pending memory the event collector
+	// releases per cycle for changefeeds with the adaptive scan window enabled. It is
+	// more aggressive than the dynstream default to keep up with the larger scan windows.
+	scanWindowReleaseMemoryRatio = 0.6
 )
 
 // DispatcherMessage is the message send to EventService.
@@ -281,6 +286,12 @@ func (c *EventCollector) PrepareAddDispatcher(
 		batchCount,
 		batchBytes,
 	)
+	// The adaptive scan window releases memory more aggressively to keep up with the
+	// larger scan windows. Only apply it when the changefeed enables the scan window,
+	// so a changefeed with the feature off keeps the pre-scan-window release ratio.
+	if target.GetEnableScanWindow() {
+		areaSetting.SetReleaseMemoryRatio(scanWindowReleaseMemoryRatio)
+	}
 	err := ds.AddPath(target.GetId(), stat, areaSetting)
 	if err != nil {
 		log.Warn("add dispatcher to dynamic stream failed", zap.Error(err))

downstreamadapter/eventcollector/event_collector_test.go

@@ -105,6 +105,10 @@ func (m *mockEventDispatcher) GetSyncPointInterval() time.Duration {
 	return time.Second
 }
 
+func (m *mockEventDispatcher) GetEnableScanWindow() bool {
+	return false
+}
+
 func (m *mockEventDispatcher) GetSkipSyncpointAtStartTs() bool {
 	return false
 }