From 045905a6ea300b1bf3adc621b51c01d11bad71a0 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 21 Aug 2026 08:56:05 +0000 Subject: [PATCH 01/11] init Signed-off-by: wk989898 --- cmd/kafka-consumer/consumer.go | 2 + cmd/kafka-consumer/writer.go | 16 +- cmd/pulsar-consumer/consumer.go | 2 + cmd/pulsar-consumer/writer.go | 11 + cmd/pulsar-consumer/writer_test.go | 12 +- cmd/storage-consumer/consumer.go | 10 + cmd/util/event_group.go | 311 ++++++++++++++++++++++++++++- cmd/util/event_group_test.go | 123 +++++++----- 8 files changed, 420 insertions(+), 67 deletions(-) diff --git a/cmd/kafka-consumer/consumer.go b/cmd/kafka-consumer/consumer.go index 2bf26f1aee..c98125ddd0 100644 --- a/cmd/kafka-consumer/consumer.go +++ b/cmd/kafka-consumer/consumer.go @@ -170,6 +170,8 @@ func (c *consumer) readMessage(ctx context.Context) error { // Run the consumer, read data and write to the downstream target. func (c *consumer) Run(ctx context.Context) error { + defer c.writer.cleanupEventsGroups() + g, ctx := errgroup.WithContext(ctx) g.Go(func() error { return c.writer.run(ctx) diff --git a/cmd/kafka-consumer/writer.go b/cmd/kafka-consumer/writer.go index bbf8388405..1284f31927 100644 --- a/cmd/kafka-consumer/writer.go +++ b/cmd/kafka-consumer/writer.go @@ -147,6 +147,17 @@ func (w *writer) run(ctx context.Context) error { return w.mysqlSink.Run(ctx) } +func (w *writer) cleanupEventsGroups() { + for _, progress := range w.progresses { + for _, group := range progress.eventsGroup { + if err := group.Cleanup(); err != nil { + log.Warn("cleanup events group spill file failed", + zap.Int32("partition", progress.partition), zap.Error(err)) + } + } + } +} + func (w *writer) flushDDLEvent(ctx context.Context, ddl *event.DDLEvent) error { var ( done = make(chan struct{}, 1) @@ -628,8 +639,9 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti group = util.NewEventsGroup(progress.partition, tableID) progress.eventsGroup[tableID] = group } - message = w.messageWithPartitionCheck(message, progress.partition, offset) - group.AppendMessage(message) + group.AppendMessageWithPostRestore(message, func(message *common.DMLMessage) *common.DMLMessage { + return w.messageWithPartitionCheck(message, progress.partition, offset) + }) if commitTs < progress.watermark { log.Warn("DML event fallback row, since less than the partition watermark, append it and sort before flush", zap.Int64("tableID", tableID), zap.Int32("partition", group.Partition), diff --git a/cmd/pulsar-consumer/consumer.go b/cmd/pulsar-consumer/consumer.go index 8c75a8843b..c6b3d600ec 100644 --- a/cmd/pulsar-consumer/consumer.go +++ b/cmd/pulsar-consumer/consumer.go @@ -124,6 +124,8 @@ func (c *consumer) readMessage(ctx context.Context) error { // Run the consumer, read data and write to the downstream target. func (c *consumer) Run(ctx context.Context) error { + defer c.writer.cleanupEventsGroups() + g, ctx := errgroup.WithContext(ctx) g.Go(func() error { return c.writer.run(ctx) diff --git a/cmd/pulsar-consumer/writer.go b/cmd/pulsar-consumer/writer.go index e0fb914e1e..e748c10a49 100644 --- a/cmd/pulsar-consumer/writer.go +++ b/cmd/pulsar-consumer/writer.go @@ -139,6 +139,17 @@ func (w *writer) run(ctx context.Context) error { return w.mysqlSink.Run(ctx) } +func (w *writer) cleanupEventsGroups() { + for _, progress := range w.progresses { + for _, group := range progress.eventsGroup { + if err := group.Cleanup(); err != nil { + log.Warn("cleanup events group spill file failed", + zap.Int32("partition", progress.partition), zap.Error(err)) + } + } + } +} + func (w *writer) flushDDLEvent(ctx context.Context, ddl *commonEvent.DDLEvent) error { var ( done = make(chan struct{}, 1) diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go index 35c9c037e3..6775aecfba 100644 --- a/cmd/pulsar-consumer/writer_test.go +++ b/cmd/pulsar-consumer/writer_test.go @@ -381,20 +381,16 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTable(t *testing.T) { w.onDDL(ddl) require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst")) - newDMLMessage := func(commitTs uint64) *codeccommon.DMLMessage { - return codeccommon.NewDMLMessage(1, "target", "dst", commitTs, common.RowTypeUpdate, nil) - } - progress := w.progresses[0] - w.appendMessage2Group(newDMLMessage(200), progress) - w.appendMessage2Group(newDMLMessage(100), progress) + w.appendMessage2Group(newDMLMessageForWriterTest(200), progress) + w.appendMessage2Group(newDMLMessageForWriterTest(100), progress) resolved := progress.eventsGroup[1].ResolveInto(150, nil) require.Len(t, resolved, 1) require.Equal(t, uint64(100), resolved[0].GetCommitTs()) } -func TestWriteMessageDefersDMLAssemblyUntilFlush(t *testing.T) { +func TestWriteMessageSpillsDMLImmediately(t *testing.T) { ctx := context.Background() ctrl := gomock.NewController(t) s := sinkmock.NewMockSink(ctrl) @@ -428,7 +424,7 @@ func TestWriteMessageDefersDMLAssemblyUntilFlush(t *testing.T) { require.Equal(t, 1, decoder.addKeyValueCount) require.Equal(t, 1, decoder.hasNextCount) require.Equal(t, 1, decoder.nextDMLMessageCount) - require.Zero(t, decoder.toDMLEventCount) + require.Equal(t, 1, decoder.toDMLEventCount) require.Len(t, progress.eventsGroup[1].ResolveInto(99, nil), 0) progress.watermark = 100 diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index d89de46cfe..9567ee57e0 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -449,6 +449,14 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error { } } +func (c *consumer) cleanupEventsGroups() { + for _, group := range c.eventsGroup { + if err := group.Cleanup(); err != nil { + log.Warn("cleanup events group spill file failed", zap.Error(err)) + } + } +} + func (c *consumer) parseDMLIndexFile(ctx context.Context, path string, dmlkey cloudstorage.DMLPathKey) { if c.globalCheckpointTs > 0 && dmlkey.TableVersion > c.globalCheckpointTs { log.Debug("skip dml index file by checkpoint", @@ -767,6 +775,8 @@ func (c *consumer) handle(ctx context.Context) error { } func (c *consumer) run(ctx context.Context) error { + defer c.cleanupEventsGroups() + g, ctx := errgroup.WithContext(ctx) g.Go(func() error { return c.sink.Run(ctx) diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index d4d324282a..e310c429b9 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -14,21 +14,38 @@ package util import ( + "encoding/binary" + "encoding/json" "math" + "os" "sort" "github.com/pingcap/log" + commonType "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" + "github.com/pingcap/ticdc/pkg/errors" codeccommon "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/pingcap/ticdc/pkg/spill" + "github.com/pingcap/tidb/pkg/types" + "github.com/pingcap/tidb/pkg/util/chunk" "go.uber.org/zap" ) +const eventsGroupSpillPattern = "ticdc-events-group-*.spill" + +type spilledMessage struct { + commitTs uint64 + handle spill.Handle + postRestore func(*codeccommon.DMLMessage) *codeccommon.DMLMessage +} + // EventsGroup stores change event messages. type EventsGroup struct { Partition int32 tableID int64 - messages []*codeccommon.DMLMessage + messages []spilledMessage + spillFile *spill.RecordFile outOfOrder bool HighWatermark uint64 } @@ -38,25 +55,69 @@ func NewEventsGroup(partition int32, tableID int64) *EventsGroup { return &EventsGroup{ Partition: partition, tableID: tableID, - messages: make([]*codeccommon.DMLMessage, 0, 1024), + messages: make([]spilledMessage, 0, 1024), } } -// AppendMessage appends a message to event groups. +// AppendMessage materializes a message and appends it to a local spill file. DMLMessage carries a +// decoder closure, so persisting its reconstructed event is necessary to release the decoder input +// retained by that closure. func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage) { + g.appendMessage(message, nil) +} + +// AppendMessageWithPostRestore appends a message and applies postRestore after it is read back from +// disk. It keeps consumer checks that intentionally run immediately before flushing out of the +// on-disk representation. +func (g *EventsGroup) AppendMessageWithPostRestore( + message *codeccommon.DMLMessage, + postRestore func(*codeccommon.DMLMessage) *codeccommon.DMLMessage, +) { + g.appendMessage(message, postRestore) +} + +func (g *EventsGroup) appendMessage( + message *codeccommon.DMLMessage, + postRestore func(*codeccommon.DMLMessage) *codeccommon.DMLMessage, +) { commitTs := message.GetCommitTs() - if len(g.messages) > 0 && commitTs < g.messages[len(g.messages)-1].GetCommitTs() { + if len(g.messages) > 0 && commitTs < g.messages[len(g.messages)-1].commitTs { g.outOfOrder = true } if commitTs > g.HighWatermark { g.HighWatermark = commitTs } - g.messages = append(g.messages, message) + + data, row, err := marshalDMLMessage(message) + if err != nil { + log.Panic("marshal DML message for spill failed", + zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + } + if g.spillFile == nil { + g.spillFile, err = spill.NewRecordFile(os.TempDir(), eventsGroupSpillPattern) + if err != nil { + log.Panic("create events group spill file failed", + zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + } + } + handle, err := g.spillFile.Append(data) + if err != nil { + log.Panic("write DML message to spill file failed", + zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + } + g.messages = append(g.messages, spilledMessage{ + commitTs: commitTs, + handle: handle, + postRestore: postRestore, + }) + // Codec decoders use this callback to release their pooled chunks. The event is durable in the + // spill file now, so the original in-memory event is no longer needed. + row.PostFlush() } // ResolveInto appends all messages with CommitTs <= resolve into dst in commit-ts order and removes -// them from the group. ResolveInto copies pointers into dst first, then clears the resolved messages -// so Go GC can reclaim them once downstream is done with them. +// them from the group. Resolved messages are restored from the spill file only when downstream needs +// them, keeping the buffered group out of heap memory. func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) []*codeccommon.DMLMessage { if len(g.messages) == 0 { return dst @@ -64,12 +125,12 @@ func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) if g.outOfOrder { sort.SliceStable(g.messages, func(i, j int) bool { - return g.messages[i].GetCommitTs() < g.messages[j].GetCommitTs() + return g.messages[i].commitTs < g.messages[j].commitTs }) } resolvedCount := sort.Search(len(g.messages), func(i int) bool { - return g.messages[i].GetCommitTs() > resolve + return g.messages[i].commitTs > resolve }) if g.outOfOrder { log.Warn("DML events are out of order before flush, sort them", @@ -83,13 +144,35 @@ func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) return dst } - dst = append(dst, g.messages[:resolvedCount]...) + for _, message := range g.messages[:resolvedCount] { + data, err := g.spillFile.Read(message.handle) + if err != nil { + log.Panic("read DML message from spill file failed", + zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + } + restored, err := unmarshalDMLMessage(data) + if err != nil { + log.Panic("unmarshal DML message from spill file failed", + zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + } + if message.postRestore != nil { + restored = message.postRestore(restored) + } + dst = append(dst, restored) + } remainingCount := len(g.messages) - resolvedCount copy(g.messages, g.messages[resolvedCount:]) clear(g.messages[remainingCount:]) g.messages = g.messages[:remainingCount] + if len(g.messages) == 0 { + if err := g.spillFile.Cleanup(); err != nil { + log.Panic("cleanup events group spill file failed", + zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + } + g.spillFile = nil + } if len(g.messages) != 0 { - firstCommitTs := g.messages[0].GetCommitTs() + firstCommitTs := g.messages[0].commitTs log.Debug("not all events resolved", zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Int("resolved", resolvedCount), zap.Int("remained", len(g.messages)), @@ -103,6 +186,212 @@ func (g *EventsGroup) GetAllMessages() []*codeccommon.DMLMessage { return g.ResolveInto(math.MaxUint64, nil) } +// Cleanup removes pending spill records when the consumer is stopping. +func (g *EventsGroup) Cleanup() error { + if g.spillFile == nil { + return nil + } + err := g.spillFile.Cleanup() + g.spillFile = nil + clear(g.messages) + g.messages = g.messages[:0] + return err +} + +func marshalDMLMessage(message *codeccommon.DMLMessage) (data []byte, row *commonEvent.DMLEvent, err error) { + if message == nil { + return nil, nil, errors.ErrSpillFileOp.FastGenByArgs("cannot spill nil DML message") + } + + row = message.ToDMLEvent() + if row == nil { + return nil, nil, errors.ErrSpillFileOp.FastGenByArgs("cannot spill DML message without event") + } + if row.Version == 0 { + row.Version = commonEvent.DMLEventVersion1 + } + eventData, err := row.Marshal() + if err != nil { + return nil, nil, errors.WrapError(errors.ErrSpillFileOp, err, "marshal DML event") + } + + var tableInfoData []byte + tableInfoStored := false + if row.TableInfo != nil { + tableInfoData, err = marshalDMLTableInfo(row.TableInfo) + if err != nil { + if row.Rows != nil && row.Rows.NumRows() > 0 { + return nil, nil, err + } + tableInfoData = nil + } else { + tableInfoStored = true + } + } + + var rowsData []byte + if row.Rows != nil && (row.Rows.NumRows() > 0 || tableInfoStored) { + rowsData, err = marshalDMLRows(row, tableInfoStored) + if err != nil { + if row.Rows.NumRows() > 0 { + return nil, nil, err + } + rowsData = nil + } + } + + checksumData, err := json.Marshal(row.Checksum) + if err != nil { + return nil, nil, errors.WrapError(errors.ErrSpillFileOp, err, "marshal DML checksums") + } + + data = make([]byte, 0, 7*8+len(eventData)+len(tableInfoData)+len(rowsData)+len(checksumData)) + data = appendUint64(data, uint64(len(eventData))) + data = append(data, eventData...) + data = appendUint64(data, uint64(len(tableInfoData))) + data = append(data, tableInfoData...) + data = appendUint64(data, uint64(len(rowsData))) + data = append(data, rowsData...) + data = appendUint64(data, uint64(len(checksumData))) + data = append(data, checksumData...) + if row.Rows != nil { + data = appendUint64(data, 1) + } else { + data = appendUint64(data, 0) + } + data = appendUint64(data, row.TableInfoVersion) + data = appendUint64(data, row.ReplicatingTs) + return data, row, nil +} + +func marshalDMLTableInfo(tableInfo *commonType.TableInfo) (data []byte, err error) { + defer func() { + if recover() != nil { + err = errors.ErrSpillFileOp.FastGenByArgs("marshal incomplete DML table info") + } + }() + + data, err = tableInfo.Marshal() + if err != nil { + return nil, errors.WrapError(errors.ErrSpillFileOp, err, "marshal DML table info") + } + return data, nil +} + +func marshalDMLRows(row *commonEvent.DMLEvent, tableInfoStored bool) (data []byte, err error) { + defer func() { + if recover() != nil { + err = errors.ErrSpillFileOp.FastGenByArgs("marshal DML rows with incomplete table info") + } + }() + + fieldTypes := []*types.FieldType(nil) + if tableInfoStored { + fieldTypes = row.TableInfo.GetFieldSlice() + } + return chunk.NewCodec(fieldTypes).Encode(row.Rows), nil +} + +func unmarshalDMLMessage(data []byte) (*codeccommon.DMLMessage, error) { + eventData, data, err := readSpilledField(data) + if err != nil { + return nil, err + } + tableInfoData, data, err := readSpilledField(data) + if err != nil { + return nil, err + } + rowsData, data, err := readSpilledField(data) + if err != nil { + return nil, err + } + checksumData, data, err := readSpilledField(data) + if err != nil { + return nil, err + } + rowsPresent, data, err := readSpilledUint64(data) + if err != nil { + return nil, err + } + if rowsPresent > 1 { + return nil, errors.ErrSpillFileOp.FastGenByArgs("invalid DML spill rows flag") + } + tableInfoVersion, data, err := readSpilledUint64(data) + if err != nil { + return nil, err + } + replicatingTs, data, err := readSpilledUint64(data) + if err != nil { + return nil, err + } + if len(data) != 0 { + return nil, errors.ErrSpillFileOp.FastGenByArgs("unexpected trailing DML spill data") + } + + row := &commonEvent.DMLEvent{} + if err := row.Unmarshal(eventData); err != nil { + return nil, errors.WrapError(errors.ErrSpillFileOp, err, "unmarshal DML event") + } + if len(tableInfoData) != 0 { + tableInfo, err := commonType.UnmarshalJSONToTableInfo(tableInfoData) + if err != nil { + return nil, errors.WrapError(errors.ErrSpillFileOp, err, "unmarshal DML table info") + } + row.TableInfo = tableInfo + } + if rowsPresent == 1 && len(rowsData) == 0 { + row.Rows = chunk.NewChunkWithCapacity(nil, 0) + } else if len(rowsData) != 0 { + fieldTypes := []*types.FieldType(nil) + if row.TableInfo != nil { + fieldTypes = row.TableInfo.GetFieldSlice() + } + row.Rows, _ = chunk.NewCodec(fieldTypes).Decode(rowsData) + } + row.TableInfoVersion = tableInfoVersion + row.ReplicatingTs = replicatingTs + if err := json.Unmarshal(checksumData, &row.Checksum); err != nil { + return nil, errors.WrapError(errors.ErrSpillFileOp, err, "unmarshal DML checksums") + } + if len(row.RowTypes) == 0 { + return nil, errors.ErrSpillFileOp.FastGenByArgs("spilled DML event has no row type") + } + + var schema, table string + if row.TableInfo != nil { + schema = row.TableInfo.GetSchemaName() + table = row.TableInfo.GetTableName() + } + + return codeccommon.NewDMLMessage(row.PhysicalTableID, schema, table, row.CommitTs, row.RowTypes[0], func() *commonEvent.DMLEvent { + return row + }), nil +} + +func appendUint64(data []byte, value uint64) []byte { + var buf [8]byte + binary.BigEndian.PutUint64(buf[:], value) + return append(data, buf[:]...) +} + +func readSpilledField(data []byte) ([]byte, []byte, error) { + length, data, err := readSpilledUint64(data) + if err != nil { + return nil, nil, err + } + if length > uint64(len(data)) { + return nil, nil, errors.ErrSpillFileOp.FastGenByArgs("invalid DML spill field length") + } + return data[:length], data[length:], nil +} + +func readSpilledUint64(data []byte) (uint64, []byte, error) { + if len(data) < 8 { + return 0, nil, errors.ErrSpillFileOp.FastGenByArgs("truncated DML spill data") + } + return binary.BigEndian.Uint64(data[:8]), data[8:], nil +} + // AppendOrMergeDMLEvent appends a DML event, or merges it into the previous event // when both events belong to the same table group and have the same commit-ts. func AppendOrMergeDMLEvent(events []*commonEvent.DMLEvent, row *commonEvent.DMLEvent) []*commonEvent.DMLEvent { diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index a756f297bd..4f37b8dbed 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -14,19 +14,25 @@ package util import ( + "os" "testing" "github.com/pingcap/log" "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" + "github.com/pingcap/ticdc/pkg/integrity" codeccommon "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/pingcap/tidb/pkg/meta/model" + "github.com/pingcap/tidb/pkg/parser/ast" + "github.com/pingcap/tidb/pkg/parser/mysql" + "github.com/pingcap/tidb/pkg/types" "github.com/pingcap/tidb/pkg/util/chunk" "github.com/stretchr/testify/require" "go.uber.org/zap/zapcore" ) func newTestDMLMessage(commitTs uint64) *codeccommon.DMLMessage { - return codeccommon.NewDMLMessage(1, "test", "t", commitTs, common.RowTypeInsert, nil) + return codeccommon.NewDMLMessageFromEvent(newTestDMLEvent(commitTs, common.RowTypeInsert)) } func newTestDMLEvent(commitTs uint64, rowTypes ...common.RowType) *commonEvent.DMLEvent { @@ -39,16 +45,16 @@ func newTestDMLEvent(commitTs uint64, rowTypes ...common.RowType) *commonEvent.D } } -func TestEventsGroupResolveIntoAppendsAndClearsResolvedMessages(t *testing.T) { +func TestEventsGroupResolveIntoAppendsAndCleansResolvedSpillRecords(t *testing.T) { // Scenario: A consumer resolves events by watermark/commit-ts and appends them into a downstream - // batch slice. We must clear resolved messages in the group's backing array to avoid retaining - // already-flushed events and causing unbounded memory growth. + // batch slice. Buffered messages are held only by a spill file, and the file must be cleaned once + // all of its records have been resolved. // // Steps: // 1. Append 3 events with increasing CommitTs. // 2. Call ResolveInto with resolve=2 and a nil dst. // 3. Verify (a) returned events are correct, (b) group keeps only the remaining event, - // (c) resolved messages in the original backing slice are cleared (nil'd). + // (c) the file survives the partial resolve. group := NewEventsGroup(0, 1) m1 := newTestDMLMessage(1) m2 := newTestDMLMessage(2) @@ -57,25 +63,23 @@ func TestEventsGroupResolveIntoAppendsAndClearsResolvedMessages(t *testing.T) { group.AppendMessage(m2) group.AppendMessage(m3) - // Keep a reference to the original slice header so we can validate that ResolveInto clears - // resolved messages in-place (this is what prevents GC retention of flushed events). - original := group.messages + spillPath := group.spillFile.Path() var dst []*codeccommon.DMLMessage dst = group.ResolveInto(2, dst) require.Len(t, dst, 2) - require.Same(t, m1, dst[0]) - require.Same(t, m2, dst[1]) + require.Equal(t, m1.GetCommitTs(), dst[0].GetCommitTs()) + require.Equal(t, m2.GetCommitTs(), dst[1].GetCommitTs()) require.Len(t, group.messages, 1) - require.Same(t, m3, group.messages[0]) + require.Equal(t, m3.GetCommitTs(), group.messages[0].commitTs) + require.FileExists(t, spillPath) - // The unresolved event is compacted to the front, and the tail is cleared so the group - // doesn't keep flushed events alive via its backing array. - require.Same(t, m3, original[0]) - require.Nil(t, original[1]) - require.Nil(t, original[2]) + group.GetAllMessages() + require.Nil(t, group.spillFile) + _, err := os.Stat(spillPath) + require.True(t, os.IsNotExist(err)) } func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) { @@ -87,18 +91,13 @@ func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) { group.AppendMessage(m1) group.AppendMessage(m2) - original := group.messages dst := make([]*codeccommon.DMLMessage, 0, 1) dst = group.ResolveInto(5, dst) require.Len(t, dst, 0) require.Len(t, group.messages, 2) - require.Same(t, m1, group.messages[0]) - require.Same(t, m2, group.messages[1]) - - // No prefix should be cleared because nothing was resolved. - require.Same(t, m1, original[0]) - require.Same(t, m2, original[1]) + require.Equal(t, m1.GetCommitTs(), group.messages[0].commitTs) + require.Equal(t, m2.GetCommitTs(), group.messages[1].commitTs) } func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) { @@ -110,17 +109,18 @@ func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) { group.AppendMessage(m1) group.AppendMessage(m2) - original := group.messages + spillPath := group.spillFile.Path() var dst []*codeccommon.DMLMessage dst = group.ResolveInto(100, dst) require.Len(t, dst, 2) - require.Same(t, m1, dst[0]) - require.Same(t, m2, dst[1]) + require.Equal(t, m1.GetCommitTs(), dst[0].GetCommitTs()) + require.Equal(t, m2.GetCommitTs(), dst[1].GetCommitTs()) require.Len(t, group.messages, 0) - require.Nil(t, original[0]) - require.Nil(t, original[1]) + require.Nil(t, group.spillFile) + _, err := os.Stat(spillPath) + require.True(t, os.IsNotExist(err)) } func TestEventsGroupResolveIntoSortsOutOfOrderResolvedMessages(t *testing.T) { @@ -132,19 +132,15 @@ func TestEventsGroupResolveIntoSortsOutOfOrderResolvedMessages(t *testing.T) { group.AppendMessage(m2) group.AppendMessage(m3) - original := group.messages var dst []*codeccommon.DMLMessage dst = group.ResolveInto(25, dst) require.Len(t, dst, 2) - require.Same(t, m2, dst[0]) - require.Same(t, m1, dst[1]) + require.Equal(t, m2.GetCommitTs(), dst[0].GetCommitTs()) + require.Equal(t, m1.GetCommitTs(), dst[1].GetCommitTs()) require.Len(t, group.messages, 1) - require.Same(t, m3, group.messages[0]) - require.Same(t, m3, original[0]) - require.Nil(t, original[1]) - require.Nil(t, original[2]) + require.Equal(t, m3.GetCommitTs(), group.messages[0].commitTs) } func TestEventsGroupResolveIntoKeepsSameCommitTsStable(t *testing.T) { @@ -160,9 +156,9 @@ func TestEventsGroupResolveIntoKeepsSameCommitTsStable(t *testing.T) { dst = group.ResolveInto(20, dst) require.Len(t, dst, 3) - require.Same(t, m2, dst[0]) - require.Same(t, m1, dst[1]) - require.Same(t, m3, dst[2]) + require.Equal(t, m2.GetCommitTs(), dst[0].GetCommitTs()) + require.Equal(t, m1.GetCommitTs(), dst[1].GetCommitTs()) + require.Equal(t, m3.GetCommitTs(), dst[2].GetCommitTs()) require.Empty(t, group.messages) } @@ -178,12 +174,49 @@ func TestEventsGroupGetAllMessagesSortsOutOfOrderMessages(t *testing.T) { messages := group.GetAllMessages() require.Len(t, messages, 3) - require.Same(t, m2, messages[0]) - require.Same(t, m1, messages[1]) - require.Same(t, m3, messages[2]) + require.Equal(t, m2.GetCommitTs(), messages[0].GetCommitTs()) + require.Equal(t, m1.GetCommitTs(), messages[1].GetCommitTs()) + require.Equal(t, m3.GetCommitTs(), messages[2].GetCommitTs()) require.Empty(t, group.messages) } +func TestEventsGroupRestoresSpilledEventRowsAndTableInfo(t *testing.T) { + tableInfo := common.WrapTableInfo("test", &model.TableInfo{ + ID: 1, + Name: ast.NewCIStr("t"), + Columns: []*model.ColumnInfo{ + { + ID: 1, + Name: ast.NewCIStr("id"), + FieldType: *types.NewFieldType(mysql.TypeLonglong), + }, + }, + }) + rows := chunk.NewChunkWithCapacity(tableInfo.GetFieldSlice(), 1) + rows.AppendInt64(0, 42) + event := commonEvent.NewDMLEvent(common.NewDispatcherID(), 1, 90, 100, tableInfo) + event.Rows = rows + event.RowTypes = []common.RowType{common.RowTypeInsert} + event.Length = 1 + event.TableInfoVersion = 88 + event.ReplicatingTs = 99 + event.Checksum = []*integrity.Checksum{{Current: 1, Previous: 2, Corrupted: true, Version: 3}} + + group := NewEventsGroup(0, 1) + group.AppendMessage(codeccommon.NewDMLMessageFromEvent(event)) + + messages := group.GetAllMessages() + require.Len(t, messages, 1) + restored := messages[0].ToDMLEvent() + require.Equal(t, uint64(100), restored.CommitTs) + require.Equal(t, uint64(88), restored.TableInfoVersion) + require.Equal(t, uint64(99), restored.ReplicatingTs) + require.Equal(t, event.Checksum, restored.Checksum) + require.Equal(t, "test", restored.TableInfo.GetSchemaName()) + require.Equal(t, "t", restored.TableInfo.GetTableName()) + require.Equal(t, int64(42), restored.Rows.GetRow(0).GetInt64(0)) +} + func BenchmarkEventsGroupResolveInto(b *testing.B) { const messageCount = 16 * 1024 @@ -215,16 +248,14 @@ func BenchmarkEventsGroupResolveInto(b *testing.B) { lastIndex := len(source) - 1 source[lastIndex-1], source[lastIndex] = source[lastIndex], source[lastIndex-1] } - group := NewEventsGroup(0, 1) - group.messages = make([]*codeccommon.DMLMessage, 0, messageCount) dst := make([]*codeccommon.DMLMessage, 0, messageCount) b.ReportAllocs() b.ResetTimer() for b.Loop() { - if len(group.messages) != messageCount { - group.messages = append(group.messages[:0], source...) - group.outOfOrder = benchmark.outOfOrder + group := NewEventsGroup(0, 1) + for _, message := range source { + group.AppendMessage(message) } dst = group.ResolveInto(benchmark.resolveTs, dst[:0]) } From 4295b183e946549c4b08f422da72e7099c49fdf1 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 21 Aug 2026 11:08:00 +0000 Subject: [PATCH 02/11] fix Signed-off-by: wk989898 --- cmd/kafka-consumer/consumer.go | 13 +++- cmd/kafka-consumer/writer.go | 63 +++++++++------ cmd/kafka-consumer/writer_test.go | 16 ++-- cmd/pulsar-consumer/consumer.go | 13 +++- cmd/pulsar-consumer/writer.go | 51 +++++++----- cmd/pulsar-consumer/writer_test.go | 28 +++++-- cmd/storage-consumer/consumer.go | 39 +++++++--- cmd/util/event_group.go | 120 ++++++++++++++++++----------- cmd/util/event_group_test.go | 44 ++++++++--- 9 files changed, 263 insertions(+), 124 deletions(-) diff --git a/cmd/kafka-consumer/consumer.go b/cmd/kafka-consumer/consumer.go index c98125ddd0..1c50034144 100644 --- a/cmd/kafka-consumer/consumer.go +++ b/cmd/kafka-consumer/consumer.go @@ -151,7 +151,10 @@ func (c *consumer) readMessage(ctx context.Context) error { log.Error("read message failed, just continue to retry", zap.Error(err)) continue } - needCommit := c.writer.WriteMessage(ctx, msg) + needCommit, err := c.writer.WriteMessage(ctx, msg) + if err != nil { + return err + } if !needCommit { continue } @@ -169,8 +172,12 @@ func (c *consumer) readMessage(ctx context.Context) error { } // Run the consumer, read data and write to the downstream target. -func (c *consumer) Run(ctx context.Context) error { - defer c.writer.cleanupEventsGroups() +func (c *consumer) Run(ctx context.Context) (err error) { + defer func() { + if cleanupErr := c.writer.cleanupEventsGroups(); err == nil && cleanupErr != nil { + err = cleanupErr + } + }() g, ctx := errgroup.WithContext(ctx) g.Go(func() error { diff --git a/cmd/kafka-consumer/writer.go b/cmd/kafka-consumer/writer.go index 1284f31927..6664ef5c7f 100644 --- a/cmd/kafka-consumer/writer.go +++ b/cmd/kafka-consumer/writer.go @@ -147,15 +147,20 @@ func (w *writer) run(ctx context.Context) error { return w.mysqlSink.Run(ctx) } -func (w *writer) cleanupEventsGroups() { +func (w *writer) cleanupEventsGroups() error { + var cleanupErr error for _, progress := range w.progresses { for _, group := range progress.eventsGroup { if err := group.Cleanup(); err != nil { log.Warn("cleanup events group spill file failed", zap.Int32("partition", progress.partition), zap.Error(err)) + if cleanupErr == nil { + cleanupErr = err + } } } } + return cleanupErr } func (w *writer) flushDDLEvent(ctx context.Context, ddl *event.DDLEvent) error { @@ -174,7 +179,10 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *event.DDLEvent) error { if !ok { continue } - messages := g.ResolveInto(commitTs, nil) + messages, err := g.ResolveInto(commitTs, nil) + if err != nil { + return err + } events := make([]*event.DMLEvent, 0, len(messages)) for _, message := range messages { events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) @@ -286,7 +294,10 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error { resolvedEvents := make([]*event.DMLEvent, 0) for _, p := range w.progresses { for _, group := range p.eventsGroup { - messages := group.ResolveInto(watermark, nil) + messages, err := group.ResolveInto(watermark, nil) + if err != nil { + return err + } events := make([]*event.DMLEvent, 0, len(messages)) for _, message := range messages { events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) @@ -331,7 +342,7 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error { // WriteMessage is to decode kafka message to event. // return true if the message is flushed to the downstream. // return error if flush messages failed. -func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) bool { +func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) (bool, error) { var ( partition = message.TopicPartition.Partition offset = message.TopicPartition.Offset @@ -366,19 +377,21 @@ func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) bool log.Info("simple protocol cached event resolved, append to the group", zap.Int64("tableID", dmlMessage.TableID), zap.Uint64("commitTs", dmlMessage.GetCommitTs()), zap.Int32("partition", partition), zap.Any("offset", offset)) - w.appendMessage2Group(dmlMessage, progress, offset) + if err := w.appendMessage2Group(dmlMessage, progress, offset); err != nil { + return false, err + } } } w.onDDL(ddl) // DDL is broadcast to all partitions, but only handle the DDL from partition-0. if partition != 0 { - return false + return false, nil } // the Query maybe empty if using simple protocol, it's comes from `bootstrap` event, no need to handle it. if ddl.Query == "" { - return false + return false, nil } w.appendDDL(ddl) log.Info("DDL event received", @@ -400,7 +413,9 @@ func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) bool break } - w.appendMessage2Group(dmlMessage, progress, offset) + if err := w.appendMessage2Group(dmlMessage, progress, offset); err != nil { + return false, err + } counter++ for { _, hasNext = progress.decoder.HasNext() @@ -416,7 +431,9 @@ func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) bool log.Debug("DML message is nil, it's cached", zap.Int32("partition", partition), zap.Any("offset", offset)) break } - w.appendMessage2Group(dmlMessage, progress, offset) + if err := w.appendMessage2Group(dmlMessage, progress, offset); err != nil { + return false, err + } counter++ } // If the message containing only one event exceeds the length limit, CDC will allow it and issue a warning. @@ -438,11 +455,11 @@ func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) bool if needFlush { return w.Write(ctx, messageType) } - return false + return false, nil } // Write will synchronously write data downstream -func (w *writer) Write(ctx context.Context, messageType common.MessageType) bool { +func (w *writer) Write(ctx context.Context, messageType common.MessageType) (bool, error) { // DDL events can be received out of commit-ts order (e.g. due to protocol-level broadcasting and // buffering differences between DDL kinds). We must execute DDLs in commit-ts order; otherwise a // "future" DDL that is not yet eligible (commitTs > watermark) can block executing earlier DDLs @@ -490,8 +507,7 @@ func (w *writer) Write(ctx context.Context, messageType common.MessageType) bool break } if err := w.flushDDLEvent(ctx, todoDDL); err != nil { - log.Panic("write DDL event failed", zap.Error(err), - zap.String("DDL", todoDDL.Query), zap.Uint64("commitTs", todoDDL.GetCommitTs())) + return false, err } } @@ -499,7 +515,7 @@ func (w *writer) Write(ctx context.Context, messageType common.MessageType) bool // since watermark is broadcast to all partitions, so that each partition can flush events individually. err := w.flushDMLEventsByWatermark(ctx) if err != nil { - log.Panic("flush dml events by the watermark failed", zap.Error(err)) + return false, err } } @@ -509,9 +525,9 @@ func (w *writer) Write(ctx context.Context, messageType common.MessageType) bool log.Info("some DDL events will be flushed in the future", zap.Uint64("watermark", watermark), zap.Int("length", len(w.ddlList))) - return false + return false, nil } - return true + return true, nil } func (w *writer) onDDL(ddl *event.DDLEvent) { @@ -611,7 +627,7 @@ func (w *writer) messageWithPartitionCheck(message *common.DMLMessage, partition }) } -func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *partitionProgress, offset kafka.Offset) { +func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *partitionProgress, offset kafka.Offset) error { // if the kafka cluster is normal, this should not hit. // else if the cluster is abnormal, the consumer may consume old message, then cause the watermark fallback. var ( @@ -631,7 +647,7 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti zap.String("schema", schema), zap.String("table", table), zap.Stringer("eventType", message.RowType), zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) - return + return nil } group := progress.eventsGroup[tableID] @@ -639,9 +655,11 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti group = util.NewEventsGroup(progress.partition, tableID) progress.eventsGroup[tableID] = group } - group.AppendMessageWithPostRestore(message, func(message *common.DMLMessage) *common.DMLMessage { + if err := group.AppendMessageWithPostRestore(message, func(message *common.DMLMessage) *common.DMLMessage { return w.messageWithPartitionCheck(message, progress.partition, offset) - }) + }); err != nil { + return err + } if commitTs < progress.watermark { log.Warn("DML event fallback row, since less than the partition watermark, append it and sort before flush", zap.Int64("tableID", tableID), zap.Int32("partition", group.Partition), @@ -651,7 +669,7 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti zap.String("schema", schema), zap.String("table", table), zap.Stringer("eventType", message.RowType), zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) - return + return nil } if commitTs >= group.HighWatermark { log.Debug("DML event append to the group", @@ -659,7 +677,7 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti zap.Uint64("commitTs", commitTs), zap.Uint64("HighWatermark", group.HighWatermark), zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), zap.Stringer("eventType", message.RowType)) - return + return nil } log.Warn("DML event commit ts fallback, append it and sort before flush", zap.Int32("partition", progress.partition), zap.Any("offset", offset), @@ -668,6 +686,7 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), zap.Stringer("eventType", message.RowType), zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) + return nil } func openDB(ctx context.Context, dsn string) (*sql.DB, error) { diff --git a/cmd/kafka-consumer/writer_test.go b/cmd/kafka-consumer/writer_test.go index 5cb904abc4..b60075d1eb 100644 --- a/cmd/kafka-consumer/writer_test.go +++ b/cmd/kafka-consumer/writer_test.go @@ -299,7 +299,9 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { w.appendMessage2Group(newDMLMessageForWriterTest(20), p, kafka.Offset(3)) p.watermark = 20 - require.True(t, w.Write(ctx, codeccommon.MessageTypeResolved)) + needCommit, err := w.Write(ctx, codeccommon.MessageTypeResolved) + require.NoError(t, err) + require.True(t, needCommit) require.Equal(t, []uint64{10, 20}, flushedCommitTs) } @@ -323,9 +325,10 @@ func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { maxMessageBytes: 1, } - needCommit := w.WriteMessage(ctx, &kafka.Message{ + needCommit, err := w.WriteMessage(ctx, &kafka.Message{ TopicPartition: kafka.TopicPartition{Partition: 0, Offset: kafka.Offset(10)}, }) + require.NoError(t, err) require.False(t, needCommit) require.Nil(t, progress.eventsGroup[1]) @@ -353,7 +356,8 @@ func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) { w.appendMessage2Group(newDMLMessageForWriterTest(10), progress, kafka.Offset(10)) require.NotNil(t, progress.eventsGroup[1]) - resolved := progress.eventsGroup[1].ResolveInto(20, nil) + resolved, err := progress.eventsGroup[1].ResolveInto(20, nil) + require.NoError(t, err) require.Len(t, resolved, 1) require.Equal(t, uint64(10), resolved[0].GetCommitTs()) } @@ -404,7 +408,8 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) { w.appendMessage2Group(codeccommon.NewDMLMessageFromEvent(newDMLEvent(200)), progress, kafka.Offset(10)) w.appendMessage2Group(codeccommon.NewDMLMessageFromEvent(newDMLEvent(100)), progress, kafka.Offset(11)) - resolved := progress.eventsGroup[1].ResolveInto(150, nil) + resolved, err := progress.eventsGroup[1].ResolveInto(150, nil) + require.NoError(t, err) require.Len(t, resolved, 1) require.Equal(t, uint64(100), resolved[0].GetCommitTs()) } @@ -452,7 +457,8 @@ func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) { w.appendMessage2Group(codeccommon.NewDMLMessageFromEvent(newDMLEvent(200)), progress, kafka.Offset(10)) w.appendMessage2Group(codeccommon.NewDMLMessageFromEvent(newDMLEvent(100)), progress, kafka.Offset(11)) - resolved := progress.eventsGroup[1].ResolveInto(150, nil) + resolved, err := progress.eventsGroup[1].ResolveInto(150, nil) + require.NoError(t, err) require.Len(t, resolved, 1) require.Equal(t, uint64(100), resolved[0].GetCommitTs()) }) diff --git a/cmd/pulsar-consumer/consumer.go b/cmd/pulsar-consumer/consumer.go index c6b3d600ec..e9284cbb7f 100644 --- a/cmd/pulsar-consumer/consumer.go +++ b/cmd/pulsar-consumer/consumer.go @@ -110,7 +110,10 @@ func (c *consumer) readMessage(ctx context.Context) error { return errors.Trace(ctx.Err()) case consumerMsg := <-msgChan: log.Debug("Received message", zap.Stringer("msgId", consumerMsg.ID()), zap.ByteString("content", consumerMsg.Payload())) - needCommit := c.writer.WriteMessage(ctx, consumerMsg) + needCommit, writeErr := c.writer.WriteMessage(ctx, consumerMsg) + if writeErr != nil { + return writeErr + } if !needCommit { continue } @@ -123,8 +126,12 @@ func (c *consumer) readMessage(ctx context.Context) error { } // Run the consumer, read data and write to the downstream target. -func (c *consumer) Run(ctx context.Context) error { - defer c.writer.cleanupEventsGroups() +func (c *consumer) Run(ctx context.Context) (err error) { + defer func() { + if cleanupErr := c.writer.cleanupEventsGroups(); err == nil && cleanupErr != nil { + err = cleanupErr + } + }() g, ctx := errgroup.WithContext(ctx) g.Go(func() error { diff --git a/cmd/pulsar-consumer/writer.go b/cmd/pulsar-consumer/writer.go index e748c10a49..136d18719b 100644 --- a/cmd/pulsar-consumer/writer.go +++ b/cmd/pulsar-consumer/writer.go @@ -139,15 +139,20 @@ func (w *writer) run(ctx context.Context) error { return w.mysqlSink.Run(ctx) } -func (w *writer) cleanupEventsGroups() { +func (w *writer) cleanupEventsGroups() error { + var cleanupErr error for _, progress := range w.progresses { for _, group := range progress.eventsGroup { if err := group.Cleanup(); err != nil { log.Warn("cleanup events group spill file failed", zap.Int32("partition", progress.partition), zap.Error(err)) + if cleanupErr == nil { + cleanupErr = err + } } } } + return cleanupErr } func (w *writer) flushDDLEvent(ctx context.Context, ddl *commonEvent.DDLEvent) error { @@ -166,7 +171,10 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *commonEvent.DDLEvent) e if !ok { continue } - messages := g.ResolveInto(commitTs, nil) + messages, err := g.ResolveInto(commitTs, nil) + if err != nil { + return err + } events := make([]*commonEvent.DMLEvent, 0, len(messages)) for _, message := range messages { events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) @@ -278,7 +286,10 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error { resolvedEvents := make([]*commonEvent.DMLEvent, 0) for _, p := range w.progresses { for _, group := range p.eventsGroup { - messages := group.ResolveInto(watermark, nil) + messages, err := group.ResolveInto(watermark, nil) + if err != nil { + return err + } events := make([]*commonEvent.DMLEvent, 0, len(messages)) for _, message := range messages { events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) @@ -321,7 +332,7 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error { // WriteMessage is to decode pulsar message to event. // return true if the message is flushed to the downstream. // return error if flush messages failed. -func (w *writer) WriteMessage(ctx context.Context, message pulsar.Message) bool { +func (w *writer) WriteMessage(ctx context.Context, message pulsar.Message) (bool, error) { progress := w.progresses[0] progress.decoder.AddKeyValue([]byte(message.Key()), message.Payload()) @@ -349,7 +360,7 @@ func (w *writer) WriteMessage(ctx context.Context, message pulsar.Message) bool // the Query maybe empty if using simple protocol, it's comes from `bootstrap` event, no need to handle it. if ddl.Query == "" { - return false + return false, nil } w.appendDDL(ddl) log.Info("DDL event received", @@ -362,18 +373,20 @@ func (w *writer) WriteMessage(ctx context.Context, message pulsar.Message) bool if dmlMessage == nil { log.Panic("DML message is nil, it's not expected") } - w.appendMessage2Group(dmlMessage, progress) + if err := w.appendMessage2Group(dmlMessage, progress); err != nil { + return false, err + } default: log.Panic("unknown message type", zap.Any("messageType", messageType)) } if needFlush { return w.Write(ctx, messageType) } - return false + return false, nil } // Write will synchronously write data downstream -func (w *writer) Write(ctx context.Context, messageType common.MessageType) bool { +func (w *writer) Write(ctx context.Context, messageType common.MessageType) (bool, error) { // DDL events can be received out of commit-ts order (e.g. due to protocol-level broadcasting and // buffering differences between DDL kinds). We must execute DDLs in commit-ts order; otherwise a // "future" DDL that is not yet eligible (commitTs > watermark) can block executing earlier DDLs @@ -420,8 +433,7 @@ func (w *writer) Write(ctx context.Context, messageType common.MessageType) bool break } if err := w.flushDDLEvent(ctx, todoDDL); err != nil { - log.Panic("write DDL event failed", zap.Error(err), - zap.String("DDL", todoDDL.Query), zap.Uint64("commitTs", todoDDL.GetCommitTs())) + return false, err } } @@ -429,7 +441,7 @@ func (w *writer) Write(ctx context.Context, messageType common.MessageType) bool // since watermark is broadcast to all partitions, so that each partition can flush events individually. err := w.flushDMLEventsByWatermark(ctx) if err != nil { - log.Panic("flush dml events by the watermark failed", zap.Error(err)) + return false, err } } @@ -439,9 +451,9 @@ func (w *writer) Write(ctx context.Context, messageType common.MessageType) bool log.Info("some DDL events will be flushed in the future", zap.Uint64("watermark", watermark), zap.Int("length", len(w.ddlList))) - return false + return false, nil } - return true + return true, nil } func (w *writer) onDDL(ddl *commonEvent.DDLEvent) { @@ -502,7 +514,7 @@ func (w *writer) addPartitionTable(schema, table string) { w.partitionTableAccessor.Add(schema, table) } -func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *partitionProgress) { +func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *partitionProgress) error { var ( tableID = message.TableID schema = message.Schema @@ -519,7 +531,7 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti zap.String("schema", schema), zap.String("table", table), zap.Stringer("eventType", message.RowType), zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) - return + return nil } group := progress.eventsGroup[tableID] @@ -527,7 +539,9 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti group = util.NewEventsGroup(progress.partition, tableID) progress.eventsGroup[tableID] = group } - group.AppendMessage(message) + if err := group.AppendMessage(message); err != nil { + return err + } if commitTs < progress.watermark { log.Warn("DML event fallback row, since less than the partition watermark, append it and sort before flush", zap.Int64("tableID", tableID), zap.Int32("partition", group.Partition), @@ -536,14 +550,14 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti zap.String("schema", schema), zap.String("table", table), zap.Stringer("eventType", message.RowType), zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) - return + return nil } if commitTs >= group.HighWatermark { log.Debug("DML event append to the group", zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), zap.Stringer("eventType", message.RowType)) - return + return nil } log.Warn("DML event commit ts fallback, append it and sort before flush", zap.Int32("partition", progress.partition), @@ -552,4 +566,5 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), zap.Stringer("eventType", message.RowType), zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) + return nil } diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go index 6775aecfba..6189b8511e 100644 --- a/cmd/pulsar-consumer/writer_test.go +++ b/cmd/pulsar-consumer/writer_test.go @@ -294,7 +294,9 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { w.appendMessage2Group(newDMLMessageForWriterTest(20), p) p.watermark = 20 - require.True(t, w.Write(ctx, codeccommon.MessageTypeResolved)) + needCommit, err := w.Write(ctx, codeccommon.MessageTypeResolved) + require.NoError(t, err) + require.True(t, needCommit) require.Equal(t, []uint64{10, 20}, flushedCommitTs) } @@ -326,7 +328,8 @@ func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { protocol: config.ProtocolCanalJSON, } - needCommit := w.WriteMessage(ctx, fakePulsarMessage{key: "k", payload: []byte(`{"fake":"row"}`)}) + needCommit, err := w.WriteMessage(ctx, fakePulsarMessage{key: "k", payload: []byte(`{"fake":"row"}`)}) + require.NoError(t, err) require.False(t, needCommit) require.Nil(t, progress.eventsGroup[1]) @@ -349,7 +352,8 @@ func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) { w.appendMessage2Group(newDMLMessageForWriterTest(10), progress) require.NotNil(t, progress.eventsGroup[1]) - resolved := progress.eventsGroup[1].ResolveInto(20, nil) + resolved, err := progress.eventsGroup[1].ResolveInto(20, nil) + require.NoError(t, err) require.Len(t, resolved, 1) require.Equal(t, uint64(10), resolved[0].GetCommitTs()) } @@ -385,7 +389,8 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTable(t *testing.T) { w.appendMessage2Group(newDMLMessageForWriterTest(200), progress) w.appendMessage2Group(newDMLMessageForWriterTest(100), progress) - resolved := progress.eventsGroup[1].ResolveInto(150, nil) + resolved, err := progress.eventsGroup[1].ResolveInto(150, nil) + require.NoError(t, err) require.Len(t, resolved, 1) require.Equal(t, uint64(100), resolved[0].GetCommitTs()) } @@ -419,21 +424,28 @@ func TestWriteMessageSpillsDMLImmediately(t *testing.T) { protocol: config.ProtocolCanalJSON, } - needCommit := w.WriteMessage(ctx, fakePulsarMessage{key: "k", payload: []byte(`{"fake":"row"}`)}) + needCommit, err := w.WriteMessage(ctx, fakePulsarMessage{key: "k", payload: []byte(`{"fake":"row"}`)}) + require.NoError(t, err) require.False(t, needCommit) require.Equal(t, 1, decoder.addKeyValueCount) require.Equal(t, 1, decoder.hasNextCount) require.Equal(t, 1, decoder.nextDMLMessageCount) require.Equal(t, 1, decoder.toDMLEventCount) - require.Len(t, progress.eventsGroup[1].ResolveInto(99, nil), 0) + resolved, err := progress.eventsGroup[1].ResolveInto(99, nil) + require.NoError(t, err) + require.Len(t, resolved, 0) progress.watermark = 100 - require.True(t, w.Write(ctx, codeccommon.MessageTypeResolved)) + needCommit, err = w.Write(ctx, codeccommon.MessageTypeResolved) + require.NoError(t, err) + require.True(t, needCommit) require.Equal(t, 1, decoder.addKeyValueCount) require.Equal(t, 1, decoder.hasNextCount) require.Equal(t, 1, decoder.nextDMLMessageCount) require.Equal(t, 1, decoder.toDMLEventCount) - require.Empty(t, progress.eventsGroup[1].ResolveInto(100, nil)) + resolved, err = progress.eventsGroup[1].ResolveInto(100, nil) + require.NoError(t, err) + require.Empty(t, resolved) require.Equal(t, []byte(`{"fake":"row"}`), decoder.lastValue) } diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index 9567ee57e0..71ed1a6d60 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -282,7 +282,7 @@ func (c *consumer) getNewFiles( return tableDMLMap, err } -func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAcrossNodes bool) { +func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAcrossNodes bool) error { var ( tableID = message.TableID schema = message.Schema @@ -295,20 +295,24 @@ func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAc c.eventsGroup[tableID] = group } if commitTs >= group.HighWatermark { - group.AppendMessage(message) + if err := group.AppendMessage(message); err != nil { + return err + } log.Debug("DML event append to the group", zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), zap.Stringer("eventType", message.RowType)) - return + return nil } if enableTableAcrossNodes { log.Warn("DML events fallback, but enableTableAcrossNodes is true, still append it", zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), zap.Stringer("eventType", message.RowType)) - group.AppendMessage(message) - return + if err := group.AppendMessage(message); err != nil { + return err + } + return nil } log.Warn("dml event commit ts fallback, ignore", zap.Uint64("commitTs", commitTs), @@ -316,6 +320,7 @@ func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAc zap.String("schema", schema), zap.String("table", table), ) + return nil } // appendDMLEvents decodes RowChangedEvents from file content and append them to event group. @@ -372,7 +377,9 @@ func (c *consumer) appendDMLEvents( c.dmlCount.Add(1) message := decoder.NextDMLMessage() - c.appendMessage2Group(messageWithPhysicalTableID(message, tableID), fileIdx.EnableTableAcrossNodes) + if err := c.appendMessage2Group(messageWithPhysicalTableID(message, tableID), fileIdx.EnableTableAcrossNodes); err != nil { + return err + } filteredCnt++ } } @@ -398,7 +405,10 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error { if group == nil { return nil } - messages := group.GetAllMessages() + messages, err := group.GetAllMessages() + if err != nil { + return err + } if len(messages) == 0 { return nil } @@ -449,12 +459,17 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error { } } -func (c *consumer) cleanupEventsGroups() { +func (c *consumer) cleanupEventsGroups() error { + var cleanupErr error for _, group := range c.eventsGroup { if err := group.Cleanup(); err != nil { log.Warn("cleanup events group spill file failed", zap.Error(err)) + if cleanupErr == nil { + cleanupErr = err + } } } + return cleanupErr } func (c *consumer) parseDMLIndexFile(ctx context.Context, path string, dmlkey cloudstorage.DMLPathKey) { @@ -774,8 +789,12 @@ func (c *consumer) handle(ctx context.Context) error { } } -func (c *consumer) run(ctx context.Context) error { - defer c.cleanupEventsGroups() +func (c *consumer) run(ctx context.Context) (err error) { + defer func() { + if cleanupErr := c.cleanupEventsGroups(); err == nil && cleanupErr != nil { + err = cleanupErr + } + }() g, ctx := errgroup.WithContext(ctx) g.Go(func() error { diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index e310c429b9..9b8a13941e 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -62,8 +62,8 @@ func NewEventsGroup(partition int32, tableID int64) *EventsGroup { // AppendMessage materializes a message and appends it to a local spill file. DMLMessage carries a // decoder closure, so persisting its reconstructed event is necessary to release the decoder input // retained by that closure. -func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage) { - g.appendMessage(message, nil) +func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage) error { + return g.appendMessage(message, nil) } // AppendMessageWithPostRestore appends a message and applies postRestore after it is read back from @@ -72,38 +72,39 @@ func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage) { func (g *EventsGroup) AppendMessageWithPostRestore( message *codeccommon.DMLMessage, postRestore func(*codeccommon.DMLMessage) *codeccommon.DMLMessage, -) { - g.appendMessage(message, postRestore) +) error { + return g.appendMessage(message, postRestore) } func (g *EventsGroup) appendMessage( message *codeccommon.DMLMessage, postRestore func(*codeccommon.DMLMessage) *codeccommon.DMLMessage, -) { - commitTs := message.GetCommitTs() - if len(g.messages) > 0 && commitTs < g.messages[len(g.messages)-1].commitTs { - g.outOfOrder = true - } - if commitTs > g.HighWatermark { - g.HighWatermark = commitTs + +) error { + if message == nil { + return errors.ErrSpillFileOp.FastGenByArgs("cannot spill nil DML message") } + commitTs := message.GetCommitTs() data, row, err := marshalDMLMessage(message) if err != nil { - log.Panic("marshal DML message for spill failed", - zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + return err } if g.spillFile == nil { g.spillFile, err = spill.NewRecordFile(os.TempDir(), eventsGroupSpillPattern) if err != nil { - log.Panic("create events group spill file failed", - zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + return err } } handle, err := g.spillFile.Append(data) if err != nil { - log.Panic("write DML message to spill file failed", - zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + return err + } + if len(g.messages) > 0 && commitTs < g.messages[len(g.messages)-1].commitTs { + g.outOfOrder = true + } + if commitTs > g.HighWatermark { + g.HighWatermark = commitTs } g.messages = append(g.messages, spilledMessage{ commitTs: commitTs, @@ -113,14 +114,15 @@ func (g *EventsGroup) appendMessage( // Codec decoders use this callback to release their pooled chunks. The event is durable in the // spill file now, so the original in-memory event is no longer needed. row.PostFlush() + return nil } // ResolveInto appends all messages with CommitTs <= resolve into dst in commit-ts order and removes // them from the group. Resolved messages are restored from the spill file only when downstream needs // them, keeping the buffered group out of heap memory. -func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) []*codeccommon.DMLMessage { +func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) ([]*codeccommon.DMLMessage, error) { if len(g.messages) == 0 { - return dst + return dst, nil } if g.outOfOrder { @@ -141,19 +143,20 @@ func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) g.outOfOrder = false } if resolvedCount == 0 { - return dst + return dst, nil + } + if g.spillFile == nil { + return dst, errors.ErrSpillFileOp.FastGenByArgs("events group spill file is missing") } for _, message := range g.messages[:resolvedCount] { data, err := g.spillFile.Read(message.handle) if err != nil { - log.Panic("read DML message from spill file failed", - zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + return dst, err } restored, err := unmarshalDMLMessage(data) if err != nil { - log.Panic("unmarshal DML message from spill file failed", - zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + return dst, err } if message.postRestore != nil { restored = message.postRestore(restored) @@ -166,8 +169,7 @@ func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) g.messages = g.messages[:remainingCount] if len(g.messages) == 0 { if err := g.spillFile.Cleanup(); err != nil { - log.Panic("cleanup events group spill file failed", - zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), zap.Error(err)) + return dst, err } g.spillFile = nil } @@ -178,11 +180,11 @@ func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) zap.Int("resolved", resolvedCount), zap.Int("remained", len(g.messages)), zap.Uint64("resolveTs", resolve), zap.Uint64("firstCommitTs", firstCommitTs)) } - return dst + return dst, nil } // GetAllMessages gets all messages. -func (g *EventsGroup) GetAllMessages() []*codeccommon.DMLMessage { +func (g *EventsGroup) GetAllMessages() ([]*codeccommon.DMLMessage, error) { return g.ResolveInto(math.MaxUint64, nil) } @@ -192,10 +194,13 @@ func (g *EventsGroup) Cleanup() error { return nil } err := g.spillFile.Cleanup() + if err != nil { + return err + } g.spillFile = nil clear(g.messages) g.messages = g.messages[:0] - return err + return nil } func marshalDMLMessage(message *codeccommon.DMLMessage) (data []byte, row *commonEvent.DMLEvent, err error) { @@ -223,6 +228,8 @@ func marshalDMLMessage(message *codeccommon.DMLMessage) (data []byte, row *commo if row.Rows != nil && row.Rows.NumRows() > 0 { return nil, nil, err } + log.Warn("spill DML event without table info", + zap.Int64("tableID", row.PhysicalTableID), zap.Uint64("commitTs", row.CommitTs), zap.Error(err)) tableInfoData = nil } else { tableInfoStored = true @@ -236,6 +243,8 @@ func marshalDMLMessage(message *codeccommon.DMLMessage) (data []byte, row *commo if row.Rows.NumRows() > 0 { return nil, nil, err } + log.Warn("spill DML event without row data", + zap.Int64("tableID", row.PhysicalTableID), zap.Uint64("commitTs", row.CommitTs), zap.Error(err)) rowsData = nil } } @@ -245,7 +254,7 @@ func marshalDMLMessage(message *codeccommon.DMLMessage) (data []byte, row *commo return nil, nil, errors.WrapError(errors.ErrSpillFileOp, err, "marshal DML checksums") } - data = make([]byte, 0, 7*8+len(eventData)+len(tableInfoData)+len(rowsData)+len(checksumData)) + data = make([]byte, 0, 10*8+len(eventData)+len(tableInfoData)+len(rowsData)+len(checksumData)+len(message.Schema)+len(message.Table)) data = appendUint64(data, uint64(len(eventData))) data = append(data, eventData...) data = appendUint64(data, uint64(len(tableInfoData))) @@ -254,6 +263,11 @@ func marshalDMLMessage(message *codeccommon.DMLMessage) (data []byte, row *commo data = append(data, rowsData...) data = appendUint64(data, uint64(len(checksumData))) data = append(data, checksumData...) + data = appendUint64(data, uint64(len(message.Schema))) + data = append(data, message.Schema...) + data = appendUint64(data, uint64(len(message.Table))) + data = append(data, message.Table...) + data = appendUint64(data, uint64(message.RowType)) if row.Rows != nil { data = appendUint64(data, 1) } else { @@ -309,6 +323,21 @@ func unmarshalDMLMessage(data []byte) (*codeccommon.DMLMessage, error) { if err != nil { return nil, err } + schemaData, data, err := readSpilledField(data) + if err != nil { + return nil, err + } + tableData, data, err := readSpilledField(data) + if err != nil { + return nil, err + } + rowType, data, err := readSpilledUint64(data) + if err != nil { + return nil, err + } + if rowType > uint64(^commonType.RowType(0)) { + return nil, errors.ErrSpillFileOp.FastGenByArgs("invalid DML spill row type") + } rowsPresent, data, err := readSpilledUint64(data) if err != nil { return nil, err @@ -346,26 +375,31 @@ func unmarshalDMLMessage(data []byte) (*codeccommon.DMLMessage, error) { if row.TableInfo != nil { fieldTypes = row.TableInfo.GetFieldSlice() } - row.Rows, _ = chunk.NewCodec(fieldTypes).Decode(rowsData) + rows, err := unmarshalDMLRows(rowsData, fieldTypes) + if err != nil { + return nil, err + } + row.Rows = rows } row.TableInfoVersion = tableInfoVersion row.ReplicatingTs = replicatingTs if err := json.Unmarshal(checksumData, &row.Checksum); err != nil { return nil, errors.WrapError(errors.ErrSpillFileOp, err, "unmarshal DML checksums") } - if len(row.RowTypes) == 0 { - return nil, errors.ErrSpillFileOp.FastGenByArgs("spilled DML event has no row type") - } - - var schema, table string - if row.TableInfo != nil { - schema = row.TableInfo.GetSchemaName() - table = row.TableInfo.GetTableName() - } + return codeccommon.NewDMLMessage(row.PhysicalTableID, string(schemaData), string(tableData), row.CommitTs, + commonType.RowType(rowType), func() *commonEvent.DMLEvent { + return row + }), nil +} - return codeccommon.NewDMLMessage(row.PhysicalTableID, schema, table, row.CommitTs, row.RowTypes[0], func() *commonEvent.DMLEvent { - return row - }), nil +func unmarshalDMLRows(data []byte, fieldTypes []*types.FieldType) (rows *chunk.Chunk, err error) { + defer func() { + if recover() != nil { + err = errors.ErrSpillFileOp.FastGenByArgs("decode DML spill rows") + } + }() + rows, _ = chunk.NewCodec(fieldTypes).Decode(data) + return rows, nil } func appendUint64(data []byte, value uint64) []byte { diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index 4f37b8dbed..c782b65773 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -66,7 +66,8 @@ func TestEventsGroupResolveIntoAppendsAndCleansResolvedSpillRecords(t *testing.T spillPath := group.spillFile.Path() var dst []*codeccommon.DMLMessage - dst = group.ResolveInto(2, dst) + dst, err := group.ResolveInto(2, dst) + require.NoError(t, err) require.Len(t, dst, 2) require.Equal(t, m1.GetCommitTs(), dst[0].GetCommitTs()) @@ -76,9 +77,10 @@ func TestEventsGroupResolveIntoAppendsAndCleansResolvedSpillRecords(t *testing.T require.Equal(t, m3.GetCommitTs(), group.messages[0].commitTs) require.FileExists(t, spillPath) - group.GetAllMessages() + _, err = group.GetAllMessages() + require.NoError(t, err) require.Nil(t, group.spillFile) - _, err := os.Stat(spillPath) + _, err = os.Stat(spillPath) require.True(t, os.IsNotExist(err)) } @@ -92,7 +94,8 @@ func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) { group.AppendMessage(m2) dst := make([]*codeccommon.DMLMessage, 0, 1) - dst = group.ResolveInto(5, dst) + dst, err := group.ResolveInto(5, dst) + require.NoError(t, err) require.Len(t, dst, 0) require.Len(t, group.messages, 2) @@ -111,7 +114,8 @@ func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) { spillPath := group.spillFile.Path() var dst []*codeccommon.DMLMessage - dst = group.ResolveInto(100, dst) + dst, err := group.ResolveInto(100, dst) + require.NoError(t, err) require.Len(t, dst, 2) require.Equal(t, m1.GetCommitTs(), dst[0].GetCommitTs()) @@ -119,7 +123,7 @@ func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) { require.Len(t, group.messages, 0) require.Nil(t, group.spillFile) - _, err := os.Stat(spillPath) + _, err = os.Stat(spillPath) require.True(t, os.IsNotExist(err)) } @@ -133,7 +137,8 @@ func TestEventsGroupResolveIntoSortsOutOfOrderResolvedMessages(t *testing.T) { group.AppendMessage(m3) var dst []*codeccommon.DMLMessage - dst = group.ResolveInto(25, dst) + dst, err := group.ResolveInto(25, dst) + require.NoError(t, err) require.Len(t, dst, 2) require.Equal(t, m2.GetCommitTs(), dst[0].GetCommitTs()) @@ -153,7 +158,8 @@ func TestEventsGroupResolveIntoKeepsSameCommitTsStable(t *testing.T) { group.AppendMessage(m3) var dst []*codeccommon.DMLMessage - dst = group.ResolveInto(20, dst) + dst, err := group.ResolveInto(20, dst) + require.NoError(t, err) require.Len(t, dst, 3) require.Equal(t, m2.GetCommitTs(), dst[0].GetCommitTs()) @@ -171,7 +177,8 @@ func TestEventsGroupGetAllMessagesSortsOutOfOrderMessages(t *testing.T) { group.AppendMessage(m2) group.AppendMessage(m3) - messages := group.GetAllMessages() + messages, err := group.GetAllMessages() + require.NoError(t, err) require.Len(t, messages, 3) require.Equal(t, m2.GetCommitTs(), messages[0].GetCommitTs()) @@ -205,7 +212,8 @@ func TestEventsGroupRestoresSpilledEventRowsAndTableInfo(t *testing.T) { group := NewEventsGroup(0, 1) group.AppendMessage(codeccommon.NewDMLMessageFromEvent(event)) - messages := group.GetAllMessages() + messages, err := group.GetAllMessages() + require.NoError(t, err) require.Len(t, messages, 1) restored := messages[0].ToDMLEvent() require.Equal(t, uint64(100), restored.CommitTs) @@ -252,12 +260,24 @@ func BenchmarkEventsGroupResolveInto(b *testing.B) { b.ReportAllocs() b.ResetTimer() + b.StopTimer() for b.Loop() { group := NewEventsGroup(0, 1) for _, message := range source { - group.AppendMessage(message) + if err := group.AppendMessage(message); err != nil { + b.Fatal(err) + } + } + b.StartTimer() + var err error + dst, err = group.ResolveInto(benchmark.resolveTs, dst[:0]) + b.StopTimer() + if err != nil { + b.Fatal(err) + } + if err := group.Cleanup(); err != nil { + b.Fatal(err) } - dst = group.ResolveInto(benchmark.resolveTs, dst[:0]) } }) } From 6f5720719fcc60228b26e266c105f0514bee224a Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 21 Aug 2026 11:10:03 +0000 Subject: [PATCH 03/11] update Signed-off-by: wk989898 --- cmd/util/event_group.go | 5 +---- cmd/util/event_group_test.go | 13 +++++++++++++ 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index 9b8a13941e..80e63c940f 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -86,7 +86,7 @@ func (g *EventsGroup) appendMessage( } commitTs := message.GetCommitTs() - data, row, err := marshalDMLMessage(message) + data, _, err := marshalDMLMessage(message) if err != nil { return err } @@ -111,9 +111,6 @@ func (g *EventsGroup) appendMessage( handle: handle, postRestore: postRestore, }) - // Codec decoders use this callback to release their pooled chunks. The event is durable in the - // spill file now, so the original in-memory event is no longer needed. - row.PostFlush() return nil } diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index c782b65773..fa6883bf0c 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -225,6 +225,19 @@ func TestEventsGroupRestoresSpilledEventRowsAndTableInfo(t *testing.T) { require.Equal(t, int64(42), restored.Rows.GetRow(0).GetInt64(0)) } +func TestEventsGroupSpillDoesNotSignalDownstreamCallbacks(t *testing.T) { + event := newTestDMLEvent(100, common.RowTypeInsert) + var enqueued, flushed int + event.AddPostEnqueueFunc(func() { enqueued++ }) + event.AddPostFlushFunc(func() { flushed++ }) + + group := NewEventsGroup(0, 1) + defer func() { require.NoError(t, group.Cleanup()) }() + require.NoError(t, group.AppendMessage(codeccommon.NewDMLMessageFromEvent(event))) + require.Zero(t, enqueued) + require.Zero(t, flushed) +} + func BenchmarkEventsGroupResolveInto(b *testing.B) { const messageCount = 16 * 1024 From d84664e6c6f45c4c9d6f774fd204194b53d6ba8b Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 21 Aug 2026 11:13:33 +0000 Subject: [PATCH 04/11] fmt Signed-off-by: wk989898 --- cmd/util/event_group.go | 1 - 1 file changed, 1 deletion(-) diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index 80e63c940f..f9349b2def 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -79,7 +79,6 @@ func (g *EventsGroup) AppendMessageWithPostRestore( func (g *EventsGroup) appendMessage( message *codeccommon.DMLMessage, postRestore func(*codeccommon.DMLMessage) *codeccommon.DMLMessage, - ) error { if message == nil { return errors.ErrSpillFileOp.FastGenByArgs("cannot spill nil DML message") From 74220635482c34837f6f076e5572ee5fe31cf0fd Mon Sep 17 00:00:00 2001 From: wk989898 Date: Mon, 24 Aug 2026 03:16:13 +0000 Subject: [PATCH 05/11] fix Signed-off-by: wk989898 --- cmd/util/event_group.go | 22 ++++++++++++-- cmd/util/event_group_test.go | 59 ++++++++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 2 deletions(-) diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index f9349b2def..852319a1b3 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -211,7 +211,11 @@ func marshalDMLMessage(message *codeccommon.DMLMessage) (data []byte, row *commo if row.Version == 0 { row.Version = commonEvent.DMLEventVersion1 } - eventData, err := row.Marshal() + // Rows can be shared by several DML events. Persist only this event's rows + // below, so its offset must be reset in the serialized event as well. + event := *row + event.PreviousTotalOffset = 0 + eventData, err := event.Marshal() if err != nil { return nil, nil, errors.WrapError(errors.ErrSpillFileOp, err, "marshal DML event") } @@ -299,7 +303,21 @@ func marshalDMLRows(row *commonEvent.DMLEvent, tableInfoStored bool) (data []byt if tableInfoStored { fieldTypes = row.TableInfo.GetFieldSlice() } - return chunk.NewCodec(fieldTypes).Encode(row.Rows), nil + begin := row.PreviousTotalOffset + end := row.Rows.NumRows() + if len(row.RowTypes) != 0 { + end = begin + len(row.RowTypes) + } + if begin < 0 || end < begin || end > row.Rows.NumRows() { + return nil, errors.ErrSpillFileOp.FastGenByArgs("DML event rows are outside the shared chunk") + } + if !tableInfoStored && begin != 0 { + return nil, errors.ErrSpillFileOp.FastGenByArgs("DML event rows require table info") + } + + rows := chunk.NewChunkWithCapacity(fieldTypes, end-begin) + rows.Append(row.Rows, begin, end) + return chunk.NewCodec(fieldTypes).Encode(rows), nil } func unmarshalDMLMessage(data []byte) (*codeccommon.DMLMessage, error) { diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index fa6883bf0c..a446f308f0 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -225,6 +225,65 @@ func TestEventsGroupRestoresSpilledEventRowsAndTableInfo(t *testing.T) { require.Equal(t, int64(42), restored.Rows.GetRow(0).GetInt64(0)) } +func TestEventsGroupRestoresRowsFromSharedChunk(t *testing.T) { + tableInfo := common.WrapTableInfo("test", &model.TableInfo{ + ID: 1, + Name: ast.NewCIStr("t"), + Columns: []*model.ColumnInfo{ + {ID: 1, Name: ast.NewCIStr("id"), FieldType: *types.NewFieldType(mysql.TypeLonglong)}, + {ID: 2, Name: ast.NewCIStr("v1"), FieldType: *types.NewFieldType(mysql.TypeLong)}, + }, + }) + rows := chunk.NewChunkWithCapacity(tableInfo.GetFieldSlice(), 4) + for _, value := range []struct { + null bool + value int64 + }{ + {null: true}, {value: 1}, {null: true}, {value: 2}, + } { + rows.AppendInt64(0, 42) + if value.null { + rows.AppendNull(1) + } else { + rows.AppendInt64(1, value.value) + } + } + + group := NewEventsGroup(0, 1) + for _, offset := range []int{0, 2} { + event := commonEvent.NewDMLEvent(common.NewDispatcherID(), 1, 90, 100, tableInfo) + event.Rows = rows + event.RowTypes = []common.RowType{common.RowTypeInsert, common.RowTypeInsert} + event.Length = 2 + event.PreviousTotalOffset = offset + require.NoError(t, group.AppendMessage(codeccommon.NewDMLMessageFromEvent(event))) + } + + messages, err := group.GetAllMessages() + require.NoError(t, err) + require.Len(t, messages, 2) + for _, message := range messages { + restored := message.ToDMLEvent() + require.Zero(t, restored.PreviousTotalOffset) + require.Len(t, restored.RowTypes, restored.Rows.NumRows()) + } + + second := messages[1].ToDMLEvent() + require.True(t, second.Rows.GetRow(0).IsNull(1)) + require.Equal(t, int64(2), second.Rows.GetRow(1).GetInt64(1)) + + var events []*commonEvent.DMLEvent + for _, message := range messages { + events = AppendOrMergeDMLEvent(events, message.ToDMLEvent()) + } + require.Len(t, events, 1) + require.Len(t, events[0].RowTypes, events[0].Rows.NumRows()) + require.True(t, events[0].Rows.GetRow(0).IsNull(1)) + require.Equal(t, int64(1), events[0].Rows.GetRow(1).GetInt64(1)) + require.True(t, events[0].Rows.GetRow(2).IsNull(1)) + require.Equal(t, int64(2), events[0].Rows.GetRow(3).GetInt64(1)) +} + func TestEventsGroupSpillDoesNotSignalDownstreamCallbacks(t *testing.T) { event := newTestDMLEvent(100, common.RowTypeInsert) var enqueued, flushed int From 527b474298fb808378fd7e88e15a3f1e1d7c8196 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Mon, 24 Aug 2026 09:04:41 +0000 Subject: [PATCH 06/11] fix Signed-off-by: wk989898 --- cmd/kafka-consumer/writer.go | 15 +++++++++------ cmd/kafka-consumer/writer_test.go | 4 ++-- cmd/pulsar-consumer/writer.go | 15 +++++++++------ cmd/pulsar-consumer/writer_test.go | 4 ++-- cmd/storage-consumer/consumer.go | 4 +++- 5 files changed, 25 insertions(+), 17 deletions(-) diff --git a/cmd/kafka-consumer/writer.go b/cmd/kafka-consumer/writer.go index 6664ef5c7f..5712eac585 100644 --- a/cmd/kafka-consumer/writer.go +++ b/cmd/kafka-consumer/writer.go @@ -183,11 +183,14 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *event.DDLEvent) error { if err != nil { return err } - events := make([]*event.DMLEvent, 0, len(messages)) + // A commit-ts is only a timestamp in a multi-source stream. It does + // not identify one transaction, so events from different upstreams can + // legitimately have the same value while carrying different schema + // snapshots. Keep restored events separate instead of merging their + // chunks by commit-ts. for _, message := range messages { - events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) + resolvedEvents = append(resolvedEvents, message.ToDMLEvent()) } - resolvedEvents = append(resolvedEvents, events...) } } @@ -298,11 +301,11 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error { if err != nil { return err } - events := make([]*event.DMLEvent, 0, len(messages)) + // See flushDDLEvent: events with an identical commit-ts can belong to + // different upstream transactions in a multi-source stream. for _, message := range messages { - events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) + resolvedEvents = append(resolvedEvents, message.ToDMLEvent()) } - resolvedEvents = append(resolvedEvents, events...) } } total := len(resolvedEvents) diff --git a/cmd/kafka-consumer/writer_test.go b/cmd/kafka-consumer/writer_test.go index b60075d1eb..b7f0bb2ce7 100644 --- a/cmd/kafka-consumer/writer_test.go +++ b/cmd/kafka-consumer/writer_test.go @@ -276,7 +276,7 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) { flushedCommitTs = append(flushedCommitTs, event.GetCommitTs()) event.PostFlush() - }).Times(2) + }).Times(3) replicaCfg := config.GetDefaultReplicaConfig() eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false) @@ -302,7 +302,7 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { needCommit, err := w.Write(ctx, codeccommon.MessageTypeResolved) require.NoError(t, err) require.True(t, needCommit) - require.Equal(t, []uint64{10, 20}, flushedCommitTs) + require.Equal(t, []uint64{10, 20, 20}, flushedCommitTs) } func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { diff --git a/cmd/pulsar-consumer/writer.go b/cmd/pulsar-consumer/writer.go index 136d18719b..aad1f06198 100644 --- a/cmd/pulsar-consumer/writer.go +++ b/cmd/pulsar-consumer/writer.go @@ -175,11 +175,14 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *commonEvent.DDLEvent) e if err != nil { return err } - events := make([]*commonEvent.DMLEvent, 0, len(messages)) + // A commit-ts is only a timestamp in a multi-source stream. It does + // not identify one transaction, so events from different upstreams can + // legitimately have the same value while carrying different schema + // snapshots. Keep restored events separate instead of merging their + // chunks by commit-ts. for _, message := range messages { - events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) + resolvedEvents = append(resolvedEvents, message.ToDMLEvent()) } - resolvedEvents = append(resolvedEvents, events...) } } @@ -290,11 +293,11 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error { if err != nil { return err } - events := make([]*commonEvent.DMLEvent, 0, len(messages)) + // See flushDDLEvent: events with an identical commit-ts can belong to + // different upstream transactions in a multi-source stream. for _, message := range messages { - events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) + resolvedEvents = append(resolvedEvents, message.ToDMLEvent()) } - resolvedEvents = append(resolvedEvents, events...) } } total := len(resolvedEvents) diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go index 6189b8511e..645f308176 100644 --- a/cmd/pulsar-consumer/writer_test.go +++ b/cmd/pulsar-consumer/writer_test.go @@ -276,7 +276,7 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) { flushedCommitTs = append(flushedCommitTs, event.GetCommitTs()) event.PostFlush() - }).Times(2) + }).Times(3) p := &partitionProgress{ partition: 0, @@ -297,7 +297,7 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { needCommit, err := w.Write(ctx, codeccommon.MessageTypeResolved) require.NoError(t, err) require.True(t, needCommit) - require.Equal(t, []uint64{10, 20}, flushedCommitTs) + require.Equal(t, []uint64{10, 20, 20}, flushedCommitTs) } func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index 71ed1a6d60..6d902407e0 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -412,9 +412,11 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error { if len(messages) == 0 { return nil } + // A commit-ts is only a timestamp in a multi-source stream. Keep restored + // events separate instead of merging chunks from independent transactions. events := make([]*event.DMLEvent, 0, len(messages)) for _, message := range messages { - events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) + events = append(events, message.ToDMLEvent()) } total := len(events) if total == 0 { From 8c2648d2bc269b8fbc6106f94320587dc0e0dcc2 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Mon, 24 Aug 2026 09:08:57 +0000 Subject: [PATCH 07/11] chore Signed-off-by: wk989898 --- cmd/util/event_group.go | 26 ---------------------- cmd/util/event_group_test.go | 42 ------------------------------------ 2 files changed, 68 deletions(-) diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index 852319a1b3..8b3af9606d 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -439,29 +439,3 @@ func readSpilledUint64(data []byte) (uint64, []byte, error) { } return binary.BigEndian.Uint64(data[:8]), data[8:], nil } - -// AppendOrMergeDMLEvent appends a DML event, or merges it into the previous event -// when both events belong to the same table group and have the same commit-ts. -func AppendOrMergeDMLEvent(events []*commonEvent.DMLEvent, row *commonEvent.DMLEvent) []*commonEvent.DMLEvent { - var lastDMLEvent *commonEvent.DMLEvent - if len(events) > 0 { - lastDMLEvent = events[len(events)-1] - } - - if lastDMLEvent == nil || lastDMLEvent.GetCommitTs() < row.GetCommitTs() { - return append(events, row) - } - - if lastDMLEvent.GetCommitTs() == row.GetCommitTs() { - lastDMLEvent.Rows.Append(row.Rows, 0, row.Rows.NumRows()) - lastDMLEvent.RowTypes = append(lastDMLEvent.RowTypes, row.RowTypes...) - lastDMLEvent.Length += row.Length - lastDMLEvent.PostTxnFlushed = append(lastDMLEvent.PostTxnFlushed, row.PostTxnFlushed...) - return events - } - - log.Panic("append event with smaller commit ts", - zap.Int64("tableID", row.GetTableID()), - zap.Uint64("lastCommitTs", lastDMLEvent.GetCommitTs()), zap.Uint64("commitTs", row.GetCommitTs())) - return events -} diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index a446f308f0..352966344e 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -271,17 +271,6 @@ func TestEventsGroupRestoresRowsFromSharedChunk(t *testing.T) { second := messages[1].ToDMLEvent() require.True(t, second.Rows.GetRow(0).IsNull(1)) require.Equal(t, int64(2), second.Rows.GetRow(1).GetInt64(1)) - - var events []*commonEvent.DMLEvent - for _, message := range messages { - events = AppendOrMergeDMLEvent(events, message.ToDMLEvent()) - } - require.Len(t, events, 1) - require.Len(t, events[0].RowTypes, events[0].Rows.NumRows()) - require.True(t, events[0].Rows.GetRow(0).IsNull(1)) - require.Equal(t, int64(1), events[0].Rows.GetRow(1).GetInt64(1)) - require.True(t, events[0].Rows.GetRow(2).IsNull(1)) - require.Equal(t, int64(2), events[0].Rows.GetRow(3).GetInt64(1)) } func TestEventsGroupSpillDoesNotSignalDownstreamCallbacks(t *testing.T) { @@ -354,34 +343,3 @@ func BenchmarkEventsGroupResolveInto(b *testing.B) { }) } } - -func TestAppendOrMergeDMLEventMergesSameCommitTs(t *testing.T) { - var flushed []int - e1 := newTestDMLEvent(10, common.RowTypeInsert) - e1.AddPostFlushFunc(func() { flushed = append(flushed, 1) }) - e2 := newTestDMLEvent(10, common.RowTypeDelete) - e2.AddPostFlushFunc(func() { flushed = append(flushed, 2) }) - - events := AppendOrMergeDMLEvent(nil, e1) - events = AppendOrMergeDMLEvent(events, e2) - - require.Len(t, events, 1) - require.Same(t, e1, events[0]) - require.Equal(t, int32(2), events[0].Length) - require.Equal(t, []common.RowType{common.RowTypeInsert, common.RowTypeDelete}, events[0].RowTypes) - - events[0].PostFlush() - require.Equal(t, []int{1, 2}, flushed) -} - -func TestAppendOrMergeDMLEventAppendsDifferentCommitTs(t *testing.T) { - e1 := newTestDMLEvent(10, common.RowTypeInsert) - e2 := newTestDMLEvent(20, common.RowTypeDelete) - - events := AppendOrMergeDMLEvent(nil, e1) - events = AppendOrMergeDMLEvent(events, e2) - - require.Len(t, events, 2) - require.Same(t, e1, events[0]) - require.Same(t, e2, events[1]) -} From f4ede3de3f7442294f77c0a4ec8d54a8eb83287c Mon Sep 17 00:00:00 2001 From: wk989898 Date: Mon, 24 Aug 2026 10:33:22 +0000 Subject: [PATCH 08/11] fix Signed-off-by: wk989898 --- cmd/util/event_group.go | 12 +++++++++++- cmd/util/event_group_test.go | 29 +++++++++++------------------ 2 files changed, 22 insertions(+), 19 deletions(-) diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index 8b3af9606d..6516604d54 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -306,7 +306,17 @@ func marshalDMLRows(row *commonEvent.DMLEvent, tableInfoStored bool) (data []byt begin := row.PreviousTotalOffset end := row.Rows.NumRows() if len(row.RowTypes) != 0 { - end = begin + len(row.RowTypes) + end = begin + for _, rowType := range row.RowTypes { + switch rowType { + case commonType.RowTypeInsert, commonType.RowTypeDelete: + end++ + case commonType.RowTypeUpdate: + end += 2 + default: + return nil, errors.ErrSpillFileOp.FastGenByArgs("DML event has invalid row type") + } + } } if begin < 0 || end < begin || end > row.Rows.NumRows() { return nil, errors.ErrSpillFileOp.FastGenByArgs("DML event rows are outside the shared chunk") diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index 352966344e..c70572540a 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -231,30 +231,19 @@ func TestEventsGroupRestoresRowsFromSharedChunk(t *testing.T) { Name: ast.NewCIStr("t"), Columns: []*model.ColumnInfo{ {ID: 1, Name: ast.NewCIStr("id"), FieldType: *types.NewFieldType(mysql.TypeLonglong)}, - {ID: 2, Name: ast.NewCIStr("v1"), FieldType: *types.NewFieldType(mysql.TypeLong)}, }, }) rows := chunk.NewChunkWithCapacity(tableInfo.GetFieldSlice(), 4) - for _, value := range []struct { - null bool - value int64 - }{ - {null: true}, {value: 1}, {null: true}, {value: 2}, - } { - rows.AppendInt64(0, 42) - if value.null { - rows.AppendNull(1) - } else { - rows.AppendInt64(1, value.value) - } + for i := int64(0); i < 4; i++ { + rows.AppendInt64(0, i) } group := NewEventsGroup(0, 1) for _, offset := range []int{0, 2} { event := commonEvent.NewDMLEvent(common.NewDispatcherID(), 1, 90, 100, tableInfo) event.Rows = rows - event.RowTypes = []common.RowType{common.RowTypeInsert, common.RowTypeInsert} - event.Length = 2 + event.RowTypes = []common.RowType{common.RowTypeUpdate} + event.Length = 1 event.PreviousTotalOffset = offset require.NoError(t, group.AppendMessage(codeccommon.NewDMLMessageFromEvent(event))) } @@ -265,12 +254,16 @@ func TestEventsGroupRestoresRowsFromSharedChunk(t *testing.T) { for _, message := range messages { restored := message.ToDMLEvent() require.Zero(t, restored.PreviousTotalOffset) - require.Len(t, restored.RowTypes, restored.Rows.NumRows()) + require.Equal(t, 2, restored.Rows.NumRows()) } second := messages[1].ToDMLEvent() - require.True(t, second.Rows.GetRow(0).IsNull(1)) - require.Equal(t, int64(2), second.Rows.GetRow(1).GetInt64(1)) + row, ok := second.GetNextRow() + require.True(t, ok) + require.Equal(t, int64(2), row.PreRow.GetInt64(0)) + require.Equal(t, int64(3), row.Row.GetInt64(0)) + _, ok = second.GetNextRow() + require.False(t, ok) } func TestEventsGroupSpillDoesNotSignalDownstreamCallbacks(t *testing.T) { From 63557caaea4d4021bb13b4818d76a17bf5309900 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Mon, 24 Aug 2026 14:27:16 +0000 Subject: [PATCH 09/11] fix Signed-off-by: wk989898 --- cmd/util/event_group.go | 19 +++++++++++++++--- cmd/util/event_group_test.go | 39 +++++++++++++++++++++++++++++++++++- 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index 6516604d54..58a282c459 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -306,13 +306,26 @@ func marshalDMLRows(row *commonEvent.DMLEvent, tableInfoStored bool) (data []byt begin := row.PreviousTotalOffset end := row.Rows.NumRows() if len(row.RowTypes) != 0 { - end = begin + end = begin + len(row.RowTypes) + // Most decoders, including batched DML events, use one RowType entry per + // physical chunk row. An update consequently appears twice. The Avro + // decoder instead represents its single logical update with one entry, + // while retaining both rows in the chunk. Length distinguishes the two + // encodings: it is the number of logical row changes. + compactRowTypes := row.Length > 0 && len(row.RowTypes) == int(row.Length) + if compactRowTypes { + end = begin + } for _, rowType := range row.RowTypes { switch rowType { case commonType.RowTypeInsert, commonType.RowTypeDelete: - end++ + if compactRowTypes { + end++ + } case commonType.RowTypeUpdate: - end += 2 + if compactRowTypes { + end += 2 + } default: return nil, errors.ErrSpillFileOp.FastGenByArgs("DML event has invalid row type") } diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index c70572540a..eb04d35e59 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -242,7 +242,9 @@ func TestEventsGroupRestoresRowsFromSharedChunk(t *testing.T) { for _, offset := range []int{0, 2} { event := commonEvent.NewDMLEvent(common.NewDispatcherID(), 1, 90, 100, tableInfo) event.Rows = rows - event.RowTypes = []common.RowType{common.RowTypeUpdate} + // A decoded update occupies two RowTypes entries, matching its before + // and after rows in the shared chunk. + event.RowTypes = []common.RowType{common.RowTypeUpdate, common.RowTypeUpdate} event.Length = 1 event.PreviousTotalOffset = offset require.NoError(t, group.AppendMessage(codeccommon.NewDMLMessageFromEvent(event))) @@ -266,6 +268,41 @@ func TestEventsGroupRestoresRowsFromSharedChunk(t *testing.T) { require.False(t, ok) } +func TestEventsGroupRestoresCompactUpdateRows(t *testing.T) { + tableInfo := common.WrapTableInfo("test", &model.TableInfo{ + ID: 1, + Name: ast.NewCIStr("t"), + Columns: []*model.ColumnInfo{ + {ID: 1, Name: ast.NewCIStr("id"), FieldType: *types.NewFieldType(mysql.TypeLonglong)}, + }, + }) + rows := chunk.NewChunkWithCapacity(tableInfo.GetFieldSlice(), 2) + rows.AppendInt64(0, 1) + rows.AppendInt64(0, 2) + + // The Avro decoder represents an update with one RowType even though the + // chunk still contains both before and after rows. + event := commonEvent.NewDMLEvent(common.NewDispatcherID(), 1, 90, 100, tableInfo) + event.Rows = rows + event.RowTypes = []common.RowType{common.RowTypeUpdate} + event.Length = 1 + + group := NewEventsGroup(0, 1) + require.NoError(t, group.AppendMessage(codeccommon.NewDMLMessageFromEvent(event))) + messages, err := group.GetAllMessages() + require.NoError(t, err) + require.Len(t, messages, 1) + + restored := messages[0].ToDMLEvent() + require.Equal(t, 2, restored.Rows.NumRows()) + row, ok := restored.GetNextRow() + require.True(t, ok) + require.Equal(t, int64(1), row.PreRow.GetInt64(0)) + require.Equal(t, int64(2), row.Row.GetInt64(0)) + _, ok = restored.GetNextRow() + require.False(t, ok) +} + func TestEventsGroupSpillDoesNotSignalDownstreamCallbacks(t *testing.T) { event := newTestDMLEvent(100, common.RowTypeInsert) var enqueued, flushed int From 70880d05b2c8b71de462f5d5c63901120c7146a8 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Tue, 25 Aug 2026 08:14:14 +0000 Subject: [PATCH 10/11] fmt Signed-off-by: wk989898 --- cmd/storage-consumer/consumer.go | 5 +---- cmd/util/event_group_test.go | 2 +- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index 6d902407e0..41564babb2 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -309,10 +309,7 @@ func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAc zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), zap.Stringer("eventType", message.RowType)) - if err := group.AppendMessage(message); err != nil { - return err - } - return nil + return group.AppendMessage(message) } log.Warn("dml event commit ts fallback, ignore", zap.Uint64("commitTs", commitTs), diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index eb04d35e59..a162f3de91 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -234,7 +234,7 @@ func TestEventsGroupRestoresRowsFromSharedChunk(t *testing.T) { }, }) rows := chunk.NewChunkWithCapacity(tableInfo.GetFieldSlice(), 4) - for i := int64(0); i < 4; i++ { + for i := range int64(4) { rows.AppendInt64(0, i) } From 50da712bdc621df8ccf5624717bc354abe03d6e2 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Wed, 26 Aug 2026 09:03:42 +0000 Subject: [PATCH 11/11] fix Signed-off-by: wk989898 --- cmd/kafka-consumer/writer.go | 15 ++-- cmd/kafka-consumer/writer_test.go | 7 +- cmd/pulsar-consumer/writer.go | 15 ++-- cmd/pulsar-consumer/writer_test.go | 7 +- cmd/storage-consumer/consumer.go | 4 +- cmd/util/event_group.go | 61 ++++++++++++++++ cmd/util/event_group_test.go | 108 +++++++++++++++++++++++++++++ 7 files changed, 192 insertions(+), 25 deletions(-) diff --git a/cmd/kafka-consumer/writer.go b/cmd/kafka-consumer/writer.go index 5712eac585..6664ef5c7f 100644 --- a/cmd/kafka-consumer/writer.go +++ b/cmd/kafka-consumer/writer.go @@ -183,14 +183,11 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *event.DDLEvent) error { if err != nil { return err } - // A commit-ts is only a timestamp in a multi-source stream. It does - // not identify one transaction, so events from different upstreams can - // legitimately have the same value while carrying different schema - // snapshots. Keep restored events separate instead of merging their - // chunks by commit-ts. + events := make([]*event.DMLEvent, 0, len(messages)) for _, message := range messages { - resolvedEvents = append(resolvedEvents, message.ToDMLEvent()) + events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) } + resolvedEvents = append(resolvedEvents, events...) } } @@ -301,11 +298,11 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error { if err != nil { return err } - // See flushDDLEvent: events with an identical commit-ts can belong to - // different upstream transactions in a multi-source stream. + events := make([]*event.DMLEvent, 0, len(messages)) for _, message := range messages { - resolvedEvents = append(resolvedEvents, message.ToDMLEvent()) + events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) } + resolvedEvents = append(resolvedEvents, events...) } } total := len(resolvedEvents) diff --git a/cmd/kafka-consumer/writer_test.go b/cmd/kafka-consumer/writer_test.go index b7f0bb2ce7..c94fd3c4f2 100644 --- a/cmd/kafka-consumer/writer_test.go +++ b/cmd/kafka-consumer/writer_test.go @@ -273,10 +273,12 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { ctrl := gomock.NewController(t) s := sinkmock.NewMockSink(ctrl) flushedCommitTs := make([]uint64, 0) + flushedRowTypeCounts := make([]int, 0) s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) { flushedCommitTs = append(flushedCommitTs, event.GetCommitTs()) + flushedRowTypeCounts = append(flushedRowTypeCounts, len(event.RowTypes)) event.PostFlush() - }).Times(3) + }).Times(2) replicaCfg := config.GetDefaultReplicaConfig() eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false) @@ -302,7 +304,8 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { needCommit, err := w.Write(ctx, codeccommon.MessageTypeResolved) require.NoError(t, err) require.True(t, needCommit) - require.Equal(t, []uint64{10, 20, 20}, flushedCommitTs) + require.Equal(t, []uint64{10, 20}, flushedCommitTs) + require.Equal(t, []int{1, 2}, flushedRowTypeCounts) } func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { diff --git a/cmd/pulsar-consumer/writer.go b/cmd/pulsar-consumer/writer.go index aad1f06198..136d18719b 100644 --- a/cmd/pulsar-consumer/writer.go +++ b/cmd/pulsar-consumer/writer.go @@ -175,14 +175,11 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *commonEvent.DDLEvent) e if err != nil { return err } - // A commit-ts is only a timestamp in a multi-source stream. It does - // not identify one transaction, so events from different upstreams can - // legitimately have the same value while carrying different schema - // snapshots. Keep restored events separate instead of merging their - // chunks by commit-ts. + events := make([]*commonEvent.DMLEvent, 0, len(messages)) for _, message := range messages { - resolvedEvents = append(resolvedEvents, message.ToDMLEvent()) + events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) } + resolvedEvents = append(resolvedEvents, events...) } } @@ -293,11 +290,11 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error { if err != nil { return err } - // See flushDDLEvent: events with an identical commit-ts can belong to - // different upstream transactions in a multi-source stream. + events := make([]*commonEvent.DMLEvent, 0, len(messages)) for _, message := range messages { - resolvedEvents = append(resolvedEvents, message.ToDMLEvent()) + events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) } + resolvedEvents = append(resolvedEvents, events...) } } total := len(resolvedEvents) diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go index 645f308176..f44f5836b3 100644 --- a/cmd/pulsar-consumer/writer_test.go +++ b/cmd/pulsar-consumer/writer_test.go @@ -273,10 +273,12 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { ctrl := gomock.NewController(t) s := sinkmock.NewMockSink(ctrl) flushedCommitTs := make([]uint64, 0) + flushedRowTypeCounts := make([]int, 0) s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) { flushedCommitTs = append(flushedCommitTs, event.GetCommitTs()) + flushedRowTypeCounts = append(flushedRowTypeCounts, len(event.RowTypes)) event.PostFlush() - }).Times(3) + }).Times(2) p := &partitionProgress{ partition: 0, @@ -297,7 +299,8 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { needCommit, err := w.Write(ctx, codeccommon.MessageTypeResolved) require.NoError(t, err) require.True(t, needCommit) - require.Equal(t, []uint64{10, 20, 20}, flushedCommitTs) + require.Equal(t, []uint64{10, 20}, flushedCommitTs) + require.Equal(t, []int{1, 2}, flushedRowTypeCounts) } func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index 41564babb2..61eed078aa 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -409,11 +409,9 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error { if len(messages) == 0 { return nil } - // A commit-ts is only a timestamp in a multi-source stream. Keep restored - // events separate instead of merging chunks from independent transactions. events := make([]*event.DMLEvent, 0, len(messages)) for _, message := range messages { - events = append(events, message.ToDMLEvent()) + events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent()) } total := len(events) if total == 0 { diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index 58a282c459..e252488828 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -18,6 +18,7 @@ import ( "encoding/json" "math" "os" + "reflect" "sort" "github.com/pingcap/log" @@ -199,6 +200,66 @@ func (g *EventsGroup) Cleanup() error { return nil } +// AppendOrMergeDMLEvent appends row to events, or merges it into the preceding +// event when both are compatible parts of the same transaction. Events with the +// same commit-ts from different sources can use different table schemas, so a +// commit-ts alone is not enough to merge their chunks safely. +func AppendOrMergeDMLEvent(events []*commonEvent.DMLEvent, row *commonEvent.DMLEvent) []*commonEvent.DMLEvent { + if len(events) == 0 || !canMergeDMLEvents(events[len(events)-1], row) { + return append(events, row) + } + + last := events[len(events)-1] + lastRowTypeCount := len(last.RowTypes) + rowRowTypeCount := len(row.RowTypes) + last.Rows.Append(row.Rows, 0, row.Rows.NumRows()) + last.RowTypes = append(last.RowTypes, row.RowTypes...) + last.RowKeys = appendOptionalDMLValues(last.RowKeys, row.RowKeys, lastRowTypeCount, rowRowTypeCount) + last.Checksum = appendOptionalDMLValues(last.Checksum, row.Checksum, lastRowTypeCount, rowRowTypeCount) + last.Length += row.Length + last.ApproximateSize += row.ApproximateSize + last.PostTxnEnqueued = append(last.PostTxnEnqueued, row.PostTxnEnqueued...) + last.PostTxnFlushed = append(last.PostTxnFlushed, row.PostTxnFlushed...) + return events +} + +func canMergeDMLEvents(last, row *commonEvent.DMLEvent) bool { + if last == nil || row == nil || + last.CommitTs != row.CommitTs || + last.StartTs != row.StartTs || + last.DispatcherID != row.DispatcherID || + last.PhysicalTableID != row.PhysicalTableID || + last.TableInfoVersion != row.TableInfoVersion || + last.TableInfo == nil || row.TableInfo == nil || + last.TableInfo.GetSchemaName() != row.TableInfo.GetSchemaName() || + last.TableInfo.GetTableName() != row.TableInfo.GetTableName() || + last.TableInfo.GetUpdateTS() != row.TableInfo.GetUpdateTS() || + last.Rows == nil || row.Rows == nil || + last.Rows.NumCols() != row.Rows.NumCols() || + last.PreviousTotalOffset != 0 || row.PreviousTotalOffset != 0 || + !reflect.DeepEqual(last.TableInfo.GetFieldSlice(), row.TableInfo.GetFieldSlice()) { + return false + } + + return hasOptionalDMLValues(last.RowKeys, len(last.RowTypes)) && + hasOptionalDMLValues(row.RowKeys, len(row.RowTypes)) && + hasOptionalDMLValues(last.Checksum, len(last.RowTypes)) && + hasOptionalDMLValues(row.Checksum, len(row.RowTypes)) +} + +func hasOptionalDMLValues[T any](values []T, rowTypeCount int) bool { + return len(values) == 0 || len(values) == rowTypeCount +} + +func appendOptionalDMLValues[T any](last, row []T, lastRowTypeCount, rowRowTypeCount int) []T { + if len(last) == 0 && len(row) != 0 { + last = make([]T, lastRowTypeCount) + } else if len(last) != 0 && len(row) == 0 { + row = make([]T, rowRowTypeCount) + } + return append(last, row...) +} + func marshalDMLMessage(message *codeccommon.DMLMessage) (data []byte, row *commonEvent.DMLEvent, err error) { if message == nil { return nil, nil, errors.ErrSpillFileOp.FastGenByArgs("cannot spill nil DML message") diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index a162f3de91..e116c86c32 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -14,6 +14,7 @@ package util import ( + "fmt" "os" "testing" @@ -45,6 +46,113 @@ func newTestDMLEvent(commitTs uint64, rowTypes ...common.RowType) *commonEvent.D } } +func newMergeTestTableInfo(tableID int64, updateTS uint64, columnCount int) *common.TableInfo { + columns := make([]*model.ColumnInfo, columnCount) + for i := range columns { + columns[i] = &model.ColumnInfo{ + ID: int64(i + 1), + Offset: i, + Name: ast.NewCIStr(fmt.Sprintf("c%d", i)), + FieldType: *types.NewFieldType(mysql.TypeLonglong), + } + } + return common.WrapTableInfo("test", &model.TableInfo{ + ID: tableID, + Name: ast.NewCIStr("t"), + UpdateTS: updateTS, + Columns: columns, + }) +} + +func newMergeTestDMLEvent( + commitTs uint64, tableInfo *common.TableInfo, value int64, +) *commonEvent.DMLEvent { + rows := chunk.NewChunkWithCapacity(tableInfo.GetFieldSlice(), 1) + for column := range tableInfo.GetFieldSlice() { + rows.AppendInt64(column, value) + } + return &commonEvent.DMLEvent{ + DispatcherID: common.DispatcherID{Low: 1}, + PhysicalTableID: tableInfo.TableName.TableID, + StartTs: commitTs - 1, + CommitTs: commitTs, + Length: 1, + RowTypes: []common.RowType{common.RowTypeInsert}, + Rows: rows, + TableInfo: tableInfo, + TableInfoVersion: tableInfo.GetUpdateTS(), + } +} + +func TestAppendOrMergeDMLEvent(t *testing.T) { + t.Run("merge compatible events", func(t *testing.T) { + tableInfo := newMergeTestTableInfo(1, 10, 1) + first := newMergeTestDMLEvent(100, tableInfo, 1) + second := newMergeTestDMLEvent(100, tableInfo, 2) + first.RowKeys = [][]byte{[]byte("first")} + second.RowKeys = [][]byte{[]byte("second")} + first.Checksum = []*integrity.Checksum{{Current: 1}} + second.Checksum = []*integrity.Checksum{{Current: 2}} + var flushed []int + first.AddPostFlushFunc(func() { flushed = append(flushed, 1) }) + second.AddPostFlushFunc(func() { flushed = append(flushed, 2) }) + + events := AppendOrMergeDMLEvent(nil, first) + events = AppendOrMergeDMLEvent(events, second) + + require.Len(t, events, 1) + require.Same(t, first, events[0]) + require.Equal(t, int32(2), first.Length) + require.Equal(t, 2, first.Rows.NumRows()) + require.Equal(t, []byte("second"), first.RowKeys[1]) + require.Equal(t, uint32(2), first.Checksum[1].Current) + first.PostFlush() + require.Equal(t, []int{1, 2}, flushed) + }) + + t.Run("keep different schema layouts separate", func(t *testing.T) { + first := newMergeTestDMLEvent(100, newMergeTestTableInfo(1, 10, 1), 1) + second := newMergeTestDMLEvent(100, newMergeTestTableInfo(1, 11, 2), 2) + + events := AppendOrMergeDMLEvent(nil, first) + events = AppendOrMergeDMLEvent(events, second) + + require.Len(t, events, 2) + require.Same(t, first, events[0]) + require.Same(t, second, events[1]) + }) + + t.Run("keep different dispatchers separate", func(t *testing.T) { + tableInfo := newMergeTestTableInfo(1, 10, 1) + first := newMergeTestDMLEvent(100, tableInfo, 1) + second := newMergeTestDMLEvent(100, tableInfo, 2) + second.DispatcherID = common.DispatcherID{Low: 2} + + events := AppendOrMergeDMLEvent(nil, first) + events = AppendOrMergeDMLEvent(events, second) + + require.Len(t, events, 2) + }) + + t.Run("merge compatible events restored from spill", func(t *testing.T) { + tableInfo := newMergeTestTableInfo(1, 10, 1) + group := NewEventsGroup(0, 1) + require.NoError(t, group.AppendMessage(codeccommon.NewDMLMessageFromEvent( + newMergeTestDMLEvent(100, tableInfo, 1)))) + require.NoError(t, group.AppendMessage(codeccommon.NewDMLMessageFromEvent( + newMergeTestDMLEvent(100, tableInfo, 2)))) + + messages, err := group.GetAllMessages() + require.NoError(t, err) + require.Len(t, messages, 2) + events := AppendOrMergeDMLEvent(nil, messages[0].ToDMLEvent()) + events = AppendOrMergeDMLEvent(events, messages[1].ToDMLEvent()) + + require.Len(t, events, 1) + require.Equal(t, 2, events[0].Rows.NumRows()) + }) +} + func TestEventsGroupResolveIntoAppendsAndCleansResolvedSpillRecords(t *testing.T) { // Scenario: A consumer resolves events by watermark/commit-ts and appends them into a downstream // batch slice. Buffered messages are held only by a spill file, and the file must be cleaned once