syncpoint: revert syncpoint batch operation (#3235)

close #3234

Author: hongyunyan

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -493,7 +493,7 @@ func (d *BasicDispatcher) handleEvents(dispatcherEvents []DispatcherEvent, wakeC
 			syncPoint := event.(*commonEvent.SyncPointEvent)
 			log.Info("dispatcher receive sync point event",
 				zap.Stringer("dispatcher", d.id),
-				zap.Any("commitTsList", syncPoint.GetCommitTsList()),
+				zap.Uint64("commitTs", syncPoint.GetCommitTs()),
 				zap.Uint64("seq", event.GetSeq()))
 
 			syncPoint.AddPostFlushFunc(func() {
@@ -700,66 +700,26 @@ func (d *BasicDispatcher) DealWithBlockEvent(event commonEvent.BlockEvent) {
 	} else {
 		d.blockEventStatus.setBlockEvent(event, heartbeatpb.BlockStage_WAITING)
 
-		if event.GetType() == commonEvent.TypeSyncPointEvent {
-			// deal with multi sync point commit ts in one Sync Point Event
-			// we only report the latest commitTs as the blockTs.
-			// If A receive [syncpont1, syncpoint2, syncpoint3]
-			// B receive [syncpoint1]
-			// C receive [syncpoint1, syncpoint2]
-			// then A report syncpoint3, B report syncpoint1, C report syncpoint2
-			// and barrier find A checkpointTs is exceed syncpoint1 and syncpoint2,
-			// so will just make B and C pass these syncpoint, to receive the latest syncpoint3
-			// then make syncpoint3 Write successfully.
-
-			// TODO(hyy): we could consider to just use the latest commitTs to represent this batch sync point event,
-			// instead of obtain a commitTsList in the sync point event.
-			commitTsList := event.(*commonEvent.SyncPointEvent).GetCommitTsList()
-			blockTables := event.GetBlockedTables().ToPB()
-			needDroppedTables := event.GetNeedDroppedTables().ToPB()
-			needAddedTables := commonEvent.ToTablesPB(event.GetNeedAddedTables())
-			commitTs := commitTsList[len(commitTsList)-1]
-			message := &heartbeatpb.TableSpanBlockStatus{
-				ID: d.id.ToPB(),
-				State: &heartbeatpb.State{
-					IsBlocked:         true,
-					BlockTs:           commitTs,
-					BlockTables:       blockTables,
-					NeedDroppedTables: needDroppedTables,
-					NeedAddedTables:   needAddedTables,
-					UpdatedSchemas:    nil,
-					IsSyncPoint:       true,
-					Stage:             heartbeatpb.BlockStage_WAITING,
-				},
-				Mode: d.GetMode(),
-			}
-			identifier := BlockEventIdentifier{
-				CommitTs:    commitTs,
-				IsSyncPoint: true,
-			}
-			d.resendTaskMap.Set(identifier, newResendTask(message, d, nil))
-			d.sharedInfo.blockStatusesChan <- message
-		} else {
-			message := &heartbeatpb.TableSpanBlockStatus{
-				ID: d.id.ToPB(),
-				State: &heartbeatpb.State{
-					IsBlocked:         true,
-					BlockTs:           event.GetCommitTs(),
-					BlockTables:       event.GetBlockedTables().ToPB(),
-					NeedDroppedTables: event.GetNeedDroppedTables().ToPB(),
-					NeedAddedTables:   commonEvent.ToTablesPB(event.GetNeedAddedTables()),
-					UpdatedSchemas:    commonEvent.ToSchemaIDChangePB(event.GetUpdatedSchemas()), // only exists for rename table and rename tables
-					IsSyncPoint:       false,
-					Stage:             heartbeatpb.BlockStage_WAITING,
-				},
-				Mode: d.GetMode(),
-			}
-			identifier := BlockEventIdentifier{
-				CommitTs:    event.GetCommitTs(),
-				IsSyncPoint: false,
-			}
-			d.resendTaskMap.Set(identifier, newResendTask(message, d, nil))
-			d.sharedInfo.blockStatusesChan <- message
+		message := &heartbeatpb.TableSpanBlockStatus{
+			ID: d.id.ToPB(),
+			State: &heartbeatpb.State{
+				IsBlocked:         true,
+				BlockTs:           event.GetCommitTs(),
+				BlockTables:       event.GetBlockedTables().ToPB(),
+				NeedDroppedTables: event.GetNeedDroppedTables().ToPB(),
+				NeedAddedTables:   commonEvent.ToTablesPB(event.GetNeedAddedTables()),
+				UpdatedSchemas:    commonEvent.ToSchemaIDChangePB(event.GetUpdatedSchemas()),
+				IsSyncPoint:       event.GetType() == commonEvent.TypeSyncPointEvent,
+				Stage:             heartbeatpb.BlockStage_WAITING,
+			},
+			Mode: d.GetMode(),
+		}
+		identifier := BlockEventIdentifier{
+			CommitTs:    event.GetCommitTs(),
+			IsSyncPoint: event.GetType() == commonEvent.TypeSyncPointEvent,
 		}
+		d.resendTaskMap.Set(identifier, newResendTask(message, d, nil))
+		d.sharedInfo.blockStatusesChan <- message
 	}
 
 	// dealing with events which update schema ids

downstreamadapter/dispatcher/event_dispatcher_test.go

@@ -338,7 +338,7 @@ func TestDispatcherHandleEvents(t *testing.T) {
 	// ===== sync point event =====
 
 	syncPointEvent := &commonEvent.SyncPointEvent{
-		CommitTsList: []uint64{6},
+		CommitTs: 6,
 	}
 	block = dispatcher.HandleEvents([]DispatcherEvent{NewDispatcherEvent(&nodeID, syncPointEvent)}, callback)
 	require.Equal(t, true, block)

downstreamadapter/dispatcher/helper.go

@@ -104,11 +104,6 @@ func (b *BlockEventStatus) setBlockEvent(event commonEvent.BlockEvent, blockStag
 	b.blockPendingEvent = event
 	b.blockStage = blockStage
 	b.blockCommitTs = event.GetCommitTs()
-
-	if event.GetType() == commonEvent.TypeSyncPointEvent {
-		commitTsList := event.(*commonEvent.SyncPointEvent).GetCommitTsList()
-		b.blockCommitTs = commitTsList[len(commitTsList)-1]
-	}
 }
 
 func (b *BlockEventStatus) updateBlockStage(blockStage heartbeatpb.BlockStage) {
@@ -131,10 +126,7 @@ func (b *BlockEventStatus) getEventAndStage() (commonEvent.BlockEvent, heartbeat
 	return b.blockPendingEvent, b.blockStage
 }
 
-// actionMatchs checks whether the action is for the current pending ddl event.
-// Most of time, the pending event only have one commitTs, so when the commitTs of the action meets the pending event's commitTs, it is enough.
-// While if the pending event is a sync point event with multiple commitTs, we only can do the action
-// when all the commitTs have been received.
+// actionMatchs checks whether the action is for the current pending ddl/sync point event.
 func (b *BlockEventStatus) actionMatchs(action *heartbeatpb.DispatcherAction) bool {
 	b.mutex.Lock()
 	defer b.mutex.Unlock()

downstreamadapter/dispatcher/table_progress.go

@@ -63,21 +63,9 @@ func NewTableProgress() *TableProgress {
 	}
 }
 
-func getFinalCommitTs(event commonEvent.Event) uint64 {
-	commitTs := event.GetCommitTs()
-	if event.GetType() == commonEvent.TypeSyncPointEvent {
-		// if the event is a sync point event, we use the last commitTs(the largest commitTs in the event) to calculate the progress.
-		// because a sync point event with multiple commitTs means there is no ddl / dmls between these commitTs.
-		// So we can just use the largest commitTs in the sync point event to calculate the progress.
-		commitTsList := event.(*commonEvent.SyncPointEvent).GetCommitTsList()
-		commitTs = commitTsList[len(commitTsList)-1]
-	}
-	return commitTs
-}
-
 // Add inserts a new event into the TableProgress.
 func (p *TableProgress) Add(event commonEvent.FlushableEvent) {
-	commitTs := getFinalCommitTs(event)
+	commitTs := event.GetCommitTs()
 	ts := Ts{startTs: event.GetStartTs(), commitTs: commitTs}
 
 	p.rwMutex.Lock()
@@ -94,7 +82,7 @@ func (p *TableProgress) Add(event commonEvent.FlushableEvent) {
 // Remove deletes an event from the TableProgress.
 // Note: Consider implementing batch removal in the future if needed.
 func (p *TableProgress) Remove(event commonEvent.FlushableEvent) {
-	ts := Ts{startTs: event.GetStartTs(), commitTs: getFinalCommitTs(event)}
+	ts := Ts{startTs: event.GetStartTs(), commitTs: event.GetCommitTs()}
 	p.rwMutex.Lock()
 	defer p.rwMutex.Unlock()
 
@@ -122,7 +110,7 @@ func (p *TableProgress) Pass(event commonEvent.FlushableEvent) {
 	p.rwMutex.Lock()
 	defer p.rwMutex.Unlock()
 
-	p.maxCommitTs = getFinalCommitTs(event)
+	p.maxCommitTs = event.GetCommitTs()
 }
 
 func (p *TableProgress) Len() int {

downstreamadapter/dispatcher/table_progress_test.go

@@ -66,36 +66,34 @@ func TestTableProgress(t *testing.T) {
 	assert.True(t, isEmpty)
 }
 
-// TestSyncPointEventWithMultipleCommitTs tests the behavior when SyncPointEvent contains multiple commitTs
-func TestSyncPointEventWithMultipleCommitTs(t *testing.T) {
+// TestSyncPointEventCommitTs tests the behavior for SyncPointEvent commitTs
+func TestSyncPointEventCommitTs(t *testing.T) {
 	tp := NewTableProgress()
 	assert.True(t, tp.Empty())
 
 	dispatcherID := common.NewDispatcherID()
 
-	// Create a SyncPointEvent with multiple commitTs
+	// Create a SyncPointEvent
 	syncPointEvent := &commonEvent.SyncPointEvent{
 		DispatcherID: dispatcherID,
-		CommitTsList: []uint64{10, 20, 30, 40},
+		CommitTs:     40,
 	}
 
-	// Should return the last (largest) commitTs from the list
-	finalCommitTs := getFinalCommitTs(syncPointEvent)
+	finalCommitTs := syncPointEvent.CommitTs
 	assert.Equal(t, uint64(40), finalCommitTs, "getFinalCommitTs should return the largest commitTs")
 
-	// Test Add method with SyncPointEvent containing multiple commitTs
+	// Test Add method with SyncPointEvent
 	tp.Add(syncPointEvent)
 	assert.False(t, tp.Empty())
 
-	// Verify that maxCommitTs is set to the largest commitTs
 	assert.Equal(t, uint64(40), tp.maxCommitTs, "maxCommitTs should be set to the largest commitTs")
 
 	// Verify GetCheckpointTs behavior
 	checkpointTs, isEmpty := tp.GetCheckpointTs()
 	assert.Equal(t, uint64(39), checkpointTs, "checkpointTs should be largest commitTs - 1")
 	assert.False(t, isEmpty)
 
-	// Test Remove method with SyncPointEvent containing multiple commitTs
+	// Test Remove method
 	tp.Remove(syncPointEvent)
 	assert.True(t, tp.Empty(), "TableProgress should be empty after removing the event")
 
@@ -104,10 +102,10 @@ func TestSyncPointEventWithMultipleCommitTs(t *testing.T) {
 	assert.Equal(t, uint64(39), checkpointTs, "checkpointTs should remain as maxCommitTs - 1 after removal")
 	assert.True(t, isEmpty)
 
-	// Create a SyncPointEvent with multiple commitTs
+	// Create another SyncPointEvent
 	syncPointEvent = &commonEvent.SyncPointEvent{
 		DispatcherID: dispatcherID,
-		CommitTsList: []uint64{50, 60},
+		CommitTs:     60,
 	}
 
 	tp.Add(syncPointEvent)
@@ -125,7 +123,7 @@ func TestSyncPointEventWithMultipleCommitTs(t *testing.T) {
 
 	syncPointEvent = &commonEvent.SyncPointEvent{
 		DispatcherID: dispatcherID,
-		CommitTsList: []uint64{80},
+		CommitTs:     80,
 	}
 
 	tp.Pass(syncPointEvent)

downstreamadapter/eventcollector/dispatcher_stat_test.go

@@ -265,8 +265,8 @@ func TestVerifyEventSequence(t *testing.T) {
 			lastEventSeq: 3,
 			event: dispatcher.DispatcherEvent{
 				Event: &commonEvent.SyncPointEvent{
-					CommitTsList: []uint64{100},
-					Seq:          5,
+					CommitTs: 100,
+					Seq:      5,
 				},
 			},
 			expectedResult: false,