From 08d90bd949df182e8ea67a857f5c00b5d2d16e9e Mon Sep 17 00:00:00 2001 From: qupeng Date: Mon, 9 Sep 2024 14:38:32 +0800 Subject: [PATCH] init Signed-off-by: qupeng --- .../sourcemanager/sorter/pebble/event_sorter.go | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/cdc/processor/sourcemanager/sorter/pebble/event_sorter.go b/cdc/processor/sourcemanager/sorter/pebble/event_sorter.go index 443f1d345bf..12917dc1022 100644 --- a/cdc/processor/sourcemanager/sorter/pebble/event_sorter.go +++ b/cdc/processor/sourcemanager/sorter/pebble/event_sorter.go @@ -311,22 +311,20 @@ func (s *EventSorter) SlotsAndHasher() (slotCount int, hasher func(tablepb.Span, // fetched or not. func (s *EventIter) Next() (event *model.PolymorphicEvent, txnFinished sorter.Position, err error) { valid := s.iter != nil && s.iter.Valid() - var value []byte var nextEvent *model.PolymorphicEvent // We need to decide whether the current event is the last event in this transactions // If the current event is the last one, we need to set txnFinished // Thus, we need to fetch the next event and compare the commitTs and startTs with it for valid { - nextStart := time.Now() - value = s.iter.Value() - s.nextDuration.Observe(time.Since(nextStart).Seconds()) - nextEvent = &model.PolymorphicEvent{} - if _, err = s.serde.Unmarshal(nextEvent, value); err != nil { + if _, err = s.serde.Unmarshal(nextEvent, s.iter.Value()); err != nil { return } + + nextStart := time.Now() valid = s.iter.Next() + s.nextDuration.Observe(time.Since(nextStart).Seconds()) if s.currentEvent != nil { break