diff --git a/cmd/kafka-consumer/writer.go b/cmd/kafka-consumer/writer.go index 5f9991e915..0ca062df60 100644 --- a/cmd/kafka-consumer/writer.go +++ b/cmd/kafka-consumer/writer.go @@ -609,22 +609,39 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti table = message.Table commitTs = message.GetCommitTs() ) + globalWatermark := w.globalWatermark() + if commitTs < globalWatermark { + log.Warn("DML event fallback row, since less than the global watermark, ignore it", + zap.Int64("tableID", tableID), zap.Int32("partition", progress.partition), + zap.Uint64("commitTs", commitTs), zap.Any("offset", offset), + zap.Uint64("globalWatermark", globalWatermark), + zap.Uint64("partitionWatermark", progress.watermark), + zap.Any("watermarkOffset", progress.watermarkOffset), + zap.String("schema", schema), zap.String("table", table), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) + return + } + group := progress.eventsGroup[tableID] if group == nil { group = util.NewEventsGroup(progress.partition, tableID) progress.eventsGroup[tableID] = group } + message = w.messageWithPartitionCheck(message, progress.partition, offset) + group.AppendMessage(message) if commitTs < progress.watermark { - log.Warn("DML Event fallback row, since less than the partition watermark, ignore it", + 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), zap.Uint64("commitTs", commitTs), zap.Any("offset", offset), zap.Uint64("watermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset), - zap.String("schema", schema), zap.String("table", table)) + zap.Uint64("globalWatermark", globalWatermark), + zap.String("schema", schema), zap.String("table", table), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) return } if commitTs >= group.HighWatermark { - message = w.messageWithPartitionCheck(message, progress.partition, offset) - group.AppendMessage(message, false) log.Debug("DML event append to the group", zap.Int32("partition", group.Partition), zap.Any("offset", offset), zap.Uint64("commitTs", commitTs), zap.Uint64("HighWatermark", group.HighWatermark), @@ -632,54 +649,13 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti zap.Stringer("eventType", message.RowType)) return } - if w.enableTableAcrossNodes { - log.Warn("DML events fallback, but enableTableAcrossNodes is true, still append it", - zap.Int32("partition", group.Partition), zap.Any("offset", offset), - 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(w.messageWithPartitionCheck(message, progress.partition, offset), true) - return - } - switch w.protocol { - case config.ProtocolSimple: - // simple protocol set the table id for all row message, it can be known which table the row message belongs to, - // also consider the table partition. - // open protocol set the partition table id if the table is partitioned. - // for normal table, the table id is generated by the fake table id generator by using schema and table name. - // so one event group for one normal table or one table partition, replayed messages can be ignored. - log.Warn("DML event fallback row, since less than the group high watermark, ignore it", - zap.Int32("partition", progress.partition), zap.Any("offset", offset), - zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), - zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType), - // zap.Any("columns", row.Columns), zap.Any("preColumns", row.PreColumns), - zap.Any("protocol", w.protocol)) - case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, - config.ProtocolDebezium, config.ProtocolDebeziumAvro: - // for partition table, these protocols cannot assign physical table id to each dml message, - // we cannot distinguish whether it's a real fallback event or not, still append it. - if w.partitionTableAccessor.IsPartitionTable(schema, table) { - log.Warn("DML events fallback, but the table is a partition table, still append it", - zap.Int32("partition", group.Partition), zap.Any("offset", offset), - 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), zap.Any("protocol", w.protocol)) - group.AppendMessage(w.messageWithPartitionCheck(message, progress.partition, offset), true) - return - } - log.Warn("DML event fallback row, since less than the group high watermark, ignore it", - zap.Int32("partition", progress.partition), zap.Any("offset", offset), - zap.Uint64("commitTs", commitTs), zap.Uint64("HighWatermark", group.HighWatermark), - zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType), - // zap.Any("columns", row.Columns), zap.Any("preColumns", row.PreColumns), - zap.Any("protocol", w.protocol)) - default: - log.Panic("unknown protocol", zap.Any("protocol", w.protocol)) - } + log.Warn("DML event commit ts fallback, append it and sort before flush", + zap.Int32("partition", progress.partition), zap.Any("offset", offset), + zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), + zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset), + 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)) } 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 7396b2ec32..b2fcb8009f 100644 --- a/cmd/kafka-consumer/writer_test.go +++ b/cmd/kafka-consumer/writer_test.go @@ -18,9 +18,11 @@ import ( "testing" "github.com/confluentinc/confluent-kafka-go/v2/kafka" + "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/cmd/util" "github.com/pingcap/ticdc/downstreamadapter/sink" "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" + "github.com/pingcap/ticdc/downstreamadapter/sink/mock" "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" @@ -284,6 +286,145 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) { require.Equal(t, "CREATE TABLE `common_1`.`a` (`a` BIGINT PRIMARY KEY,`b` INT)", w.ddlList[0].Query) } +func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + s := mock.NewMockSink(ctrl) + flushedCommitTs := make([]uint64, 0) + s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) { + flushedCommitTs = append(flushedCommitTs, event.GetCommitTs()) + event.PostFlush() + }).Times(2) + + replicaCfg := config.GetDefaultReplicaConfig() + eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false) + require.NoError(t, err) + + p := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 0, + } + w := &writer{ + progresses: []*partitionProgress{p}, + mysqlSink: s, + eventRouter: eventRouter, + protocol: config.ProtocolOpen, + } + + w.appendMessage2Group(newDMLMessageForWriterTest(20), p, kafka.Offset(1)) + w.appendMessage2Group(newDMLMessageForWriterTest(10), p, kafka.Offset(2)) + w.appendMessage2Group(newDMLMessageForWriterTest(20), p, kafka.Offset(3)) + + p.watermark = 20 + require.True(t, w.Write(ctx, codecCommon.MessageTypeResolved)) + require.Equal(t, []uint64{10, 20}, flushedCommitTs) +} + +func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + s := mock.NewMockSink(ctrl) + s.EXPECT().AddDMLEvent(gomock.Any()).Times(0) + + progress := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 20, + decoder: &singleDMLDecoder{message: newDMLMessageForWriterTest(10)}, + } + w := &writer{ + progresses: []*partitionProgress{progress}, + mysqlSink: s, + protocol: config.ProtocolOpen, + maxBatchSize: 64, + maxMessageBytes: 1, + } + + needCommit := w.WriteMessage(ctx, &kafka.Message{ + TopicPartition: kafka.TopicPartition{Partition: 0, Offset: kafka.Offset(10)}, + }) + + require.False(t, needCommit) + require.Nil(t, progress.eventsGroup[1]) +} + +func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) { + replicaCfg := config.GetDefaultReplicaConfig() + eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false) + require.NoError(t, err) + + progress := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 20, + } + w := &writer{ + progresses: []*partitionProgress{ + progress, + {partition: 1, watermark: 5}, + }, + eventRouter: eventRouter, + protocol: config.ProtocolOpen, + } + + w.appendMessage2Group(newDMLMessageForWriterTest(10), progress, kafka.Offset(10)) + + require.NotNil(t, progress.eventsGroup[1]) + resolved := progress.eventsGroup[1].ResolveInto(20, nil) + require.Len(t, resolved, 1) + require.Equal(t, uint64(10), resolved[0].GetCommitTs()) +} + +func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) { + replicaCfg := config.GetDefaultReplicaConfig() + eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, true) + require.NoError(t, err) + + w := &writer{ + progresses: []*partitionProgress{{partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}}, + eventRouter: eventRouter, + protocol: config.ProtocolAvro, + partitionTableAccessor: codecCommon.NewPartitionTableAccessor(), + } + + ddl := &commonEvent.DDLEvent{ + Query: "CREATE TABLE `target`.`dst` LIKE `target`.`src`", + SchemaName: "source", + TableName: "dst", + Type: byte(timodel.ActionCreateTable), + TableInfo: &common.TableInfo{ + TableName: common.TableName{ + Schema: "source", + Table: "dst", + IsPartition: true, + }, + }, + } + w.onDDL(ddl) + require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst")) + + newDMLEvent := func(commitTs uint64) *commonEvent.DMLEvent { + return &commonEvent.DMLEvent{ + PhysicalTableID: 1, + CommitTs: commitTs, + RowTypes: []common.RowType{common.RowTypeUpdate}, + Rows: chunk.NewChunkWithCapacity(nil, 0), + TableInfo: &common.TableInfo{ + TableName: common.TableName{Schema: "target", Table: "dst"}, + }, + } + } + + progress := w.progresses[0] + 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) + require.Len(t, resolved, 1) + require.Equal(t, uint64(100), resolved[0].GetCommitTs()) +} + func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) { for _, protocol := range []config.Protocol{ config.ProtocolDebezium, @@ -333,3 +474,42 @@ func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) { }) } } + +func newDMLMessageForWriterTest(commitTs uint64) *codecCommon.DMLMessage { + return codecCommon.NewDMLMessage(1, "test", "t", commitTs, common.RowTypeUpdate, func() *commonEvent.DMLEvent { + return &commonEvent.DMLEvent{ + PhysicalTableID: 1, + CommitTs: commitTs, + RowTypes: []common.RowType{common.RowTypeUpdate}, + Rows: chunk.NewChunkWithCapacity(nil, 0), + TableInfo: &common.TableInfo{ + TableName: common.TableName{Schema: "test", Table: "t", TableID: 1}, + }, + } + }) +} + +type singleDMLDecoder struct { + message *codecCommon.DMLMessage + consumed bool +} + +func (d *singleDMLDecoder) AddKeyValue(_, _ []byte) { +} + +func (d *singleDMLDecoder) HasNext() (codecCommon.MessageType, bool) { + return codecCommon.MessageTypeRow, !d.consumed +} + +func (d *singleDMLDecoder) NextResolvedEvent() uint64 { + return 0 +} + +func (d *singleDMLDecoder) NextDMLMessage() *codecCommon.DMLMessage { + d.consumed = true + return d.message +} + +func (d *singleDMLDecoder) NextDDLEvent() *commonEvent.DDLEvent { + return nil +} diff --git a/cmd/pulsar-consumer/writer.go b/cmd/pulsar-consumer/writer.go index 418fc5fc35..1802e27f90 100644 --- a/cmd/pulsar-consumer/writer.go +++ b/cmd/pulsar-consumer/writer.go @@ -498,55 +498,47 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti table = message.Table commitTs = message.GetCommitTs() ) + globalWatermark := w.globalWatermark() + if commitTs < globalWatermark { + log.Warn("DML event fallback row, since less than the global watermark, ignore it", + zap.Int64("tableID", tableID), zap.Int32("partition", progress.partition), + zap.Uint64("commitTs", commitTs), + zap.Uint64("globalWatermark", globalWatermark), + zap.Uint64("partitionWatermark", progress.watermark), + zap.String("schema", schema), zap.String("table", table), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) + return + } + group := progress.eventsGroup[tableID] if group == nil { group = util.NewEventsGroup(progress.partition, tableID) progress.eventsGroup[tableID] = group } + group.AppendMessage(message) if commitTs < progress.watermark { - log.Warn("DML Event fallback row, since less than the partition watermark, ignore it", + 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), zap.Uint64("commitTs", commitTs), zap.Uint64("watermark", progress.watermark), - zap.String("schema", schema), zap.String("table", table)) + zap.Uint64("globalWatermark", globalWatermark), + zap.String("schema", schema), zap.String("table", table), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) return } if commitTs >= group.HighWatermark { - group.AppendMessage(message, false) 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 } - if w.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, true) - return - } - switch w.protocol { - case config.ProtocolCanalJSON: - // for partition table, the canal-json message cannot assign physical table id to each dml message, - // we cannot distinguish whether it's a real fallback event or not, still append it. - isPartitionTable := w.partitionTableAccessor != nil && - w.partitionTableAccessor.IsPartitionTable(schema, table) - if isPartitionTable { - log.Warn("DML events fallback, but it's canal-json and partition table, 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, true) - return - } - log.Warn("DML event fallback row, since less than the group high watermark, ignore it", - zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), - zap.Any("partitionWatermark", progress.watermark), zap.Any("watermark", progress.watermark), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType), - zap.Any("protocol", w.protocol), zap.Bool("IsPartition", isPartitionTable)) - default: - log.Panic("unknown protocol", zap.Any("protocol", w.protocol)) - } + log.Warn("DML event commit ts fallback, append it and sort before flush", + zap.Int32("partition", progress.partition), + zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), + zap.Any("partitionWatermark", progress.watermark), + 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)) } diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go index f071376b97..93f2678d26 100644 --- a/cmd/pulsar-consumer/writer_test.go +++ b/cmd/pulsar-consumer/writer_test.go @@ -28,6 +28,7 @@ import ( "github.com/pingcap/ticdc/pkg/config" codeccommon "github.com/pingcap/ticdc/pkg/sink/codec/common" timodel "github.com/pingcap/tidb/pkg/meta/model" + "github.com/pingcap/tidb/pkg/util/chunk" "github.com/stretchr/testify/require" ) @@ -285,6 +286,130 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) { require.Equal(t, "CREATE TABLE `common_1`.`a` (`a` BIGINT PRIMARY KEY,`b` INT)", w.ddlList[0].Query) } +func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + s := sinkmock.NewMockSink(ctrl) + flushedCommitTs := make([]uint64, 0) + s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) { + flushedCommitTs = append(flushedCommitTs, event.GetCommitTs()) + event.PostFlush() + }).Times(2) + + p := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 0, + } + w := &writer{ + progresses: []*partitionProgress{p}, + mysqlSink: s, + protocol: config.ProtocolCanalJSON, + } + + w.appendMessage2Group(newDMLMessageForWriterTest(20), p) + w.appendMessage2Group(newDMLMessageForWriterTest(10), p) + w.appendMessage2Group(newDMLMessageForWriterTest(20), p) + + p.watermark = 20 + require.True(t, w.Write(ctx, codeccommon.MessageTypeResolved)) + require.Equal(t, []uint64{10, 20}, flushedCommitTs) +} + +func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + s := sinkmock.NewMockSink(ctrl) + s.EXPECT().AddDMLEvent(gomock.Any()).Times(0) + + decoder := &deferredDMLDecoder{ + row: &commonEvent.DMLEvent{ + PhysicalTableID: 1, + CommitTs: 10, + RowTypes: []common.RowType{common.RowTypeInsert}, + TableInfo: &common.TableInfo{ + TableName: common.TableName{Schema: "test", Table: "t", TableID: 1}, + }, + }, + } + progress := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 20, + decoder: decoder, + } + w := &writer{ + progresses: []*partitionProgress{progress}, + mysqlSink: s, + protocol: config.ProtocolCanalJSON, + } + + needCommit := w.WriteMessage(ctx, fakePulsarMessage{key: "k", payload: []byte(`{"fake":"row"}`)}) + + require.False(t, needCommit) + require.Nil(t, progress.eventsGroup[1]) +} + +func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) { + progress := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 20, + } + w := &writer{ + progresses: []*partitionProgress{ + progress, + {partition: 1, watermark: 5}, + }, + protocol: config.ProtocolCanalJSON, + } + + w.appendMessage2Group(newDMLMessageForWriterTest(10), progress) + + require.NotNil(t, progress.eventsGroup[1]) + resolved := progress.eventsGroup[1].ResolveInto(20, nil) + require.Len(t, resolved, 1) + require.Equal(t, uint64(10), resolved[0].GetCommitTs()) +} + +func TestOnDDLMarksRoutedCreateTableLikePartitionTable(t *testing.T) { + w := &writer{ + progresses: []*partitionProgress{ + {partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}, + }, + protocol: config.ProtocolCanalJSON, + partitionTableAccessor: codeccommon.NewPartitionTableAccessor(), + } + + ddl := &commonEvent.DDLEvent{ + Query: "CREATE TABLE `target`.`dst` LIKE `target`.`src`", + SchemaName: "source", + TableName: "dst", + Type: byte(timodel.ActionCreateTable), + TableInfo: &common.TableInfo{ + TableName: common.TableName{ + Schema: "source", + Table: "dst", + IsPartition: true, + }, + }, + } + 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) + + 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) { ctx := context.Background() ctrl := gomock.NewController(t) @@ -358,7 +483,7 @@ func (d *deferredDMLDecoder) NextResolvedEvent() uint64 { func (d *deferredDMLDecoder) NextDMLMessage() *codeccommon.DMLMessage { d.nextDMLMessageCount++ - return codeccommon.NewDMLMessage(1, "test", "t", 100, common.RowTypeInsert, func() *commonEvent.DMLEvent { + return codeccommon.NewDMLMessage(1, "test", "t", d.row.CommitTs, common.RowTypeInsert, func() *commonEvent.DMLEvent { d.toDMLEventCount++ return d.row }) @@ -368,6 +493,20 @@ func (d *deferredDMLDecoder) NextDDLEvent() *commonEvent.DDLEvent { return nil } +func newDMLMessageForWriterTest(commitTs uint64) *codeccommon.DMLMessage { + return codeccommon.NewDMLMessage(1, "test", "t", commitTs, common.RowTypeUpdate, func() *commonEvent.DMLEvent { + return &commonEvent.DMLEvent{ + PhysicalTableID: 1, + CommitTs: commitTs, + RowTypes: []common.RowType{common.RowTypeUpdate}, + Rows: chunk.NewChunkWithCapacity(nil, 0), + TableInfo: &common.TableInfo{ + TableName: common.TableName{Schema: "test", Table: "t", TableID: 1}, + }, + } + }) +} + type fakePulsarMessage struct { key string payload []byte diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index 7e3ec687d0..42081fbc9d 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -264,7 +264,7 @@ func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAc c.eventsGroup[tableID] = group } if commitTs >= group.HighWatermark { - group.AppendMessage(message, false) + group.AppendMessage(message) 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), @@ -276,7 +276,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)) - group.AppendMessage(message, true) + group.AppendMessage(message) return } log.Warn("dml event commit ts fallback, ignore", diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index b6eebe2039..2056212408 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -14,6 +14,7 @@ package util import ( + "math" "sort" "github.com/pingcap/log" @@ -48,65 +49,79 @@ func NewEventsGroup(partition int32, tableID int64) *EventsGroup { } // AppendMessage appends a message to event groups. -func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage, force bool) { +func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage) { commitTs := message.GetCommitTs() if commitTs > g.HighWatermark { g.HighWatermark = commitTs } + g.messages = append(g.messages, message) +} - var lastMessage *codeccommon.DMLMessage - if len(g.messages) > 0 { - lastMessage = g.messages[len(g.messages)-1] +// 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. +func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) []*codeccommon.DMLMessage { + if len(g.messages) == 0 { + return dst } - if lastMessage == nil || lastMessage.GetCommitTs() <= commitTs { - g.messages = append(g.messages, message) - return + original := g.messages + remaining := g.messages[:0] + resolved := make([]*codeccommon.DMLMessage, 0, len(g.messages)) + + var ( + lastCommitTs uint64 + outOfOrder bool + outOfOrderLastTs uint64 + outOfOrderCommitTs uint64 + ) + for _, message := range g.messages { + commitTs := message.GetCommitTs() + if commitTs > resolve { + remaining = append(remaining, message) + continue + } + if len(resolved) > 0 && commitTs < lastCommitTs && !outOfOrder { + outOfOrder = true + outOfOrderLastTs = lastCommitTs + outOfOrderCommitTs = commitTs + } + lastCommitTs = commitTs + resolved = append(resolved, message) } - - if force { - i := sort.Search(len(g.messages), func(i int) bool { - return g.messages[i].GetCommitTs() > commitTs - }) - g.messages = append(g.messages, nil) - copy(g.messages[i+1:], g.messages[i:]) - g.messages[i] = message - return + if len(resolved) == 0 { + return dst } - log.Panic("append event with smaller commit ts", - zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), - zap.Uint64("lastCommitTs", lastMessage.GetCommitTs()), zap.Uint64("commitTs", commitTs)) -} -// ResolveInto appends all messages with CommitTs <= resolve into dst and removes them from the group. -// ResolveInto copies pointers into dst first, then clears the resolved prefix so Go GC can reclaim -// resolved messages once downstream is done with them. -func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) []*codeccommon.DMLMessage { - i := sort.Search(len(g.messages), func(i int) bool { - return g.messages[i].GetCommitTs() > resolve - }) - if i == 0 { - return dst + if outOfOrder { + log.Warn("DML events are out of order before flush, sort them", + zap.Int32("partition", g.Partition), + zap.Int64("tableID", g.tableID), + zap.Uint64("resolveTs", resolve), + zap.Int("resolved", len(resolved)), + zap.Uint64("lastCommitTs", outOfOrderLastTs), + zap.Uint64("commitTs", outOfOrderCommitTs)) + sort.SliceStable(resolved, func(i, j int) bool { + return resolved[i].GetCommitTs() < resolved[j].GetCommitTs() + }) } - // Copy pointers out first so we can safely clear the group's slice without affecting callers. - dst = append(dst, g.messages[:i]...) - clear(g.messages[:i]) - g.messages = g.messages[i:] + dst = append(dst, resolved...) + clear(original[len(remaining):]) + g.messages = remaining if len(g.messages) != 0 { + firstCommitTs := g.messages[0].GetCommitTs() log.Debug("not all events resolved", zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), - zap.Int("resolved", i), zap.Int("remained", len(g.messages)), - zap.Uint64("resolveTs", resolve), zap.Uint64("firstCommitTs", g.messages[0].GetCommitTs())) + zap.Int("resolved", len(resolved)), zap.Int("remained", len(g.messages)), + zap.Uint64("resolveTs", resolve), zap.Uint64("firstCommitTs", firstCommitTs)) } return dst } // GetAllMessages gets all messages. func (g *EventsGroup) GetAllMessages() []*codeccommon.DMLMessage { - result := g.messages - g.messages = nil - return result + return g.ResolveInto(math.MaxUint64, nil) } // AppendOrMergeDMLEvent appends a DML event, or merges it into the previous event diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index 0bb4f82fe9..da2cbec9c4 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -37,25 +37,26 @@ func newTestDMLEvent(commitTs uint64, rowTypes ...common.RowType) *commonEvent.D } } -func TestEventsGroupResolveIntoAppendsAndClearsResolvedPrefix(t *testing.T) { - // Scenario: A consumer resolves a prefix of events by watermark/commit-ts and appends them - // array to avoid retaining already-flushed events and causing unbounded memory growth. +func TestEventsGroupResolveIntoAppendsAndClearsResolvedMessages(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. // // 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) the resolved prefix in the original backing slice is cleared (nil'd). + // (c) resolved messages in the original backing slice are cleared (nil'd). group := NewEventsGroup(0, 1) m1 := newTestDMLMessage(1) m2 := newTestDMLMessage(2) m3 := newTestDMLMessage(3) - group.AppendMessage(m1, false) - group.AppendMessage(m2, false) - group.AppendMessage(m3, false) + group.AppendMessage(m1) + group.AppendMessage(m2) + group.AppendMessage(m3) // Keep a reference to the original slice header so we can validate that ResolveInto clears - // the resolved prefix in-place (this is what prevents GC retention of flushed events). + // resolved messages in-place (this is what prevents GC retention of flushed events). original := group.messages var dst []*codeccommon.DMLMessage @@ -68,11 +69,11 @@ func TestEventsGroupResolveIntoAppendsAndClearsResolvedPrefix(t *testing.T) { require.Len(t, group.messages, 1) require.Same(t, m3, group.messages[0]) - // The resolved prefix must be nil so the group doesn't keep flushed events alive via its - // backing array (classic Go slice memory retention pitfall). - require.Nil(t, original[0]) + // 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.Same(t, m3, original[2]) + require.Nil(t, original[2]) } func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) { @@ -81,8 +82,8 @@ func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) { group := NewEventsGroup(0, 1) m1 := newTestDMLMessage(10) m2 := newTestDMLMessage(20) - group.AppendMessage(m1, false) - group.AppendMessage(m2, false) + group.AppendMessage(m1) + group.AppendMessage(m2) original := group.messages dst := make([]*codeccommon.DMLMessage, 0, 1) @@ -104,8 +105,8 @@ func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) { group := NewEventsGroup(0, 1) m1 := newTestDMLMessage(1) m2 := newTestDMLMessage(2) - group.AppendMessage(m1, false) - group.AppendMessage(m2, false) + group.AppendMessage(m1) + group.AppendMessage(m2) original := group.messages var dst []*codeccommon.DMLMessage @@ -120,6 +121,67 @@ func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) { require.Nil(t, original[1]) } +func TestEventsGroupResolveIntoSortsOutOfOrderResolvedMessages(t *testing.T) { + group := NewEventsGroup(0, 1) + m1 := newTestDMLMessage(20) + m2 := newTestDMLMessage(10) + m3 := newTestDMLMessage(30) + group.AppendMessage(m1) + 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.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]) +} + +func TestEventsGroupResolveIntoKeepsSameCommitTsStable(t *testing.T) { + group := NewEventsGroup(0, 1) + m1 := newTestDMLMessage(20) + m2 := newTestDMLMessage(10) + m3 := newTestDMLMessage(20) + group.AppendMessage(m1) + group.AppendMessage(m2) + group.AppendMessage(m3) + + var dst []*codeccommon.DMLMessage + 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.Empty(t, group.messages) +} + +func TestEventsGroupGetAllMessagesSortsOutOfOrderMessages(t *testing.T) { + group := NewEventsGroup(0, 1) + m1 := newTestDMLMessage(20) + m2 := newTestDMLMessage(10) + m3 := newTestDMLMessage(30) + group.AppendMessage(m1) + group.AppendMessage(m2) + group.AppendMessage(m3) + + 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.Empty(t, group.messages) +} + func TestAppendOrMergeDMLEventMergesSameCommitTs(t *testing.T) { var flushed []int e1 := newTestDMLEvent(10, common.RowTypeInsert)