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/30 05:11:23 UTC
(skywalking-banyandb) 19/24: fix(measure,stream,trace): eliminate flusher/introducer data race on shutdown (#1094)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch v0.10.x
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 74d9f7409ed6b48ac98fb75f519318ae49140b30
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Mon Apr 27 14:27:38 2026 +0800
fix(measure,stream,trace): eliminate flusher/introducer data race on shutdown (#1094)
---
banyand/measure/flusher.go | 5 +-
banyand/measure/introducer.go | 39 +++-
banyand/measure/merger.go | 6 +-
banyand/measure/syncer.go | 6 +-
banyand/measure/tstable.go | 10 +-
banyand/stream/flusher.go | 5 +-
banyand/stream/introducer.go | 39 +++-
banyand/stream/merger.go | 6 +-
banyand/stream/syncer.go | 6 +-
banyand/stream/tstable.go | 10 +-
banyand/trace/flusher.go | 5 +-
banyand/trace/introducer.go | 39 +++-
banyand/trace/merger.go | 6 +-
banyand/trace/merger_test.go | 417 ++++++++++++++++++++++++++++++++++++++++++
banyand/trace/syncer.go | 6 +-
banyand/trace/tstable.go | 10 +-
16 files changed, 543 insertions(+), 72 deletions(-)
diff --git a/banyand/measure/flusher.go b/banyand/measure/flusher.go
index c0390f50b..38c01d751 100644
--- a/banyand/measure/flusher.go
+++ b/banyand/measure/flusher.go
@@ -204,10 +204,7 @@ func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction)
case <-tst.loopCloser.CloseNotify():
return
}
- select {
- case <-ind.applied:
- case <-tst.loopCloser.CloseNotify():
- }
+ <-ind.applied
tst.incTotalFlushIntroLatency(time.Since(end).Seconds())
}
diff --git a/banyand/measure/introducer.go b/banyand/measure/introducer.go
index 92dc02c4e..53fb8df5b 100644
--- a/banyand/measure/introducer.go
+++ b/banyand/measure/introducer.go
@@ -117,7 +117,23 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
for {
select {
case <-tst.loopCloser.CloseNotify():
- return
+ // Drain pending introductions so senders don't block on applied.
+ for {
+ select {
+ case next := <-tst.introductions:
+ tst.introducePart(next, epoch)
+ epoch++
+ case next := <-flushCh:
+ tst.introduceFlushed(next, epoch)
+ epoch++
+ case next := <-mergeCh:
+ tst.introduceMerged(next, epoch)
+ epoch++
+ default:
+ tst.gc.clean()
+ return
+ }
+ }
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
tst.introducePart(next, epoch)
@@ -151,7 +167,26 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, me
for {
select {
case <-tst.loopCloser.CloseNotify():
- return
+ // Drain pending introductions so senders don't block on applied.
+ for {
+ select {
+ case next := <-tst.introductions:
+ tst.introducePart(next, epoch)
+ epoch++
+ case next := <-flushCh:
+ tst.introduceFlushed(next, epoch)
+ epoch++
+ case next := <-mergeCh:
+ tst.introduceMerged(next, epoch)
+ epoch++
+ case next := <-syncCh:
+ tst.introduceSync(next, epoch)
+ epoch++
+ default:
+ tst.gc.clean()
+ return
+ }
+ }
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
tst.introducePart(next, epoch)
diff --git a/banyand/measure/merger.go b/banyand/measure/merger.go
index ef359288c..206dee25d 100644
--- a/banyand/measure/merger.go
+++ b/banyand/measure/merger.go
@@ -163,11 +163,7 @@ func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, part
case <-tst.loopCloser.CloseNotify():
return newPart, errClosed
}
- select {
- case <-mi.applied:
- case <-tst.loopCloser.CloseNotify():
- return newPart, errClosed
- }
+ <-mi.applied
return newPart, nil
}
diff --git a/banyand/measure/syncer.go b/banyand/measure/syncer.go
index 7b06e0c19..a62f42870 100644
--- a/banyand/measure/syncer.go
+++ b/banyand/measure/syncer.go
@@ -466,11 +466,7 @@ func (tst *tsTable) sendSyncIntroduction(partsToSync []*part, syncCh chan *syncI
case <-tst.loopCloser.CloseNotify():
return errClosed
}
- select {
- case <-si.applied:
- case <-tst.loopCloser.CloseNotify():
- return errClosed
- }
+ <-si.applied
return nil
}
diff --git a/banyand/measure/tstable.go b/banyand/measure/tstable.go
index 6c3fff58e..34e989601 100644
--- a/banyand/measure/tstable.go
+++ b/banyand/measure/tstable.go
@@ -364,10 +364,7 @@ func (tst *tsTable) mustAddFilePart(partID uint64) {
ind.part.decRef()
return
}
- select {
- case <-ind.applied:
- case <-tst.loopCloser.CloseNotify():
- }
+ <-ind.applied
}
func (tst *tsTable) mustAddMemPart(mp *memPart) {
@@ -388,10 +385,7 @@ func (tst *tsTable) mustAddMemPart(mp *memPart) {
ind.part.decRef()
return
}
- select {
- case <-ind.applied:
- case <-tst.loopCloser.CloseNotify():
- }
+ <-ind.applied
tst.incTotalWritten(int(totalCount))
tst.incTotalBatch(1)
tst.incTotalBatchIntroLatency(time.Since(startTime).Seconds())
diff --git a/banyand/stream/flusher.go b/banyand/stream/flusher.go
index 04788af0c..f2c7f0742 100644
--- a/banyand/stream/flusher.go
+++ b/banyand/stream/flusher.go
@@ -234,10 +234,7 @@ func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction)
case <-tst.loopCloser.CloseNotify():
return
}
- select {
- case <-ind.applied:
- case <-tst.loopCloser.CloseNotify():
- }
+ <-ind.applied
tst.incTotalFlushIntroLatency(time.Since(end).Seconds())
}
diff --git a/banyand/stream/introducer.go b/banyand/stream/introducer.go
index e2255578d..621ea4238 100644
--- a/banyand/stream/introducer.go
+++ b/banyand/stream/introducer.go
@@ -146,7 +146,23 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
for {
select {
case <-tst.loopCloser.CloseNotify():
- return
+ // Drain pending introductions so senders don't block on applied.
+ for {
+ select {
+ case next := <-tst.introductions:
+ tst.introducePart(next, epoch)
+ epoch++
+ case next := <-flushCh:
+ tst.introduceFlushed(next, epoch)
+ epoch++
+ case next := <-mergeCh:
+ tst.introduceMerged(next, epoch)
+ epoch++
+ default:
+ tst.gc.clean()
+ return
+ }
+ }
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
tst.introducePart(next, epoch)
@@ -180,7 +196,26 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, me
for {
select {
case <-tst.loopCloser.CloseNotify():
- return
+ // Drain pending introductions so senders don't block on applied.
+ for {
+ select {
+ case next := <-tst.introductions:
+ tst.introducePart(next, epoch)
+ epoch++
+ case next := <-flushCh:
+ tst.introduceFlushed(next, epoch)
+ epoch++
+ case next := <-mergeCh:
+ tst.introduceMerged(next, epoch)
+ epoch++
+ case next := <-syncCh:
+ tst.introduceSync(next, epoch)
+ epoch++
+ default:
+ tst.gc.clean()
+ return
+ }
+ }
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
tst.introducePart(next, epoch)
diff --git a/banyand/stream/merger.go b/banyand/stream/merger.go
index 2646c0b8b..ba7486878 100644
--- a/banyand/stream/merger.go
+++ b/banyand/stream/merger.go
@@ -162,11 +162,7 @@ func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, part
case <-tst.loopCloser.CloseNotify():
return newPart, errClosed
}
- select {
- case <-mi.applied:
- case <-tst.loopCloser.CloseNotify():
- return newPart, errClosed
- }
+ <-mi.applied
return newPart, nil
}
diff --git a/banyand/stream/syncer.go b/banyand/stream/syncer.go
index 37d3040c9..55c39c1b4 100644
--- a/banyand/stream/syncer.go
+++ b/banyand/stream/syncer.go
@@ -439,11 +439,7 @@ func (tst *tsTable) sendSyncIntroduction(partsToSync []*part, syncCh chan *syncI
case <-tst.loopCloser.CloseNotify():
return errClosed
}
- select {
- case <-si.applied:
- case <-tst.loopCloser.CloseNotify():
- return errClosed
- }
+ <-si.applied
return nil
}
diff --git a/banyand/stream/tstable.go b/banyand/stream/tstable.go
index 9f3f5e7a0..40f72581f 100644
--- a/banyand/stream/tstable.go
+++ b/banyand/stream/tstable.go
@@ -375,10 +375,7 @@ func (tst *tsTable) mustAddFilePart(partID uint64) {
ind.part.decRef()
return
}
- select {
- case <-ind.applied:
- case <-tst.loopCloser.CloseNotify():
- }
+ <-ind.applied
}
func (tst *tsTable) mustAddMemPart(mp *memPart) {
@@ -399,10 +396,7 @@ func (tst *tsTable) mustAddMemPart(mp *memPart) {
ind.part.decRef()
return
}
- select {
- case <-ind.applied:
- case <-tst.loopCloser.CloseNotify():
- }
+ <-ind.applied
tst.incTotalWritten(int(totalCount))
tst.incTotalBatch(1)
tst.incTotalBatchIntroLatency(time.Since(startTime).Seconds())
diff --git a/banyand/trace/flusher.go b/banyand/trace/flusher.go
index 8ebe7baf0..e82645476 100644
--- a/banyand/trace/flusher.go
+++ b/banyand/trace/flusher.go
@@ -273,10 +273,7 @@ func (tst *tsTable) flush(snapshot *snapshot, flushCh chan *flusherIntroduction)
}
return
}
- select {
- case <-ind.applied:
- case <-tst.loopCloser.CloseNotify():
- }
+ <-ind.applied
tst.incTotalFlushIntroLatency(time.Since(end).Seconds())
}
diff --git a/banyand/trace/introducer.go b/banyand/trace/introducer.go
index 93e23da08..b72047fd0 100644
--- a/banyand/trace/introducer.go
+++ b/banyand/trace/introducer.go
@@ -159,7 +159,23 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
for {
select {
case <-tst.loopCloser.CloseNotify():
- return
+ // Drain pending introductions so senders don't block on applied.
+ for {
+ select {
+ case next := <-tst.introductions:
+ tst.introducePart(next, epoch)
+ epoch++
+ case next := <-flushCh:
+ tst.introduceFlushed(next, epoch)
+ epoch++
+ case next := <-mergeCh:
+ tst.introduceMerged(next, epoch)
+ epoch++
+ default:
+ tst.gc.clean()
+ return
+ }
+ }
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
tst.introducePart(next, epoch)
@@ -193,7 +209,26 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, me
for {
select {
case <-tst.loopCloser.CloseNotify():
- return
+ // Drain pending introductions so senders don't block on applied.
+ for {
+ select {
+ case next := <-tst.introductions:
+ tst.introducePart(next, epoch)
+ epoch++
+ case next := <-flushCh:
+ tst.introduceFlushedForSync(next, epoch)
+ epoch++
+ case next := <-mergeCh:
+ tst.introduceMerged(next, epoch)
+ epoch++
+ case next := <-syncCh:
+ tst.introduceSync(next, epoch)
+ epoch++
+ default:
+ tst.gc.clean()
+ return
+ }
+ }
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
tst.introducePart(next, epoch)
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 875f98fb8..480897746 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -209,11 +209,7 @@ func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, part
case <-tst.loopCloser.CloseNotify():
return newPart, errClosed
}
- select {
- case <-mi.applied:
- case <-tst.loopCloser.CloseNotify():
- return newPart, errClosed
- }
+ <-mi.applied
return newPart, nil
}
diff --git a/banyand/trace/merger_test.go b/banyand/trace/merger_test.go
index f162b400e..f091cd166 100644
--- a/banyand/trace/merger_test.go
+++ b/banyand/trace/merger_test.go
@@ -1456,3 +1456,420 @@ 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 m := range merges {
+ _ = m
+ }
+ }()
+
+ // 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, mergeTypeFile, req.typ)
+ require.Equal(t, mergeLaneSlow, req.lane)
+ 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 a mock introducer to consume from merges and close applied.
+ mockIntroducerDone := make(chan struct{})
+ go func() {
+ defer close(mockIntroducerDone)
+ for mi := range merges {
+ if mi.newPart != nil {
+ mi.newPart.decRef()
+ }
+ if mi.applied != nil {
+ close(mi.applied)
+ }
+ }
+ }()
+
+ // 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: mergeTypeFile,
+ lane: mergeLaneFast,
+ }
+ close(ch)
+
+ // Close the closer to signal shutdown.
+ 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")
+ }
+ close(merges)
+ <-mockIntroducerDone
+
+ // Verify inFlight was cleaned up
+ tst.inFlightMu.RLock()
+ require.Empty(t, tst.inFlight, "inFlight map should be empty after worker completes")
+ tst.inFlightMu.RUnlock()
+}
+
+func Test_mergePartsWithConflictingTagTypes(t *testing.T) {
+ tests := []struct {
+ name string
+ parts []*traces
+ wantTagTypes map[string]pbv1.ValueType
+ wantDeleted []string
+ }{
+ {
+ name: "multiple parts with int and str for same tag",
+ parts: []*traces{
+ {
+ traceIDs: []string{"trace1"},
+ timestamps: []int64{1},
+ tags: [][]*tagValue{
+ {
+ {tag: "state", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)},
+ {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("svc1")},
+ },
+ },
+ spans: [][]byte{[]byte("span1")},
+ spanIDs: []string{"span1"},
+ },
+ {
+ traceIDs: []string{"trace2"},
+ timestamps: []int64{2},
+ tags: [][]*tagValue{
+ {
+ {tag: "state", valueType: pbv1.ValueTypeStr, value: []byte("active")},
+ {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("svc2")},
+ },
+ },
+ spans: [][]byte{[]byte("span2")},
+ spanIDs: []string{"span2"},
+ },
+ },
+ wantTagTypes: map[string]pbv1.ValueType{
+ "state#int": pbv1.ValueTypeInt64,
+ "state#str": pbv1.ValueTypeStr,
+ "name": pbv1.ValueTypeStr,
+ },
+ wantDeleted: []string{"state"},
+ },
+ {
+ name: "merge typed-suffix part with plain part",
+ parts: []*traces{
+ {
+ traceIDs: []string{"trace1"},
+ timestamps: []int64{1},
+ tags: [][]*tagValue{
+ {{tag: "state#int", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)}},
+ },
+ spans: [][]byte{[]byte("span1")},
+ spanIDs: []string{"span1"},
+ },
+ {
+ traceIDs: []string{"trace2"},
+ timestamps: []int64{2},
+ tags: [][]*tagValue{
+ {{tag: "state", valueType: pbv1.ValueTypeStr, value: []byte("pending")}},
+ },
+ spans: [][]byte{[]byte("span2")},
+ spanIDs: []string{"span2"},
+ },
+ },
+ wantTagTypes: map[string]pbv1.ValueType{
+ "state#int": pbv1.ValueTypeInt64,
+ "state#str": pbv1.ValueTypeStr,
+ },
+ wantDeleted: []string{"state"},
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+
+ partIDs := make([]uint64, len(tt.parts))
+ for idx := range partIDs {
+ partIDs[idx] = uint64(idx)
+ }
+ dstID := uint64(999)
+ dstPath := mergePartsWithConflictTags(t, fileSystem, tmpPath, tt.parts, partIDs, dstID)
+
+ mergedPart := mustOpenFilePart(dstID, tmpPath, fileSystem)
+ for tag, wantType := range tt.wantTagTypes {
+ gotType, exists := mergedPart.tagType[tag]
+ require.True(t, exists, "expected tag type entry for %q", tag)
+ require.Equal(t, wantType, gotType, "tag type mismatch for %q", tag)
+ }
+ for _, tag := range tt.wantDeleted {
+ _, ok := mergedPart.tagType[tag]
+ require.False(t, ok, "tag %q should not exist in tagType after merge", tag)
+ dataFile := filepath.Join(dstPath, tag+tagsFilenameExt)
+ require.False(t, fileSystem.IsExist(dataFile), "tag data file %q should not exist", dataFile)
+ metaFile := filepath.Join(dstPath, tag+tagsMetadataFilenameExt)
+ require.False(t, fileSystem.IsExist(metaFile), "tag metadata file %q should not exist", metaFile)
+ }
+ })
+ }
+}
+
+func mergePartsWithConflictTags(t *testing.T, fileSystem fs.FileSystem, tmpPath string, tsList []*traces, partIDs []uint64, dstID uint64) string {
+ t.Helper()
+ var parts []*partWrapper
+ for i, traces := range tsList {
+ mp := generateMemPart()
+ mp.mustInitFromTraces(traces)
+ mp.mustFlush(fileSystem, partPath(tmpPath, partIDs[i]))
+ p := mustOpenFilePart(partIDs[i], tmpPath, fileSystem)
+ parts = append(parts, &partWrapper{p: p})
+ releaseMemPart(mp)
+ }
+
+ conflictTags := collectConflictTags(parts)
+
+ var pmi []*partMergeIter
+ var traceSize uint64
+ for i := range tsList {
+ p := mustOpenFilePart(partIDs[i], tmpPath, fileSystem)
+ iter := generatePartMergeIter()
+ iter.mustInitFromPart(p)
+ pmi = append(pmi, iter)
+ traceSize += uint64(len(tsList[i].traceIDs))
+ }
+
+ br := generateBlockReader()
+ br.init(pmi)
+ bw := generateBlockWriter()
+ dstPath := partPath(tmpPath, dstID)
+ bw.mustInitForFilePart(fileSystem, dstPath, false, int(traceSize))
+
+ closeCh := make(chan struct{})
+ pm, tf, tagTypes, mergeErr := mergeBlocks(closeCh, bw, br, conflictTags)
+ close(closeCh)
+ require.NoError(t, mergeErr)
+ require.NotNil(t, pm)
+
+ releaseBlockWriter(bw)
+ releaseBlockReader(br)
+ for _, iter := range pmi {
+ releasePartMergeIter(iter)
+ }
+
+ pm.mustWriteMetadata(fileSystem, dstPath)
+ tf.mustWriteTraceIDFilter(fileSystem, dstPath)
+ tagTypes.mustWriteTagType(fileSystem, dstPath)
+ fileSystem.SyncPath(dstPath)
+ return dstPath
+}
diff --git a/banyand/trace/syncer.go b/banyand/trace/syncer.go
index eb942b107..a03471cdc 100644
--- a/banyand/trace/syncer.go
+++ b/banyand/trace/syncer.go
@@ -475,11 +475,7 @@ func (tst *tsTable) handleSyncIntroductions(partsToSync []*part, syncCh chan *sy
return errClosed
}
- select {
- case <-si.applied:
- case <-tst.loopCloser.CloseNotify():
- return errClosed
- }
+ <-si.applied
return nil
}
diff --git a/banyand/trace/tstable.go b/banyand/trace/tstable.go
index 05d5e11d1..cad0dea0b 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -518,10 +518,7 @@ func (tst *tsTable) mustAddFilePart(partID uint64, sidxFilePartsMap map[string]s
ind.part.decRef()
return
}
- select {
- case <-ind.applied:
- case <-tst.loopCloser.CloseNotify():
- }
+ <-ind.applied
}
func (tst *tsTable) mustAddMemPart(mp *memPart, sidxReqsMap map[string]*sidx.MemPart) {
@@ -543,10 +540,7 @@ func (tst *tsTable) mustAddMemPart(mp *memPart, sidxReqsMap map[string]*sidx.Mem
ind.part.decRef()
return
}
- select {
- case <-ind.applied:
- case <-tst.loopCloser.CloseNotify():
- }
+ <-ind.applied
tst.incTotalWritten(int(totalCount))
tst.incTotalBatch(1)
tst.incTotalBatchIntroLatency(time.Since(startTime).Seconds())