You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/04/22 05:29:22 UTC
(skywalking-banyandb) 01/02: feat(trace): introduce fair fast/slow lane scheduling for part merges
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch feat/merge
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 6c546f86ff64c8336829c9d3714e1f1c6fadc17b
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Tue Apr 21 04:21:02 2026 +0000
feat(trace): introduce fair fast/slow lane scheduling for part merges
Previously a single merge worker processed all file-part merges serially,
so a long-running large merge could stall short merges indefinitely. Split
the merge pipeline into a dispatcher goroutine plus two lane workers
(fast lane sized CPUs/2, slow lane sized 1) and classify each merge by
total compressed size against a memory-derived threshold (memLimit/16,
clamped to at least 32MB). In-flight tracking prevents the same part
being scheduled twice across lanes. Expose queue wait time as
total_merge_queue_latency for observability.
Fixes apache/skywalking#13776
---
CHANGES.md | 1 +
banyand/trace/merger.go | 252 +++++++++++++++++++++++++++++++++++--------
banyand/trace/merger_test.go | 249 ++++++++++++++++++++++++++++++++++++++++++
banyand/trace/metrics.go | 27 +++--
banyand/trace/tstable.go | 2 +
5 files changed, 483 insertions(+), 48 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 0dcd8a960..ff8ad5adc 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -12,6 +12,7 @@ Release Notes.
- Sync lifecycle e2e test from SkyWalking stages test.
- Add periodic health check for property schema connection.
- Persist segment end time in per-segment metadata so boundaries don't shift across restarts or config changes.
+- Introduce fair fast/slow lane scheduling for trace part merges to prevent short merges from being blocked by long-running merges; expose queue wait time as `total_merge_queue_latency`.
### Bug Fixes
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 875f98fb8..934a201b6 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -21,6 +21,7 @@ import (
"errors"
"fmt"
"path/filepath"
+ "sync"
"sync/atomic"
"time"
@@ -40,72 +41,234 @@ var (
mergeTypeFile = "file"
)
+var (
+ mergeLaneFast = "fast"
+ mergeLaneSlow = "slow"
+)
+
+const defaultSmallMergeThreshold = 32 << 20 // 32MB fallback
+
+func computeSmallMergeThreshold() uint64 {
+ memLimit, err := cgroups.MemoryLimit()
+ if err != nil || memLimit <= 0 {
+ return defaultSmallMergeThreshold
+ }
+ threshold := uint64(memLimit) / 16
+ if threshold < defaultSmallMergeThreshold {
+ return defaultSmallMergeThreshold
+ }
+ return threshold
+}
+
+type mergeDispatchRequest struct {
+ enqueuedAt time.Time
+ parts []*partWrapper
+ toBeMerged map[uint64]struct{}
+ typ string
+}
+
func (tst *tsTable) mergeLoop(merges chan *mergerIntroduction, flusherNotifier watcher.Channel) {
defer tst.loopCloser.Done()
- var epoch uint64
+ var lastProcessedEpoch uint64
ew := flusherNotifier.Add(0, tst.loopCloser.CloseNotify())
if ew == nil {
return
}
- var pwsChunk []*partWrapper
+ threshold := computeSmallMergeThreshold()
+ fastWorkers := max(1, cgroups.CPUs()/2)
+ fastCh := make(chan *mergeDispatchRequest, fastWorkers)
+ slowCh := make(chan *mergeDispatchRequest, 1)
+ triggerCh := make(chan struct{}, 1)
+
+ var workersWg, dispatcherWg sync.WaitGroup
+
+ for i := 0; i < fastWorkers; i++ {
+ workersWg.Add(1)
+ go func() {
+ defer workersWg.Done()
+ tst.mergeLaneWorker(fastCh, merges)
+ }()
+ }
+ workersWg.Add(1)
+ go func() {
+ defer workersWg.Done()
+ tst.mergeLaneWorker(slowCh, merges)
+ }()
+
+ dispatcherWg.Add(1)
+ go func() {
+ defer dispatcherWg.Done()
+ tst.dispatcherLoop(triggerCh, threshold, fastCh, slowCh)
+ }()
+
+ // Shutdown order: stop dispatcher first so no new work enters the lane
+ // channels, then close the lane channels so idle workers exit their range
+ // loops, then wait for workers to drain any in-flight merges.
+ defer func() {
+ close(triggerCh)
+ dispatcherWg.Wait()
+ close(fastCh)
+ close(slowCh)
+ workersWg.Wait()
+ }()
for {
select {
case <-tst.loopCloser.CloseNotify():
return
case <-ew.Watch():
- if func() bool {
- curSnapshot := tst.currentSnapshot()
- if curSnapshot == nil {
- return false
- }
- defer curSnapshot.decRef()
- if curSnapshot.epoch != epoch {
- select {
- case mergeMaxConcurrencyCh <- struct{}{}:
- defer func() {
- <-mergeMaxConcurrencyCh
- }()
- case <-tst.loopCloser.CloseNotify():
- return true
- }
- tst.incTotalMergeLoopStarted(1)
- defer tst.incTotalMergeLoopFinished(1)
- var err error
- if pwsChunk, err = tst.mergeSnapshot(curSnapshot, merges, pwsChunk[:0]); err != nil {
- if errors.Is(err, errClosed) {
- return true
- }
- tst.l.Logger.Warn().Err(err).Msgf("cannot merge snapshot: %d", curSnapshot.epoch)
- tst.incTotalMergeLoopErr(1)
- return false
- }
- epoch = curSnapshot.epoch
+ curSnapshot := tst.currentSnapshot()
+ if curSnapshot == nil {
+ return
+ }
+ if curSnapshot.epoch > lastProcessedEpoch {
+ tst.incTotalMergeLoopStarted(1)
+ select {
+ case triggerCh <- struct{}{}:
+ default:
}
- ew = flusherNotifier.Add(epoch, tst.loopCloser.CloseNotify())
- return ew == nil
- }() {
+ lastProcessedEpoch = curSnapshot.epoch
+ tst.incTotalMergeLoopFinished(1)
+ }
+ curSnapshot.decRef()
+ ew = flusherNotifier.Add(lastProcessedEpoch, tst.loopCloser.CloseNotify())
+ if ew == nil {
return
}
}
}
}
-func (tst *tsTable) mergeSnapshot(curSnapshot *snapshot, merges chan *mergerIntroduction, dst []*partWrapper) ([]*partWrapper, error) {
- freeDiskSize := tst.freeDiskSpace(tst.root)
- var toBeMerged map[uint64]struct{}
- dst, toBeMerged = tst.getPartsToMerge(curSnapshot, freeDiskSize, dst)
- if len(dst) < 2 {
- return nil, nil
+func (tst *tsTable) dispatcherLoop(triggerCh chan struct{}, threshold uint64, fastCh, slowCh chan *mergeDispatchRequest) {
+ for {
+ select {
+ case <-tst.loopCloser.CloseNotify():
+ return
+ case _, ok := <-triggerCh:
+ if !ok {
+ return
+ }
+ if tst.dispatchAllMerges(threshold, fastCh, slowCh) {
+ return
+ }
+ }
}
- if _, err := tst.mergePartsThenSendIntroduction(snapshotCreatorMerger, dst,
- toBeMerged, merges, tst.loopCloser.CloseNotify(), mergeTypeFile); err != nil {
- return dst, err
+}
+
+func (tst *tsTable) dispatchAllMerges(threshold uint64, fastCh, slowCh chan *mergeDispatchRequest) bool {
+ for {
+ curSnapshot := tst.currentSnapshot()
+ if curSnapshot == nil {
+ return false
+ }
+ freeDiskSize := tst.freeDiskSpace(tst.root)
+ var dst []*partWrapper
+ dst, toBeMerged := tst.getPartsToMerge(curSnapshot, freeDiskSize, dst)
+ if len(dst) < 2 {
+ curSnapshot.decRef()
+ return false
+ }
+ for _, pw := range dst {
+ pw.incRef()
+ }
+ curSnapshot.decRef()
+
+ tst.inFlightMu.Lock()
+ if tst.inFlight == nil {
+ tst.inFlight = make(map[uint64]struct{})
+ }
+ for _, pw := range dst {
+ tst.inFlight[pw.ID()] = struct{}{}
+ }
+ tst.inFlightMu.Unlock()
+
+ var totalSize uint64
+ for _, pw := range dst {
+ totalSize += pw.p.partMetadata.CompressedSizeBytes
+ }
+
+ lane := mergeLaneSlow
+ targetCh := slowCh
+ if totalSize < threshold {
+ lane = mergeLaneFast
+ targetCh = fastCh
+ }
+
+ typ := fmt.Sprintf("%s_%s", mergeTypeFile, lane)
+ req := &mergeDispatchRequest{
+ parts: dst,
+ toBeMerged: toBeMerged,
+ typ: typ,
+ enqueuedAt: time.Now(),
+ }
+
+ tst.l.Info().
+ Str("lane", lane).
+ Uint64("totalSize", totalSize).
+ Uint64("threshold", threshold).
+ Int("partCount", len(dst)).
+ Msg("dispatching merge")
+
+ select {
+ case targetCh <- req:
+ case <-tst.loopCloser.CloseNotify():
+ tst.inFlightMu.Lock()
+ for _, pw := range dst {
+ delete(tst.inFlight, pw.ID())
+ }
+ tst.inFlightMu.Unlock()
+ for _, pw := range dst {
+ pw.decRef()
+ }
+ return true
+ }
+ }
+}
+
+func (tst *tsTable) mergeLaneWorker(ch chan *mergeDispatchRequest, merges chan *mergerIntroduction) {
+ for req := range ch {
+ if !req.enqueuedAt.IsZero() {
+ tst.incTotalMergeQueueLatency(time.Since(req.enqueuedAt).Seconds(), req.typ)
+ }
+ select {
+ case mergeMaxConcurrencyCh <- struct{}{}:
+ case <-tst.loopCloser.CloseNotify():
+ tst.inFlightMu.Lock()
+ for _, pw := range req.parts {
+ delete(tst.inFlight, pw.ID())
+ }
+ tst.inFlightMu.Unlock()
+ for _, pw := range req.parts {
+ pw.decRef()
+ }
+ return
+ }
+
+ _, mergeErr := tst.mergePartsThenSendIntroduction(
+ snapshotCreatorMerger, req.parts, req.toBeMerged, merges,
+ tst.loopCloser.CloseNotify(), req.typ,
+ )
+ <-mergeMaxConcurrencyCh
+
+ tst.inFlightMu.Lock()
+ for _, pw := range req.parts {
+ delete(tst.inFlight, pw.ID())
+ }
+ tst.inFlightMu.Unlock()
+ for _, pw := range req.parts {
+ pw.decRef()
+ }
+
+ if mergeErr != nil {
+ if !errors.Is(mergeErr, errClosed) {
+ tst.l.Logger.Warn().Err(mergeErr).Str("typ", req.typ).Msg("merge lane worker error")
+ tst.incTotalMergeLoopErr(1)
+ }
+ }
}
- return dst, nil
}
func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, parts []*partWrapper, merged map[uint64]struct{}, merges chan *mergerIntroduction,
@@ -249,12 +412,17 @@ var reservedDiskSpace uint64
func (tst *tsTable) getPartsToMerge(snapshot *snapshot, freeDiskSize uint64, dst []*partWrapper) ([]*partWrapper, map[uint64]struct{}) {
var parts []*partWrapper
+ tst.inFlightMu.RLock()
for _, pw := range snapshot.parts {
if pw.mp != nil || pw.p.partMetadata.TotalCount < 1 {
continue
}
+ if _, inFlight := tst.inFlight[pw.ID()]; inFlight {
+ continue
+ }
parts = append(parts, pw)
}
+ tst.inFlightMu.RUnlock()
dst = tst.option.mergePolicy.getPartsToMerge(dst, parts, freeDiskSize)
if len(dst) == 0 {
diff --git a/banyand/trace/merger_test.go b/banyand/trace/merger_test.go
index f162b400e..131c1bf81 100644
--- a/banyand/trace/merger_test.go
+++ b/banyand/trace/merger_test.go
@@ -23,6 +23,7 @@ import (
"path/filepath"
"reflect"
"testing"
+ "time"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
@@ -1456,3 +1457,251 @@ func Test_flush_cleansUpOnSidxFlushError(t *testing.T) {
sidxPartPath := filepath.Join(tmpPath, sidxDirName, "idx1", fmt.Sprintf("%016x", partID))
require.False(t, fileSystem.IsExist(sidxPartPath), "sidx part directory should be removed on sidx flush failure")
}
+
+func Test_computeSmallMergeThreshold(t *testing.T) {
+ threshold := computeSmallMergeThreshold()
+ require.GreaterOrEqual(t, threshold, uint64(defaultSmallMergeThreshold),
+ "threshold should be at least the default (%d bytes)", defaultSmallMergeThreshold)
+}
+
+func Test_getPartsToMerge_skipsInFlightParts(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+
+ fileSystem := fs.NewLocalFileSystem()
+
+ // Create 4 file parts
+ var allParts []*partWrapper
+ for i := range 4 {
+ partID := uint64(100 + i)
+ mp := generateMemPart()
+ mp.mustInitFromTraces(tsTS1)
+ mp.mustFlush(fileSystem, partPath(tmpPath, partID))
+ p := mustOpenFilePart(partID, tmpPath, fileSystem)
+ p.partMetadata.ID = partID
+ allParts = append(allParts, newPartWrapper(nil, p))
+ releaseMemPart(mp)
+ }
+
+ closer := run.NewCloser(1)
+ defer closer.Done()
+
+ tst := &tsTable{
+ pm: protector.Nop{},
+ fileSystem: fileSystem,
+ root: tmpPath,
+ loopCloser: closer,
+ l: logger.GetLogger("test"),
+ curPartID: 200,
+ option: option{
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ },
+ }
+
+ snp := &snapshot{
+ parts: allParts,
+ epoch: 1,
+ }
+ snp.incRef()
+ defer snp.decRef()
+
+ // Without inFlight: all parts should be eligible
+ dst, _ := tst.getPartsToMerge(snp, uint64(1<<30), nil)
+ require.GreaterOrEqual(t, len(dst), 2, "should find parts to merge when none are in-flight")
+
+ // Mark part 100 and 101 as in-flight
+ tst.inFlight = make(map[uint64]struct{})
+ tst.inFlight[100] = struct{}{}
+ tst.inFlight[101] = struct{}{}
+
+ // With inFlight: parts 100 and 101 should be skipped
+ dst2, _ := tst.getPartsToMerge(snp, uint64(1<<30), nil)
+ if len(dst2) > 0 {
+ for _, pw := range dst2 {
+ require.NotEqual(t, uint64(100), pw.ID(), "part 100 should be skipped (in-flight)")
+ require.NotEqual(t, uint64(101), pw.ID(), "part 101 should be skipped (in-flight)")
+ }
+ }
+
+ // Mark all parts as in-flight: no parts should be eligible
+ tst.inFlight[102] = struct{}{}
+ tst.inFlight[103] = struct{}{}
+ dst3, _ := tst.getPartsToMerge(snp, uint64(1<<30), nil)
+ require.Nil(t, dst3, "no parts should be eligible when all are in-flight")
+}
+
+func Test_dispatchAllMerges_laneClassification(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+
+ fileSystem := fs.NewLocalFileSystem()
+
+ closer := run.NewCloser(1)
+
+ tst := &tsTable{
+ pm: protector.Nop{},
+ fileSystem: fileSystem,
+ root: tmpPath,
+ loopCloser: closer,
+ l: logger.GetLogger("test"),
+ curPartID: 200,
+ option: option{
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ },
+ }
+
+ // Use a very low threshold to force all merges into slow lane
+ smallThreshold := uint64(1) // 1 byte - everything is "large"
+ slowCh := make(chan *mergeDispatchRequest, 10)
+ fastCh := make(chan *mergeDispatchRequest, 10)
+
+ // Create 3 file parts and build a snapshot
+ var allParts []*partWrapper
+ for i := range 3 {
+ partID := uint64(200 + i)
+ mp := generateMemPart()
+ mp.mustInitFromTraces(tsTS1)
+ mp.mustFlush(fileSystem, partPath(tmpPath, partID))
+ p := mustOpenFilePart(partID, tmpPath, fileSystem)
+ p.partMetadata.ID = partID
+ allParts = append(allParts, newPartWrapper(nil, p))
+ releaseMemPart(mp)
+ }
+
+ snp := &snapshot{
+ parts: allParts,
+ epoch: 1,
+ }
+ snp.incRef()
+ tst.snapshot = snp
+ defer snp.decRef()
+
+ // Drain introductions
+ merges := make(chan *mergerIntroduction, 10)
+ go func() {
+ for range merges {
+ }
+ }()
+
+ // Run dispatchAllMerges - it should dispatch to slowCh since threshold is 1 byte
+ dispatchDone := make(chan struct{})
+ go func() {
+ defer close(dispatchDone)
+ tst.dispatchAllMerges(smallThreshold, fastCh, slowCh)
+ }()
+
+ // Should receive on slowCh
+ select {
+ case req := <-slowCh:
+ require.Equal(t, fmt.Sprintf("%s_%s", mergeTypeFile, mergeLaneSlow), req.typ)
+ require.GreaterOrEqual(t, len(req.parts), 2, "should merge at least 2 parts")
+ // Verify inFlight was set
+ tst.inFlightMu.RLock()
+ for _, pw := range req.parts {
+ _, exists := tst.inFlight[pw.ID()]
+ require.True(t, exists, "part %d should be in inFlight map", pw.ID())
+ }
+ tst.inFlightMu.RUnlock()
+ // Don't clear inFlight — dispatchAllMerges loop will find no eligible parts and exit naturally
+ case <-fastCh:
+ t.Fatal("should not dispatch to fast lane with threshold=1")
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out waiting for dispatch")
+ }
+
+ // Wait for dispatchAllMerges to finish (it exits when all parts are in inFlight)
+ select {
+ case <-dispatchDone:
+ case <-time.After(5 * time.Second):
+ t.Fatal("dispatchAllMerges did not finish")
+ }
+
+ // Close the closer after dispatchAllMerges has exited
+ closer.Done()
+ closer.CloseThenWait()
+ close(fastCh)
+ close(slowCh)
+ close(merges)
+}
+
+func Test_mergeLaneWorker_cleansUpInFlight(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+
+ fileSystem := fs.NewLocalFileSystem()
+
+ closer := run.NewCloser(1)
+
+ tst := &tsTable{
+ pm: protector.Nop{},
+ fileSystem: fileSystem,
+ root: tmpPath,
+ loopCloser: closer,
+ l: logger.GetLogger("test"),
+ curPartID: 0,
+ option: option{
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ },
+ inFlight: make(map[uint64]struct{}),
+ }
+
+ // Create 2 file parts to merge
+ var parts []*partWrapper
+ for i := range 2 {
+ partID := uint64(300 + i)
+ mp := generateMemPart()
+ mp.mustInitFromTraces(tsTS1)
+ mp.mustFlush(fileSystem, partPath(tmpPath, partID))
+ p := mustOpenFilePart(partID, tmpPath, fileSystem)
+ p.partMetadata.ID = partID
+ parts = append(parts, newPartWrapper(nil, p))
+ releaseMemPart(mp)
+ }
+
+ // Mark parts as inFlight
+ toBeMerged := make(map[uint64]struct{})
+ for _, pw := range parts {
+ tst.inFlight[pw.ID()] = struct{}{}
+ toBeMerged[pw.ID()] = struct{}{}
+ }
+
+ require.Len(t, tst.inFlight, 2)
+
+ // Create channels
+ ch := make(chan *mergeDispatchRequest, 1)
+ merges := make(chan *mergerIntroduction, 1)
+
+ // Start worker in background
+ workerDone := make(chan struct{})
+ go func() {
+ defer close(workerDone)
+ tst.mergeLaneWorker(ch, merges)
+ }()
+
+ // Send a dispatch request
+ ch <- &mergeDispatchRequest{
+ parts: parts,
+ toBeMerged: toBeMerged,
+ typ: fmt.Sprintf("%s_%s", mergeTypeFile, mergeLaneFast),
+ }
+ close(ch)
+
+ // The worker will try to merge and send introduction.
+ // Since there's no introducerLoop, mergePartsThenSendIntroduction
+ // will block on mi.applied. Close the closer to unblock it.
+ time.Sleep(100 * time.Millisecond)
+ closer.Done()
+ closer.CloseThenWait()
+
+ // Wait for worker to finish
+ select {
+ case <-workerDone:
+ case <-time.After(5 * time.Second):
+ t.Fatal("worker did not finish")
+ }
+
+ // Verify inFlight was cleaned up
+ tst.inFlightMu.RLock()
+ require.Empty(t, tst.inFlight, "inFlight map should be empty after worker completes")
+ tst.inFlightMu.RUnlock()
+}
diff --git a/banyand/trace/metrics.go b/banyand/trace/metrics.go
index 95479e817..9066fca3b 100644
--- a/banyand/trace/metrics.go
+++ b/banyand/trace/metrics.go
@@ -61,9 +61,10 @@ type metrics struct {
totalFlushIntroLatency meter.Counter
totalFlushLatency meter.Counter
- totalMergedParts meter.Counter
- totalMergeLatency meter.Counter
- totalMerged meter.Counter
+ totalMergedParts meter.Counter
+ totalMergeLatency meter.Counter
+ totalMerged meter.Counter
+ totalMergeQueueLatency meter.Counter
tbMetrics
}
@@ -250,6 +251,13 @@ func (tst *tsTable) incTotalMerged(delta int, typ string) {
tst.metrics.totalMerged.Inc(float64(delta), typ)
}
+func (tst *tsTable) incTotalMergeQueueLatency(delta float64, typ string) {
+ if tst == nil || tst.metrics == nil {
+ return
+ }
+ tst.metrics.totalMergeQueueLatency.Inc(delta, typ)
+}
+
func (tst *tsTable) addPendingDataCount(delta int64) {
tst.pendingDataCount.Add(delta)
if tst.metrics == nil {
@@ -301,9 +309,14 @@ func (m *metrics) DeleteAll() {
m.totalMergedParts.Delete("mem")
m.totalMergeLatency.Delete("mem")
m.totalMerged.Delete("mem")
- m.totalMergedParts.Delete("file")
- m.totalMergeLatency.Delete("file")
- m.totalMerged.Delete("file")
+ m.totalMergedParts.Delete("file_fast")
+ m.totalMergeLatency.Delete("file_fast")
+ m.totalMerged.Delete("file_fast")
+ m.totalMergedParts.Delete("file_slow")
+ m.totalMergeLatency.Delete("file_slow")
+ m.totalMerged.Delete("file_slow")
+ m.totalMergeQueueLatency.Delete("file_fast")
+ m.totalMergeQueueLatency.Delete("file_slow")
}
func (s *supplier) newMetrics(p common.Position) storage.Metrics {
@@ -335,6 +348,7 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalMergedParts: factory.NewCounter("total_merged_parts", "type"),
totalMergeLatency: factory.NewCounter("total_merge_latency", "type"),
totalMerged: factory.NewCounter("total_merged", "type"),
+ totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type"),
tbMetrics: tbMetrics{
totalMemParts: factory.NewGauge("total_mem_part", common.ShardLabelNames()...),
totalMemElements: factory.NewGauge("total_mem_elements", common.ShardLabelNames()...),
@@ -381,6 +395,7 @@ func (qs *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observa
totalMergedParts: factory.NewCounter("total_merged_parts", "type"),
totalMergeLatency: factory.NewCounter("total_merge_latency", "type"),
totalMerged: factory.NewCounter("total_merged", "type"),
+ totalMergeQueueLatency: factory.NewCounter("total_merge_queue_latency", "type"),
tbMetrics: tbMetrics{
totalMemParts: factory.NewGauge("total_mem_part", common.ShardLabelNames()...),
totalMemElements: factory.NewGauge("total_mem_elements", common.ShardLabelNames()...),
diff --git a/banyand/trace/tstable.go b/banyand/trace/tstable.go
index dcb68f9b9..59e797725 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -65,6 +65,8 @@ type tsTable struct {
option option
curPartID uint64
pendingDataCount atomic.Int64
+ inFlight map[uint64]struct{}
+ inFlightMu sync.RWMutex
sync.RWMutex
shardID common.ShardID
}