dispatcher,event,cloudstorage: add DML two-stage ack

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -545,7 +545,8 @@ func (d *BasicDispatcher) HandleEvents(dispatcherEvents []DispatcherEvent, wakeC
 // When we handle events, we don't have any previous events still in sink.
 //
 // wakeCallback is used to wake the dynamic stream to handle the next batch events.
-// It will be called when all the events are flushed to downstream successfully.
+// It is triggered after DML events are enqueued to sink pipeline, while checkpoint
+// still advances only after PostFlush callbacks are completed.
 func (d *BasicDispatcher) handleEvents(dispatcherEvents []DispatcherEvent, wakeCallback func()) bool {
 	if d.GetRemovingStatus() {
 		log.Warn("dispatcher is removing", zap.Any("id", d.id))
@@ -622,13 +623,8 @@ func (d *BasicDispatcher) handleEvents(dispatcherEvents []DispatcherEvent, wakeC
 
 			block = true
 			dml.ReplicatingTs = d.creationPDTs
-			dml.AddPostFlushFunc(func() {
-				// Considering dml event in sink may be written to downstream not in order,
-				// thus, we use tableProgress.Empty() to ensure these events are flushed to downstream completely
-				// and wake dynamic stream to handle the next events.
-				if d.tableProgress.Empty() {
-					dmlWakeOnce.Do(wakeCallback)
-				}
+			dml.AddPostEnqueueFunc(func() {
+				dmlWakeOnce.Do(wakeCallback)
 			})
 			dmlEvents = append(dmlEvents, dml)
 		case commonEvent.TypeDDLEvent:

downstreamadapter/sink/cloudstorage/dml_writers.go

@@ -116,6 +116,7 @@ func (d *dmlWriters) AddDMLEvent(event *commonEvent.DMLEvent) {
 	_ = d.statistics.RecordBatchExecution(func() (int, int64, error) {
 		// emit a TxnCallbackableEvent encoupled with a sequence number starting from one.
 		d.msgCh.Push(newEventFragment(seq, tbl, event))
+		event.PostEnqueue()
 		return int(event.Len()), event.GetSize(), nil
 	})
 }

pkg/common/event/dml_event.go

@@ -392,9 +392,14 @@ type DMLEvent struct {
 
 	// The following fields are set and used by dispatcher.
 	ReplicatingTs uint64 `json:"replicating_ts"`
+	// PostTxnEnqueued is the functions to be executed after the transaction is
+	// enqueued into sink internal pipeline.
+	PostTxnEnqueued []func() `json:"-"`
 	// PostTxnFlushed is the functions to be executed after the transaction is flushed.
 	// It is set and used by dispatcher.
 	PostTxnFlushed []func() `json:"-"`
+	// postEnqueueCalled ensures PostTxnEnqueued callbacks are triggered at most once.
+	postEnqueueCalled bool `json:"-"`
 
 	// eventSize is the size of the event in bytes. It is set when it's unmarshaled.
 	eventSize int64 `json:"-"`
@@ -631,11 +636,22 @@ func (t *DMLEvent) GetStartTs() common.Ts {
 }
 
 func (t *DMLEvent) PostFlush() {
+	t.PostEnqueue()
 	for _, f := range t.PostTxnFlushed {
 		f()
 	}
 }
 
+func (t *DMLEvent) PostEnqueue() {
+	if t.postEnqueueCalled {
+		return
+	}
+	t.postEnqueueCalled = true
+	for _, f := range t.PostTxnEnqueued {
+		f()
+	}
+}
+
 func (t *DMLEvent) GetSeq() uint64 {
 	return t.Seq
 }
@@ -656,6 +672,14 @@ func (t *DMLEvent) AddPostFlushFunc(f func()) {
 	t.PostTxnFlushed = append(t.PostTxnFlushed, f)
 }
 
+func (t *DMLEvent) ClearPostEnqueueFunc() {
+	t.PostTxnEnqueued = t.PostTxnEnqueued[:0]
+}
+
+func (t *DMLEvent) AddPostEnqueueFunc(f func()) {
+	t.PostTxnEnqueued = append(t.PostTxnEnqueued, f)
+}
+
 // Rewind reset the offset to 0, So that the next GetNextRow will return the first row
 func (t *DMLEvent) Rewind() {
 	t.offset = 0

pkg/common/event/dml_event_test.go

@@ -15,6 +15,7 @@ package event
 
 import (
 	"encoding/binary"
+	"sync/atomic"
 	"testing"
 
 	"github.com/pingcap/ticdc/pkg/common"
@@ -342,3 +343,41 @@ func TestBatchDMLEventHeaderValidation(t *testing.T) {
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "incomplete data")
 }
+
+func TestDMLEventPostEnqueueFuncs(t *testing.T) {
+	t.Parallel()
+
+	event := &DMLEvent{}
+	var called int64
+	event.AddPostEnqueueFunc(func() {
+		atomic.AddInt64(&called, 1)
+	})
+	event.AddPostEnqueueFunc(func() {
+		atomic.AddInt64(&called, 1)
+	})
+
+	event.PostEnqueue()
+	event.PostEnqueue()
+
+	require.Equal(t, int64(2), atomic.LoadInt64(&called))
+}
+
+func TestDMLEventPostFlushTriggersPostEnqueueOnce(t *testing.T) {
+	t.Parallel()
+
+	event := &DMLEvent{}
+	var enqueueCalled int64
+	var flushCalled int64
+	event.AddPostEnqueueFunc(func() {
+		atomic.AddInt64(&enqueueCalled, 1)
+	})
+	event.AddPostFlushFunc(func() {
+		atomic.AddInt64(&flushCalled, 1)
+	})
+
+	event.PostFlush()
+	event.PostFlush()
+
+	require.Equal(t, int64(1), atomic.LoadInt64(&enqueueCalled))
+	require.Equal(t, int64(2), atomic.LoadInt64(&flushCalled))
+}