From bfce54288ae8ef31192b318aec02fe7b7b51a1d8 Mon Sep 17 00:00:00 2001 From: wlwilliamx Date: Wed, 12 Aug 2026 15:38:53 +0800 Subject: [PATCH] event: account for raw rows in batch DML event size Remote batch DML events retain their row payload in RawRows until the event collector handles them. Include that payload in GetSize so dynamic stream memory and byte batching account for it. --- pkg/common/event/dml_event.go | 5 +++++ pkg/common/event/dml_event_test.go | 8 ++++++++ 2 files changed, 13 insertions(+) diff --git a/pkg/common/event/dml_event.go b/pkg/common/event/dml_event.go index bb0f19dcf3..a9b3443a63 100644 --- a/pkg/common/event/dml_event.go +++ b/pkg/common/event/dml_event.go @@ -344,7 +344,12 @@ func (b *BatchDMLEvent) GetStartTs() common.Ts { return b.DMLEvents[0].GetStartTs() } +// GetSize returns the in-memory size of the row payload. +// Remote events keep their encoded rows in RawRows until AssembleRows decodes them. func (b *BatchDMLEvent) GetSize() int64 { + if b.Rows == nil { + return int64(len(b.RawRows)) + } return b.Rows.MemoryUsage() } diff --git a/pkg/common/event/dml_event_test.go b/pkg/common/event/dml_event_test.go index 7a04d9df42..8f4e499fed 100644 --- a/pkg/common/event/dml_event_test.go +++ b/pkg/common/event/dml_event_test.go @@ -216,6 +216,8 @@ func TestBatchDMLEventAssembleRowsKeepsOriginalTableInfoForLocalRowsWithoutRouti require.Equal(t, "t", batchDMLEvent.TableInfo.GetTargetTableName()) } +// TestBatchDMLEventAssembleRowsDecodesRemoteRawRows verifies size accounting +// before and after a remotely received batch decodes its raw row payload. func TestBatchDMLEventAssembleRowsDecodesRemoteRawRows(t *testing.T) { helper := NewEventTestHelper(t) defer helper.Close() @@ -233,6 +235,9 @@ func TestBatchDMLEventAssembleRowsDecodesRemoteRawRows(t *testing.T) { Rows: dmlEvent.Rows, TableInfo: dmlEvent.TableInfo, } + require.Equal(t, batchDMLEvent.Rows.MemoryUsage(), batchDMLEvent.GetSize()) + require.Positive(t, batchDMLEvent.GetSize()) + data, err := batchDMLEvent.Marshal() require.NoError(t, err) @@ -241,10 +246,13 @@ func TestBatchDMLEventAssembleRowsDecodesRemoteRawRows(t *testing.T) { require.NoError(t, err) require.Nil(t, reverseEvents.Rows) require.NotEmpty(t, reverseEvents.RawRows) + require.Equal(t, int64(len(reverseEvents.RawRows)), reverseEvents.GetSize()) reverseEvents.AssembleRows(batchDMLEvent.TableInfo) require.Nil(t, reverseEvents.RawRows) + require.Equal(t, reverseEvents.Rows.MemoryUsage(), reverseEvents.GetSize()) + require.Positive(t, reverseEvents.GetSize()) require.Same(t, batchDMLEvent.TableInfo, reverseEvents.TableInfo) require.Equal(t, batchDMLEvent.Rows.ToString(batchDMLEvent.TableInfo.GetFieldSlice()), reverseEvents.Rows.ToString(batchDMLEvent.TableInfo.GetFieldSlice())) }