scheduler: Introduce new scheduling algorithm to dynmaicly balance spliited table by traffic (pingcap#1468)

close pingcap#1409

Author: hongyunyan

api/v2/model.go

@@ -514,7 +514,6 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 			RegionThreshold:            c.Scheduler.RegionThreshold,
 			RegionCountPerSpan:         c.Scheduler.RegionCountPerSpan,
 			WriteKeyThreshold:          c.Scheduler.WriteKeyThreshold,
-			SplitNumberPerNode:         c.Scheduler.SplitNumberPerNode,
 			SchedulingTaskCountPerNode: c.Scheduler.SchedulingTaskCountPerNode,
 		}
 	}
@@ -840,7 +839,6 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 			RegionThreshold:            cloned.Scheduler.RegionThreshold,
 			RegionCountPerSpan:         cloned.Scheduler.RegionCountPerSpan,
 			WriteKeyThreshold:          cloned.Scheduler.WriteKeyThreshold,
-			SplitNumberPerNode:         cloned.Scheduler.SplitNumberPerNode,
 			SchedulingTaskCountPerNode: cloned.Scheduler.SchedulingTaskCountPerNode,
 		}
 	}
@@ -1053,8 +1051,6 @@ type ChangefeedSchedulerConfig struct {
 	RegionCountPerSpan int `toml:"region-count-per-span" json:"region-count-per-span"`
 	// WriteKeyThreshold is the written keys threshold of splitting a table.
 	WriteKeyThreshold int `toml:"write_key_threshold" json:"write_key_threshold"`
-	// SplitNumberPerNode is the number of splits per node.
-	SplitNumberPerNode int `toml:"split_number_per_node" json:"split_number_per_node"`
 	// SchedulingTaskCountPerNode is the upper limit for scheduling tasks each node.
 	SchedulingTaskCountPerNode int `toml:"scheduling-task-count-per-node" json:"scheduling-task-per-node"`
 }

cmd/kafka-consumer/event_group.go

@@ -82,6 +82,5 @@ func (g *eventsGroup) Resolve(resolve uint64) []*commonEvent.DMLEvent {
 			zap.Int("resolved", len(result)), zap.Int("remained", len(g.events)),
 			zap.Uint64("resolveTs", resolve), zap.Uint64("firstCommitTs", g.events[0].CommitTs))
 	}
-	g.highWatermark = resolve
 	return result
 }

cmd/pulsar-consumer/event_group.go

@@ -82,6 +82,5 @@ func (g *eventsGroup) Resolve(resolve uint64) []*commonEvent.DMLEvent {
 			zap.Int("resolved", len(result)), zap.Int("remained", len(g.events)),
 			zap.Uint64("resolveTs", resolve), zap.Uint64("firstCommitTs", g.events[0].CommitTs))
 	}
-	g.highWatermark = resolve
 	return result
 }

coordinator/controller.go

@@ -233,7 +233,7 @@ func (c *Controller) onMessage(msg *messaging.TargetMessage) {
 }
 
 func (c *Controller) onNodeChanged() {
-	currentNodes := c.bootstrapper.GetAllNodes()
+	currentNodes := c.bootstrapper.GetAllNodeIDs()
 
 	activeNodes := c.nodeManager.GetAliveNodes()
 	newNodes := make([]*node.Info, 0, len(activeNodes))

coordinator/operator/operator_add.go

@@ -116,3 +116,8 @@ func (m *AddMaintainerOperator) String() string {
 func (m *AddMaintainerOperator) Type() string {
 	return "add"
 }
+
+func (m *AddMaintainerOperator) BlockTsForward() bool {
+	log.Panic("unreachable code")
+	return false
+}

coordinator/operator/operator_move.go

@@ -184,3 +184,8 @@ func (m *MoveMaintainerOperator) String() string {
 func (m *MoveMaintainerOperator) Type() string {
 	return "move"
 }
+
+func (m *MoveMaintainerOperator) BlockTsForward() bool {
+	log.Panic("unreachable code")
+	return false
+}

coordinator/operator/operator_stop.go

@@ -126,3 +126,8 @@ func (m *StopChangefeedOperator) String() string {
 func (m *StopChangefeedOperator) Type() string {
 	return "stop"
 }
+
+func (m *StopChangefeedOperator) BlockTsForward() bool {
+	log.Panic("unreachable code")
+	return false
+}

downstreamadapter/dispatcher/table_progress.go

@@ -163,7 +163,7 @@ func (p *TableProgress) GetEventSizePerSecond() float32 {
 	if eventSizePerSecond == 0 {
 		// The event size will only send to maintainer once per second.
 		// So if no data is write, we use a tiny value instead of 0 to distinguish it from the status without eventSize
-		return 0.1
+		return 1
 	}
 
 	return eventSizePerSecond

downstreamadapter/eventcollector/helper.go

@@ -24,7 +24,7 @@ import (
 
 func NewEventDynamicStream(collector *EventCollector) dynstream.DynamicStream[common.GID, common.DispatcherID, dispatcher.DispatcherEvent, *dispatcherStat, *EventsHandler] {
 	option := dynstream.NewOption()
-	option.BatchCount = 67136
+	option.BatchCount = 4096
 	option.UseBuffer = false
 	// Enable memory control for dispatcher events dynamic stream.
 	option.EnableMemoryControl = true

maintainer/barrier.go

@@ -14,6 +14,7 @@
 package maintainer
 
 import (
+	"math"
 	"sync"
 
 	"github.com/pingcap/log"
@@ -273,19 +274,16 @@ func (b *Barrier) Resend() []*messaging.TargetMessage {
 	return msgs
 }
 
-// ShouldBlockCheckpointTs returns ture if there is a block event need block the checkpoint ts forwarding
-// currently, when the block event is a create table event, we should block the checkpoint ts forwarding
-// because on the complete checkpointTs calculation should consider the new dispatcher.
-func (b *Barrier) ShouldBlockCheckpointTs() bool {
-	flag := false
-	b.blockedEvents.RangeWoLock(func(key eventKey, barrierEvent *BarrierEvent) bool {
-		if barrierEvent.hasNewTable {
-			flag = true
-			return false
+// GetMinBlockedCheckpointTsForNewTables returns the minimum checkpoint ts for the new tables
+func (b *Barrier) GetMinBlockedCheckpointTsForNewTables() uint64 {
+	minCheckpointTs := uint64(math.MaxUint64)
+	b.blockedEvents.Range(func(key eventKey, barrierEvent *BarrierEvent) bool {
+		if barrierEvent.hasNewTable && minCheckpointTs > barrierEvent.commitTs {
+			minCheckpointTs = barrierEvent.commitTs
 		}
 		return true
 	})
-	return flag
+	return minCheckpointTs
 }
 
 func (b *Barrier) handleOneStatus(changefeedID *heartbeatpb.ChangefeedID, status *heartbeatpb.TableSpanBlockStatus) (*BarrierEvent, *heartbeatpb.DispatcherStatus) {