diff --git a/cmd/kafka-consumer/consumer.go b/cmd/kafka-consumer/consumer.go
index 2bf26f1aee..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,7 +172,13 @@ 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 {
+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 {
 		return c.writer.run(ctx)
diff --git a/cmd/kafka-consumer/writer.go b/cmd/kafka-consumer/writer.go
index bbf8388405..6664ef5c7f 100644
--- a/cmd/kafka-consumer/writer.go
+++ b/cmd/kafka-consumer/writer.go
@@ -147,6 +147,22 @@ func (w *writer) run(ctx context.Context) error {
 	return w.mysqlSink.Run(ctx)
 }
 
+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 {
 	var (
 		done = make(chan struct{}, 1)
@@ -163,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())
@@ -275,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())
@@ -320,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
@@ -355,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",
@@ -389,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()
@@ -405,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.
@@ -427,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
@@ -479,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
 		}
 	}
 
@@ -488,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
 		}
 	}
 
@@ -498,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) {
@@ -600,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 (
@@ -620,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]
@@ -628,8 +655,11 @@ 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)
+	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),
@@ -639,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",
@@ -647,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),
@@ -656,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..c94fd3c4f2 100644
--- a/cmd/kafka-consumer/writer_test.go
+++ b/cmd/kafka-consumer/writer_test.go
@@ -273,8 +273,10 @@ 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(2)
 
@@ -299,8 +301,11 @@ 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)
+	require.Equal(t, []int{1, 2}, flushedRowTypeCounts)
 }
 
 func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) {
@@ -323,9 +328,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 +359,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 +411,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 +460,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 8c75a8843b..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,7 +126,13 @@ 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 {
+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 {
 		return c.writer.run(ctx)
diff --git a/cmd/pulsar-consumer/writer.go b/cmd/pulsar-consumer/writer.go
index e0fb914e1e..136d18719b 100644
--- a/cmd/pulsar-consumer/writer.go
+++ b/cmd/pulsar-consumer/writer.go
@@ -139,6 +139,22 @@ func (w *writer) run(ctx context.Context) error {
 	return w.mysqlSink.Run(ctx)
 }
 
+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 {
 	var (
 		done = make(chan struct{}, 1)
@@ -155,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())
@@ -267,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())
@@ -310,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())
 
@@ -338,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",
@@ -351,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
@@ -409,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
 		}
 	}
 
@@ -418,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
 		}
 	}
 
@@ -428,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) {
@@ -491,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
@@ -508,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]
@@ -516,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),
@@ -525,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),
@@ -541,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 35c9c037e3..f44f5836b3 100644
--- a/cmd/pulsar-consumer/writer_test.go
+++ b/cmd/pulsar-consumer/writer_test.go
@@ -273,8 +273,10 @@ 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(2)
 
@@ -294,8 +296,11 @@ 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)
+	require.Equal(t, []int{1, 2}, flushedRowTypeCounts)
 }
 
 func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) {
@@ -326,7 +331,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 +355,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())
 }
@@ -381,20 +388,17 @@ 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)
+	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())
 }
 
-func TestWriteMessageDefersDMLAssemblyUntilFlush(t *testing.T) {
+func TestWriteMessageSpillsDMLImmediately(t *testing.T) {
 	ctx := context.Background()
 	ctrl := gomock.NewController(t)
 	s := sinkmock.NewMockSink(ctrl)
@@ -423,21 +427,28 @@ func TestWriteMessageDefersDMLAssemblyUntilFlush(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.Zero(t, decoder.toDMLEventCount)
-	require.Len(t, progress.eventsGroup[1].ResolveInto(99, nil), 0)
+	require.Equal(t, 1, decoder.toDMLEventCount)
+	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 d89de46cfe..61eed078aa 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,21 @@ 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
+		return group.AppendMessage(message)
 	}
 	log.Warn("dml event commit ts fallback, ignore",
 		zap.Uint64("commitTs", commitTs),
@@ -316,6 +317,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 +374,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 +402,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,6 +456,19 @@ func (c *consumer) flushDMLEvents(ctx context.Context, tableID int64) error {
 	}
 }
 
+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) {
 	if c.globalCheckpointTs > 0 && dmlkey.TableVersion > c.globalCheckpointTs {
 		log.Debug("skip dml index file by checkpoint",
@@ -766,7 +786,13 @@ func (c *consumer) handle(ctx context.Context) error {
 	}
 }
 
-func (c *consumer) run(ctx context.Context) error {
+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 {
 		return c.sink.Run(ctx)
diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go
index d4d324282a..e252488828 100644
--- a/cmd/util/event_group.go
+++ b/cmd/util/event_group.go
@@ -14,21 +14,39 @@
 package util
 
 import (
+	"encoding/binary"
+	"encoding/json"
 	"math"
+	"os"
+	"reflect"
 	"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,38 +56,80 @@ 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.
-func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage) {
+// 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) error {
+	return 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,
+) error {
+	return g.appendMessage(message, postRestore)
+}
+
+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")
+	}
 	commitTs := message.GetCommitTs()
-	if len(g.messages) > 0 && commitTs < g.messages[len(g.messages)-1].GetCommitTs() {
+
+	data, _, err := marshalDMLMessage(message)
+	if err != nil {
+		return err
+	}
+	if g.spillFile == nil {
+		g.spillFile, err = spill.NewRecordFile(os.TempDir(), eventsGroupSpillPattern)
+		if err != nil {
+			return err
+		}
+	}
+	handle, err := g.spillFile.Append(data)
+	if err != nil {
+		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, message)
+	g.messages = append(g.messages, spilledMessage{
+		commitTs:    commitTs,
+		handle:      handle,
+		postRestore: postRestore,
+	})
+	return nil
 }
 
 // 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 {
+// 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, error) {
 	if len(g.messages) == 0 {
-		return dst
+		return dst, nil
 	}
 
 	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",
@@ -80,51 +140,386 @@ 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")
 	}
 
-	dst = append(dst, g.messages[:resolvedCount]...)
+	for _, message := range g.messages[:resolvedCount] {
+		data, err := g.spillFile.Read(message.handle)
+		if err != nil {
+			return dst, err
+		}
+		restored, err := unmarshalDMLMessage(data)
+		if err != nil {
+			return dst, 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 {
+			return dst, 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)),
 			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)
 }
 
-// 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]
+// 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()
+	if err != nil {
+		return err
 	}
+	g.spillFile = nil
+	clear(g.messages)
+	g.messages = g.messages[:0]
+	return nil
+}
 
-	if lastDMLEvent == nil || lastDMLEvent.GetCommitTs() < row.GetCommitTs() {
+// 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)
 	}
 
-	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
+	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...)
+}
 
-	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
+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
+	}
+	// 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")
+	}
+
+	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
+			}
+			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
+		}
+	}
+
+	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
+			}
+			log.Warn("spill DML event without row data",
+				zap.Int64("tableID", row.PhysicalTableID), zap.Uint64("commitTs", row.CommitTs), zap.Error(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, 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)))
+	data = append(data, tableInfoData...)
+	data = appendUint64(data, uint64(len(rowsData)))
+	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 {
+		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()
+	}
+	begin := row.PreviousTotalOffset
+	end := row.Rows.NumRows()
+	if len(row.RowTypes) != 0 {
+		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:
+				if compactRowTypes {
+					end++
+				}
+			case commonType.RowTypeUpdate:
+				if compactRowTypes {
+					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")
+	}
+	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) {
+	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
+	}
+	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
+	}
+	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()
+		}
+		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")
+	}
+	return codeccommon.NewDMLMessage(row.PhysicalTableID, string(schemaData), string(tableData), row.CommitTs,
+		commonType.RowType(rowType), 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 {
+	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
 }
diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go
index a756f297bd..e116c86c32 100644
--- a/cmd/util/event_group_test.go
+++ b/cmd/util/event_group_test.go
@@ -14,19 +14,26 @@
 package util
 
 import (
+	"fmt"
+	"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 +46,123 @@ func newTestDMLEvent(commitTs uint64, rowTypes ...common.RowType) *commonEvent.D
 	}
 }
 
-func TestEventsGroupResolveIntoAppendsAndClearsResolvedMessages(t *testing.T) {
+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. 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 +171,25 @@ 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)
+	dst, err := group.ResolveInto(2, dst)
+	require.NoError(t, err)
 
 	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])
-
-	// 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])
+	require.Equal(t, m3.GetCommitTs(), group.messages[0].commitTs)
+	require.FileExists(t, spillPath)
+
+	_, err = group.GetAllMessages()
+	require.NoError(t, err)
+	require.Nil(t, group.spillFile)
+	_, err = os.Stat(spillPath)
+	require.True(t, os.IsNotExist(err))
 }
 
 func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) {
@@ -87,18 +201,14 @@ 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)
+	dst, err := group.ResolveInto(5, dst)
+	require.NoError(t, err)
 
 	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 +220,19 @@ 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)
+	dst, err := group.ResolveInto(100, dst)
+	require.NoError(t, err)
 
 	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 +244,16 @@ func TestEventsGroupResolveIntoSortsOutOfOrderResolvedMessages(t *testing.T) {
 	group.AppendMessage(m2)
 	group.AppendMessage(m3)
 
-	original := group.messages
 	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.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) {
@@ -157,12 +266,13 @@ 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.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)
 }
 
@@ -175,15 +285,145 @@ 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.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, err := group.GetAllMessages()
+	require.NoError(t, err)
+	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 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)},
+		},
+	})
+	rows := chunk.NewChunkWithCapacity(tableInfo.GetFieldSlice(), 4)
+	for i := range int64(4) {
+		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
+		// 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)))
+	}
+
+	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.Equal(t, 2, restored.Rows.NumRows())
+	}
+
+	second := messages[1].ToDMLEvent()
+	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 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
+	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
 
@@ -215,50 +455,29 @@ 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()
+			b.StopTimer()
 			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 {
+					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])
 			}
 		})
 	}
 }
-
-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])
-}