scheduler, processor(ticdc): advance redo resolvedTs before start sink (

#9277)

close #9172

cdc/processor/processor.go

@@ -108,7 +108,7 @@ var _ scheduler.TableExecutor = (*processor)(nil)
 // 2. Prepare phase for 2 phase scheduling, `isPrepare` should be true.
 // 3. Replicating phase for 2 phase scheduling, `isPrepare` should be false
 func (p *processor) AddTableSpan(
-	ctx context.Context, span tablepb.Span, startTs model.Ts, isPrepare bool,
+	ctx context.Context, span tablepb.Span, startTs model.Ts, isPrepare bool, barrier *schedulepb.Barrier,
 ) (bool, error) {
 	if !p.checkReadyForMessages() {
 		return false, nil
@@ -143,6 +143,16 @@ func (p *processor) AddTableSpan(
 			// table is `prepared`, and a `isPrepare = false` request indicate that old table should
 			// be stopped on original capture already, it's safe to start replicating data now.
 			if !isPrepare {
+				if p.redo.r.Enabled() {
+					var redoResolvedTs model.Ts
+					if barrier != nil {
+						redoResolvedTs = barrier.GlobalBarrierTs
+					} else {
+						stats := p.sinkManager.r.GetTableStats(span)
+						redoResolvedTs = stats.BarrierTs
+					}
+					p.redo.r.StartTable(span, redoResolvedTs)
+				}
 				if err := p.sinkManager.r.StartTable(span, startTs); err != nil {
 					return false, errors.Trace(err)
 				}
@@ -222,7 +232,8 @@ func (p *processor) IsAddTableSpanFinished(span tablepb.Span, isPrepare bool) bo
 	var tableResolvedTs, tableCheckpointTs uint64
 	var state tablepb.TableState
 	done := func() bool {
-		state, alreadyExist := p.sinkManager.r.GetTableState(span)
+		var alreadyExist bool
+		state, alreadyExist = p.sinkManager.r.GetTableState(span)
 		if alreadyExist {
 			stats := p.sinkManager.r.GetTableStats(span)
 			tableResolvedTs = stats.ResolvedTs

cdc/processor/processor_test.go

@@ -213,7 +213,7 @@ func TestTableExecutorAddingTableIndirectly(t *testing.T) {
 
 	// table-1: `preparing` -> `prepared` -> `replicating`
 	span := spanz.TableIDToComparableSpan(1)
-	ok, err := p.AddTableSpan(ctx, span, 20, true)
+	ok, err := p.AddTableSpan(ctx, span, 20, true, nil)
 	require.NoError(t, err)
 	require.True(t, ok)
 	p.sinkManager.r.UpdateBarrierTs(20, nil)
@@ -252,7 +252,7 @@ func TestTableExecutorAddingTableIndirectly(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, tablepb.TableStatePrepared, state)
 
-	ok, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 30, true)
+	ok, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 30, true, nil)
 	require.NoError(t, err)
 	require.True(t, ok)
 	stats = p.sinkManager.r.GetTableStats(span)
@@ -261,7 +261,7 @@ func TestTableExecutorAddingTableIndirectly(t *testing.T) {
 	require.Equal(t, model.Ts(20), stats.BarrierTs)
 
 	// Start to replicate table-1.
-	ok, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 30, false)
+	ok, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 30, false, nil)
 	require.NoError(t, err)
 	require.True(t, ok)
 
@@ -357,10 +357,10 @@ func TestProcessorClose(t *testing.T) {
 	tester.MustApplyPatches()
 
 	// add tables
-	done, err := p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 20, false)
+	done, err := p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 20, false, nil)
 	require.Nil(t, err)
 	require.True(t, done)
-	done, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(2), 30, false)
+	done, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(2), 30, false, nil)
 	require.Nil(t, err)
 	require.True(t, done)
 
@@ -396,10 +396,10 @@ func TestProcessorClose(t *testing.T) {
 	tester.MustApplyPatches()
 
 	// add tables
-	done, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 20, false)
+	done, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 20, false, nil)
 	require.Nil(t, err)
 	require.True(t, done)
-	done, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(2), 30, false)
+	done, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(2), 30, false, nil)
 	require.Nil(t, err)
 	require.True(t, done)
 	err = p.Tick(ctx)
@@ -441,10 +441,10 @@ func TestPositionDeleted(t *testing.T) {
 	tester.MustApplyPatches()
 
 	// add table
-	done, err := p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 30, false)
+	done, err := p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(1), 30, false, nil)
 	require.Nil(t, err)
 	require.True(t, done)
-	done, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(2), 40, false)
+	done, err = p.AddTableSpan(ctx, spanz.TableIDToComparableSpan(2), 40, false, nil)
 	require.Nil(t, err)
 	require.True(t, done)
 
@@ -550,7 +550,7 @@ func TestUpdateBarrierTs(t *testing.T) {
 	tester.MustApplyPatches()
 
 	span := spanz.TableIDToComparableSpan(1)
-	done, err := p.AddTableSpan(ctx, span, 5, false)
+	done, err := p.AddTableSpan(ctx, span, 5, false, nil)
 	require.True(t, done)
 	require.Nil(t, err)
 	err = p.Tick(ctx)

cdc/processor/sinkmanager/redo_log_advancer_test.go

@@ -79,6 +79,12 @@ func (m *mockRedoDMLManager) UpdateResolvedTs(ctx context.Context,
 	return nil
 }
 
+func (m *mockRedoDMLManager) StartTable(span tablepb.Span, resolvedTs uint64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.resolvedTss[span.TableID] = resolvedTs
+}
+
 func (m *mockRedoDMLManager) GetResolvedTs(span tablepb.Span) model.Ts {
 	m.mu.Lock()
 	defer m.mu.Unlock()

cdc/redo/manager.go

@@ -102,6 +102,7 @@ func (m *ddlManager) GetResolvedTs() model.Ts {
 type DMLManager interface {
 	redoManager
 	AddTable(span tablepb.Span, startTs uint64)
+	StartTable(span tablepb.Span, startTs uint64)
 	RemoveTable(span tablepb.Span)
 	UpdateResolvedTs(ctx context.Context, span tablepb.Span, resolvedTs uint64) error
 	GetResolvedTs(span tablepb.Span) model.Ts
@@ -305,6 +306,12 @@ func (m *logManager) emitRedoEvents(
 	})
 }
 
+// StartTable starts a table, which means the table is ready to emit redo events.
+// Note that this function should only be called once when adding a new table to processor.
+func (m *logManager) StartTable(span tablepb.Span, resolvedTs uint64) {
+	m.onResolvedTsMsg(span, resolvedTs)
+}
+
 // UpdateResolvedTs asynchronously updates resolved ts of a single table.
 func (m *logManager) UpdateResolvedTs(
 	ctx context.Context,

cdc/scheduler/internal/table_executor.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
+	"github.com/pingcap/tiflow/cdc/scheduler/schedulepb"
 )
 
 // TableExecutor is an abstraction for "Processor".
@@ -30,7 +31,7 @@ type TableExecutor interface {
 	// if `isPrepare` is true, the 1st phase of the 2 phase scheduling protocol.
 	// if `isPrepare` is false, the 2nd phase.
 	AddTableSpan(
-		ctx context.Context, span tablepb.Span, startTs model.Ts, isPrepare bool,
+		ctx context.Context, span tablepb.Span, startTs model.Ts, isPrepare bool, barrier *schedulepb.Barrier,
 	) (done bool, err error)
 
 	// IsAddTableSpanFinished make sure the requested table span is in the proper status

cdc/scheduler/internal/v3/agent/agent.go

@@ -211,7 +211,7 @@ func (a *agent) Tick(ctx context.Context) (*schedulepb.Barrier, error) {
 		return nil, errors.Trace(err)
 	}
 
-	responses, err := a.tableM.poll(ctx)
+	responses, err := a.tableM.poll(ctx, barrier)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

cdc/scheduler/internal/v3/agent/agent_test.go

@@ -164,7 +164,7 @@ func TestAgentHandleMessageDispatchTable(t *testing.T) {
 	// remove table not exist
 	ctx := context.Background()
 	a.handleMessageDispatchTableRequest(removeTableRequest, processorEpoch)
-	responses, err := a.tableM.poll(ctx)
+	responses, err := a.tableM.poll(ctx, &schedulepb.Barrier{})
 	require.NoError(t, err)
 	require.Len(t, responses, 0)
 
@@ -183,7 +183,7 @@ func TestAgentHandleMessageDispatchTable(t *testing.T) {
 	mockTableExecutor.On("AddTableSpan", mock.Anything, mock.Anything,
 		mock.Anything, mock.Anything).Return(false, nil)
 	a.handleMessageDispatchTableRequest(addTableRequest, processorEpoch)
-	responses, err = a.tableM.poll(ctx)
+	responses, err = a.tableM.poll(ctx, &schedulepb.Barrier{})
 	require.NoError(t, err)
 	require.Len(t, responses, 1)
 
@@ -203,14 +203,14 @@ func TestAgentHandleMessageDispatchTable(t *testing.T) {
 	mockTableExecutor.On("IsAddTableSpanFinished", mock.Anything,
 		mock.Anything, mock.Anything).Return(false, nil)
 	a.handleMessageDispatchTableRequest(addTableRequest, processorEpoch)
-	_, err = a.tableM.poll(ctx)
+	_, err = a.tableM.poll(ctx, &schedulepb.Barrier{})
 	require.NoError(t, err)
 
 	mockTableExecutor.ExpectedCalls = mockTableExecutor.ExpectedCalls[:1]
 	mockTableExecutor.On("IsAddTableSpanFinished", mock.Anything,
 		mock.Anything, mock.Anything).Return(true, nil)
 	a.handleMessageDispatchTableRequest(addTableRequest, processorEpoch)
-	responses, err = a.tableM.poll(ctx)
+	responses, err = a.tableM.poll(ctx, &schedulepb.Barrier{})
 	require.NoError(t, err)
 	require.Len(t, responses, 1)
 
@@ -232,7 +232,7 @@ func TestAgentHandleMessageDispatchTable(t *testing.T) {
 		mock.Anything, mock.Anything).Return(false, nil)
 
 	a.handleMessageDispatchTableRequest(addTableRequest, processorEpoch)
-	responses, err = a.tableM.poll(ctx)
+	responses, err = a.tableM.poll(ctx, &schedulepb.Barrier{})
 	require.NoError(t, err)
 	require.Len(t, responses, 1)
 
@@ -247,7 +247,7 @@ func TestAgentHandleMessageDispatchTable(t *testing.T) {
 	mockTableExecutor.On("IsAddTableSpanFinished", mock.Anything,
 		mock.Anything, mock.Anything).Return(true, nil)
 	a.handleMessageDispatchTableRequest(addTableRequest, processorEpoch)
-	responses, err = a.tableM.poll(ctx)
+	responses, err = a.tableM.poll(ctx, &schedulepb.Barrier{})
 	require.NoError(t, err)
 	require.Len(t, responses, 1)
 
@@ -262,7 +262,7 @@ func TestAgentHandleMessageDispatchTable(t *testing.T) {
 		Return(false)
 	// remove table in the replicating state failed, should still in replicating.
 	a.handleMessageDispatchTableRequest(removeTableRequest, processorEpoch)
-	responses, err = a.tableM.poll(ctx)
+	responses, err = a.tableM.poll(ctx, &schedulepb.Barrier{})
 	require.NoError(t, err)
 	require.Len(t, responses, 1)
 	removeTableResponse, ok := responses[0].DispatchTableResponse.
@@ -279,7 +279,7 @@ func TestAgentHandleMessageDispatchTable(t *testing.T) {
 		Return(3, false)
 	// remove table in the replicating state failed, should still in replicating.
 	a.handleMessageDispatchTableRequest(removeTableRequest, processorEpoch)
-	responses, err = a.tableM.poll(ctx)
+	responses, err = a.tableM.poll(ctx, &schedulepb.Barrier{})
 	require.NoError(t, err)
 	require.Len(t, responses, 1)
 	removeTableResponse, ok = responses[0].DispatchTableResponse.
@@ -293,7 +293,7 @@ func TestAgentHandleMessageDispatchTable(t *testing.T) {
 		Return(3, true)
 	// remove table in the replicating state success, should in stopped
 	a.handleMessageDispatchTableRequest(removeTableRequest, processorEpoch)
-	responses, err = a.tableM.poll(ctx)
+	responses, err = a.tableM.poll(ctx, &schedulepb.Barrier{})
 	require.NoError(t, err)
 	require.Len(t, responses, 1)
 	removeTableResponse, ok = responses[0].DispatchTableResponse.
@@ -809,7 +809,7 @@ func TestAgentCommitAddTableDuringStopping(t *testing.T) {
 
 	// Prepare add table is still in-progress.
 	mockTableExecutor.
-		On("AddTableSpan", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		On("AddTableSpan", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(true, nil).Once()
 	mockTableExecutor.
 		On("IsAddTableSpanFinished", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
@@ -819,7 +819,7 @@ func TestAgentCommitAddTableDuringStopping(t *testing.T) {
 	require.Len(t, trans.SendBuffer, 0)
 
 	mockTableExecutor.
-		On("AddTableSpan", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		On("AddTableSpan", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(true, nil).Once()
 	mockTableExecutor.
 		On("IsAddTableSpanFinished", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
@@ -1069,7 +1069,7 @@ func newMockTableExecutor() *MockTableExecutor {
 
 // AddTableSpan adds a table span to the executor.
 func (e *MockTableExecutor) AddTableSpan(
-	ctx context.Context, span tablepb.Span, startTs model.Ts, isPrepare bool,
+	ctx context.Context, span tablepb.Span, startTs model.Ts, isPrepare bool, _ *schedulepb.Barrier,
 ) (bool, error) {
 	log.Info("AddTableSpan",
 		zap.String("span", span.String()),

cdc/scheduler/internal/v3/agent/table.go

@@ -170,14 +170,14 @@ func (t *tableSpan) handleRemoveTableTask() *schedulepb.Message {
 }
 
 func (t *tableSpan) handleAddTableTask(
-	ctx context.Context,
+	ctx context.Context, barrier *schedulepb.Barrier,
 ) (result *schedulepb.Message, err error) {
 	state, _ := t.getAndUpdateTableSpanState()
 	changed := true
 	for changed {
 		switch state {
 		case tablepb.TableStateAbsent:
-			done, err := t.executor.AddTableSpan(ctx, t.task.Span, t.task.StartTs, t.task.IsPrepare)
+			done, err := t.executor.AddTableSpan(ctx, t.task.Span, t.task.StartTs, t.task.IsPrepare, barrier)
 			if err != nil || !done {
 				log.Warn("schedulerv3: agent add table failed",
 					zap.String("namespace", t.changefeedID.Namespace),
@@ -208,7 +208,7 @@ func (t *tableSpan) handleAddTableTask(
 			}
 
 			if t.task.status == dispatchTableTaskReceived {
-				done, err := t.executor.AddTableSpan(ctx, t.task.Span, t.task.StartTs, false)
+				done, err := t.executor.AddTableSpan(ctx, t.task.Span, t.task.StartTs, false, barrier)
 				if err != nil || !done {
 					log.Warn("schedulerv3: agent add table failed",
 						zap.String("namespace", t.changefeedID.Namespace),
@@ -283,14 +283,14 @@ func (t *tableSpan) injectDispatchTableTask(task *dispatchTableTask) {
 		zap.Any("ignoredTask", task))
 }
 
-func (t *tableSpan) poll(ctx context.Context) (*schedulepb.Message, error) {
+func (t *tableSpan) poll(ctx context.Context, barrier *schedulepb.Barrier) (*schedulepb.Message, error) {
 	if t.task == nil {
 		return nil, nil
 	}
 	if t.task.IsRemove {
 		return t.handleRemoveTableTask(), nil
 	}
-	return t.handleAddTableTask(ctx)
+	return t.handleAddTableTask(ctx, barrier)
 }
 
 type tableSpanManager struct {
@@ -310,12 +310,12 @@ func newTableSpanManager(
 	}
 }
 
-func (tm *tableSpanManager) poll(ctx context.Context) ([]*schedulepb.Message, error) {
+func (tm *tableSpanManager) poll(ctx context.Context, barrier *schedulepb.Barrier) ([]*schedulepb.Message, error) {
 	result := make([]*schedulepb.Message, 0)
 	var err error
 	toBeDropped := []tablepb.Span{}
 	tm.tables.Ascend(func(span tablepb.Span, table *tableSpan) bool {
-		message, err1 := table.poll(ctx)
+		message, err1 := table.poll(ctx, barrier)
 		if err != nil {
 			err = errors.Trace(err1)
 			return false