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 2025/08/14 02:16:01 UTC
(skywalking-banyandb) branch new-feature created (now 27c821e7)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch new-feature
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 27c821e7 Enhance flusher and introducer loops to support merging operations
This branch includes the following new commits:
new 27c821e7 Enhance flusher and introducer loops to support merging operations
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-banyandb) 01/01: Enhance flusher and introducer loops to support merging operations
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch new-feature
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 27c821e7338aa249173412f847268e6746332f66
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Aug 14 07:10:26 2025 +0800
Enhance flusher and introducer loops to support merging operations
- Updated `flusherLoopNoMerger` to merge parts before flushing, improving efficiency by eliminating the need for a separate merge loop.
- Modified `introducerLoopWithSync` to handle merged data, allowing for better integration of flushing and merging processes.
- Adjusted the `startLoopNoMerge` function to include a merge channel, facilitating communication between the introducer and flusher.
- Removed unused Marshal and Unmarshal methods from `memPart`, streamlining the codebase.
These changes optimize the data handling process within the system, ensuring smoother operations during data flushing and merging.
---
CHANGES.md | 1 +
banyand/measure/flusher.go | 37 +++--
banyand/measure/introducer.go | 10 +-
banyand/measure/part.go | 261 ---------------------------------
banyand/measure/topn.go | 6 +-
banyand/measure/tstable.go | 5 +-
banyand/protector/protector.go | 6 +-
banyand/stream/part.go | 293 -------------------------------------
banyand/stream/part_test.go | 319 -----------------------------------------
pkg/test/setup/setup.go | 2 +
10 files changed, 52 insertions(+), 888 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index dc0e9afb..df0ebd87 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -30,6 +30,7 @@ Release Notes.
- Backup/Restore: Add support for AWS S3, Google Cloud Storage (GCS), and Azure Blob Storage as remote targets for backup and restore operations.
- Improve TopN processing by adding "source" tag to track node-specific data, enhancing data handling across distributed nodes
- Implement Login with Username/Password authentication in BanyanDB
+- Enhance flusher and introducer loops to support merging operations, improving efficiency by eliminating the need for a separate merge loop and optimizing data handling process during flushing and merging
### Bug Fixes
diff --git a/banyand/measure/flusher.go b/banyand/measure/flusher.go
index 01f4443f..a16f2f53 100644
--- a/banyand/measure/flusher.go
+++ b/banyand/measure/flusher.go
@@ -186,8 +186,10 @@ func (tst *tsTable) persistSnapshot(snapshot *snapshot) {
tst.gc.registerSnapshot(snapshot)
}
-// flusherLoopNoMerger is like flusherLoop but does not perform merging, only flushing.
-func (tst *tsTable) flusherLoopNoMerger(flushCh chan *flusherIntroduction, introducerWatcher, flusherWatcher watcher.Channel, epoch uint64) {
+// flusherLoopNoMerger is like flusherLoop but merges parts before flushing without a separate merge loop.
+func (tst *tsTable) flusherLoopNoMerger(flushCh chan *flusherIntroduction, mergeCh chan *mergerIntroduction,
+ introducerWatcher, flusherWatcher watcher.Channel, epoch uint64,
+) {
defer tst.loopCloser.Done()
epochWatcher := introducerWatcher.Add(epoch, tst.loopCloser.CloseNotify())
if epochWatcher == nil {
@@ -211,12 +213,32 @@ func (tst *tsTable) flusherLoopNoMerger(flushCh chan *flusherIntroduction, intro
}()
curSnapshot := tst.currentSnapshot()
if curSnapshot != nil {
- defer func() {
- curSnapshot.decRef()
- curSnapshot = nil
- }()
- tst.flush(curSnapshot, flushCh)
+ flusherWatchers = tst.pauseFlusherToPileupMemParts(epoch, flusherWatcher, flusherWatchers)
+ curSnapshot.decRef()
+ curSnapshot = nil
+ }
+ tst.RLock()
+ if tst.snapshot != nil && tst.snapshot.epoch > epoch {
+ curSnapshot = tst.snapshot
+ curSnapshot.incRef()
+ }
+ tst.RUnlock()
+ if curSnapshot != nil {
+ defer curSnapshot.decRef()
+ merged, err := tst.mergeMemParts(curSnapshot, mergeCh)
+ if err != nil {
+ tst.l.Logger.Warn().Err(err).Msgf("cannot merge snapshot: %d", curSnapshot.epoch)
+ tst.incTotalFlushLoopErr(1)
+ return false
+ }
+ if !merged {
+ tst.flush(curSnapshot, flushCh)
+ }
epoch = curSnapshot.epoch
+ // Notify merger to start a new round of merge.
+ // This round might have be triggered in pauseFlusherToPileupMemParts.
+ flusherWatchers.Notify(math.MaxUint64)
+ flusherWatchers = nil
if tst.currentEpoch() != epoch {
tst.incTotalFlushLoopProgress(1)
return false
@@ -228,6 +250,5 @@ func (tst *tsTable) flusherLoopNoMerger(flushCh chan *flusherIntroduction, intro
return
}
}
- flusherWatchers.Notify(epoch)
}
}
diff --git a/banyand/measure/introducer.go b/banyand/measure/introducer.go
index f8f3f811..e60d071a 100644
--- a/banyand/measure/introducer.go
+++ b/banyand/measure/introducer.go
@@ -143,7 +143,9 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
}
}
-func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, syncCh chan *syncIntroduction, watcherCh watcher.Channel, epoch uint64) {
+func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, mergeCh chan *mergerIntroduction,
+ syncCh chan *syncIntroduction, watcherCh watcher.Channel, epoch uint64,
+) {
var introducerWatchers watcher.Epochs
defer tst.loopCloser.Done()
for {
@@ -161,6 +163,12 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, sy
tst.incTotalIntroduceLoopFinished("flush")
tst.gc.clean()
epoch++
+ case next := <-mergeCh:
+ tst.incTotalIntroduceLoopStarted("merge")
+ tst.introduceMerged(next, epoch)
+ tst.incTotalIntroduceLoopFinished("merge")
+ tst.gc.clean()
+ epoch++
case next := <-syncCh:
tst.incTotalIntroduceLoopStarted("sync")
tst.introduceSync(next, epoch)
diff --git a/banyand/measure/part.go b/banyand/measure/part.go
index cca054f2..d7d639ea 100644
--- a/banyand/measure/part.go
+++ b/banyand/measure/part.go
@@ -18,9 +18,7 @@
package measure
import (
- "encoding/json"
"fmt"
- "io"
"path"
"path/filepath"
"sort"
@@ -30,8 +28,6 @@ import (
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/bytes"
- "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/pool"
@@ -152,263 +148,6 @@ func (mp *memPart) reset() {
}
}
-// Marshal serializes the memPart to []byte for network transmission.
-// The format is:
-// - partMetadata (JSON)
-// - meta buffer length + data
-// - primary buffer length + data
-// - timestamps buffer length + data
-// - fieldValues buffer length + data
-// - tagFamilies count + (name + buffer length + data) for each
-// - tagFamilyMetadata count + (name + buffer length + data) for each.
-func (mp *memPart) Marshal() ([]byte, error) {
- bb := bigValuePool.Generate()
- defer bigValuePool.Release(bb)
-
- // Marshal partMetadata as JSON
- metadataBytes, err := json.Marshal(mp.partMetadata)
- if err != nil {
- return nil, fmt.Errorf("cannot marshal partMetadata: %w", err)
- }
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(metadataBytes)))
- bb.Buf = append(bb.Buf, metadataBytes...)
-
- // Marshal meta buffer
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.meta.Buf)))
- bb.Buf = append(bb.Buf, mp.meta.Buf...)
-
- // Marshal primary buffer
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.primary.Buf)))
- bb.Buf = append(bb.Buf, mp.primary.Buf...)
-
- // Marshal timestamps buffer
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.timestamps.Buf)))
- bb.Buf = append(bb.Buf, mp.timestamps.Buf...)
-
- // Marshal fieldValues buffer
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.fieldValues.Buf)))
- bb.Buf = append(bb.Buf, mp.fieldValues.Buf...)
-
- // Marshal tagFamilies
- if mp.tagFamilies == nil {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
- } else {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFamilies)))
- for name, tf := range mp.tagFamilies {
- // Write name length and name
- nameBytes := []byte(name)
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
- bb.Buf = append(bb.Buf, nameBytes...)
- // Write buffer length and data
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tf.Buf)))
- bb.Buf = append(bb.Buf, tf.Buf...)
- }
- }
-
- // Marshal tagFamilyMetadata
- if mp.tagFamilyMetadata == nil {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
- } else {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFamilyMetadata)))
- for name, tfh := range mp.tagFamilyMetadata {
- // Write name length and name
- nameBytes := []byte(name)
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
- bb.Buf = append(bb.Buf, nameBytes...)
- // Write buffer length and data
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tfh.Buf)))
- bb.Buf = append(bb.Buf, tfh.Buf...)
- }
- }
-
- result := make([]byte, len(bb.Buf))
- copy(result, bb.Buf)
- return result, nil
-}
-
-// Unmarshal deserializes the memPart from []byte.
-func (mp *memPart) Unmarshal(data []byte) error {
- mp.reset()
-
- // Unmarshal partMetadata
- tail, metadataLen := encoding.BytesToVarUint64(data)
- if uint64(len(tail)) < metadataLen {
- return fmt.Errorf("insufficient data for partMetadata: need %d bytes, have %d", metadataLen, len(tail))
- }
- metadataBytes := tail[:metadataLen]
- tail = tail[metadataLen:]
- if err := json.Unmarshal(metadataBytes, &mp.partMetadata); err != nil {
- return fmt.Errorf("cannot unmarshal partMetadata: %w", err)
- }
-
- // Unmarshal meta buffer
- tail, metaLen := encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < metaLen {
- return fmt.Errorf("insufficient data for meta buffer: need %d bytes, have %d", metaLen, len(tail))
- }
- mp.meta.Buf = append(mp.meta.Buf[:0], tail[:metaLen]...)
- tail = tail[metaLen:]
-
- // Unmarshal primary buffer
- tail, primaryLen := encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < primaryLen {
- return fmt.Errorf("insufficient data for primary buffer: need %d bytes, have %d", primaryLen, len(tail))
- }
- mp.primary.Buf = append(mp.primary.Buf[:0], tail[:primaryLen]...)
- tail = tail[primaryLen:]
-
- // Unmarshal timestamps buffer
- tail, timestampsLen := encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < timestampsLen {
- return fmt.Errorf("insufficient data for timestamps buffer: need %d bytes, have %d", timestampsLen, len(tail))
- }
- mp.timestamps.Buf = append(mp.timestamps.Buf[:0], tail[:timestampsLen]...)
- tail = tail[timestampsLen:]
-
- // Unmarshal fieldValues buffer
- tail, fieldValuesLen := encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < fieldValuesLen {
- return fmt.Errorf("insufficient data for fieldValues buffer: need %d bytes, have %d", fieldValuesLen, len(tail))
- }
- mp.fieldValues.Buf = append(mp.fieldValues.Buf[:0], tail[:fieldValuesLen]...)
- tail = tail[fieldValuesLen:]
-
- var nameLen, bufLen uint64
- // Unmarshal tagFamilies
- tail, tagFamiliesCount := encoding.BytesToVarUint64(tail)
- if tagFamiliesCount > 0 {
- mp.tagFamilies = make(map[string]*bytes.Buffer)
- for i := uint64(0); i < tagFamiliesCount; i++ {
- // Read name length and name
- tail, nameLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < nameLen {
- return fmt.Errorf("insufficient data for tag family name: need %d bytes, have %d", nameLen, len(tail))
- }
- name := string(tail[:nameLen])
- tail = tail[nameLen:]
-
- // Read buffer length and data
- tail, bufLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < bufLen {
- return fmt.Errorf("insufficient data for tag family buffer: need %d bytes, have %d", bufLen, len(tail))
- }
- tf := &bytes.Buffer{}
- tf.Buf = append(tf.Buf[:0], tail[:bufLen]...)
- mp.tagFamilies[name] = tf
- tail = tail[bufLen:]
- }
- }
-
- // Unmarshal tagFamilyMetadata
- tail, tagFamilyMetadataCount := encoding.BytesToVarUint64(tail)
- if tagFamilyMetadataCount > 0 {
- mp.tagFamilyMetadata = make(map[string]*bytes.Buffer)
- for i := uint64(0); i < tagFamilyMetadataCount; i++ {
- // Read name length and name
- tail, nameLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < nameLen {
- return fmt.Errorf("insufficient data for tag family metadata name: need %d bytes, have %d", nameLen, len(tail))
- }
- name := string(tail[:nameLen])
- tail = tail[nameLen:]
-
- // Read buffer length and data
- tail, bufLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < bufLen {
- return fmt.Errorf("insufficient data for tag family metadata buffer: need %d bytes, have %d", bufLen, len(tail))
- }
- tfh := &bytes.Buffer{}
- tfh.Buf = append(tfh.Buf[:0], tail[:bufLen]...)
- mp.tagFamilyMetadata[name] = tfh
- tail = tail[bufLen:]
- }
- }
-
- if len(tail) > 0 {
- return fmt.Errorf("unexpected trailing data: %d bytes", len(tail))
- }
-
- return nil
-}
-
-//nolint:unused // This function is used in test files
-func (mp *memPart) mustInitFromPart(p *part) {
- mp.reset()
-
- // Copy part metadata
- mp.partMetadata = p.partMetadata
-
- // Read primary data
- sr := p.primary.SequentialRead()
- data, err := io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read primary data from %s: %s", p.primary.Path(), err)
- }
- fs.MustClose(sr)
- mp.primary.Buf = append(mp.primary.Buf[:0], data...)
-
- // Read timestamps data
- sr = p.timestamps.SequentialRead()
- data, err = io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read timestamps data from %s: %s", p.timestamps.Path(), err)
- }
- fs.MustClose(sr)
- mp.timestamps.Buf = append(mp.timestamps.Buf[:0], data...)
-
- // Read fieldValues data
- sr = p.fieldValues.SequentialRead()
- data, err = io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read fieldValues data from %s: %s", p.fieldValues.Path(), err)
- }
- fs.MustClose(sr)
- mp.fieldValues.Buf = append(mp.fieldValues.Buf[:0], data...)
-
- // Read tag families data
- if p.tagFamilies != nil {
- mp.tagFamilies = make(map[string]*bytes.Buffer)
- mp.tagFamilyMetadata = make(map[string]*bytes.Buffer)
-
- for name, reader := range p.tagFamilies {
- sr = reader.SequentialRead()
- data, err = io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read tag family data from %s: %s", reader.Path(), err)
- }
- fs.MustClose(sr)
-
- mp.tagFamilies[name] = &bytes.Buffer{}
- mp.tagFamilies[name].Buf = append(mp.tagFamilies[name].Buf[:0], data...)
- }
-
- for name, reader := range p.tagFamilyMetadata {
- sr = reader.SequentialRead()
- data, err = io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read tag family metadata from %s: %s", reader.Path(), err)
- }
- fs.MustClose(sr)
-
- mp.tagFamilyMetadata[name] = &bytes.Buffer{}
- mp.tagFamilyMetadata[name].Buf = append(mp.tagFamilyMetadata[name].Buf[:0], data...)
- }
- }
-
- // Encode primaryBlockMetadata to memPart.meta
- // This is the reverse process of mustReadPrimaryBlockMetadata
- bb := bigValuePool.Generate()
- defer bigValuePool.Release(bb)
-
- // Marshal all primaryBlockMetadata entries
- for _, pbm := range p.primaryBlockMetadata {
- bb.Buf = pbm.marshal(bb.Buf)
- }
-
- // Compress the marshaled data
- mp.meta.Buf = zstd.Compress(mp.meta.Buf[:0], bb.Buf, 1)
-}
-
func (mp *memPart) mustInitFromDataPoints(dps *dataPoints) {
mp.reset()
diff --git a/banyand/measure/topn.go b/banyand/measure/topn.go
index de1665d0..326e7412 100644
--- a/banyand/measure/topn.go
+++ b/banyand/measure/topn.go
@@ -95,7 +95,11 @@ func (sr *schemaRepo) getSteamingManager(source *commonv1.Metadata, pipeline que
if v, ok := sr.topNProcessorMap.Load(key); ok {
pre := v.(*topNProcessorManager)
pre.init(sourceMeasure.GetSchema())
- if pre.m.GetMetadata().GetModRevision() < sourceMeasure.schema.GetMetadata().GetModRevision() {
+ var modRevision int64
+ pre.RLock()
+ modRevision = pre.m.GetMetadata().GetModRevision()
+ pre.RUnlock()
+ if modRevision < sourceMeasure.schema.GetMetadata().GetModRevision() {
defer pre.Close()
manager = &topNProcessorManager{
l: sr.l,
diff --git a/banyand/measure/tstable.go b/banyand/measure/tstable.go
index 7f7a288a..721a9faf 100644
--- a/banyand/measure/tstable.go
+++ b/banyand/measure/tstable.go
@@ -130,11 +130,12 @@ func (tst *tsTable) startLoopNoMerge(cur uint64) {
tst.loopCloser = run.NewCloser(1 + 3)
tst.introductions = make(chan *introduction)
flushCh := make(chan *flusherIntroduction)
+ mergeCh := make(chan *mergerIntroduction)
syncCh := make(chan *syncIntroduction)
introducerWatcher := make(watcher.Channel, 1)
flusherWatcher := make(watcher.Channel, 1)
- go tst.introducerLoopWithSync(flushCh, syncCh, introducerWatcher, cur+1)
- go tst.flusherLoopNoMerger(flushCh, introducerWatcher, flusherWatcher, cur)
+ go tst.introducerLoopWithSync(flushCh, mergeCh, syncCh, introducerWatcher, cur+1)
+ go tst.flusherLoopNoMerger(flushCh, mergeCh, introducerWatcher, flusherWatcher, cur)
go tst.syncLoop(syncCh, flusherWatcher)
}
diff --git a/banyand/protector/protector.go b/banyand/protector/protector.go
index 2607f1f9..a1d8a792 100644
--- a/banyand/protector/protector.go
+++ b/banyand/protector/protector.go
@@ -74,9 +74,9 @@ func NewMemory(omr observability.MetricsRegistry) Memory {
omr: omr,
blockedChan: make(chan struct{}, queueSize),
closed: make(chan struct{}),
-
- limitGauge: factory.NewGauge("limit"),
- usageGauge: factory.NewGauge("usage"),
+ l: logger.GetLogger("memory-protector"),
+ limitGauge: factory.NewGauge("limit"),
+ usageGauge: factory.NewGauge("usage"),
}
}
diff --git a/banyand/stream/part.go b/banyand/stream/part.go
index e9fac1cd..07e5a22f 100644
--- a/banyand/stream/part.go
+++ b/banyand/stream/part.go
@@ -18,9 +18,7 @@
package stream
import (
- "encoding/json"
"fmt"
- "io"
"path"
"path/filepath"
"sort"
@@ -30,8 +28,6 @@ import (
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/bytes"
- "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/pool"
@@ -164,295 +160,6 @@ func (mp *memPart) reset() {
}
}
-// Marshal serializes the memPart to []byte for network transmission.
-// The format is:
-// - partMetadata (JSON)
-// - meta buffer length + data
-// - primary buffer length + data
-// - timestamps buffer length + data
-// - tagFamilies count + (name + buffer length + data) for each
-// - tagFamilyMetadata count + (name + buffer length + data) for each
-// - tagFamilyFilter count + (name + buffer length + data) for each.
-func (mp *memPart) Marshal() ([]byte, error) {
- bb := bigValuePool.Generate()
- defer bigValuePool.Release(bb)
-
- // Marshal partMetadata as JSON
- metadataBytes, err := json.Marshal(mp.partMetadata)
- if err != nil {
- return nil, fmt.Errorf("cannot marshal partMetadata: %w", err)
- }
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(metadataBytes)))
- bb.Buf = append(bb.Buf, metadataBytes...)
-
- // Marshal meta buffer
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.meta.Buf)))
- bb.Buf = append(bb.Buf, mp.meta.Buf...)
-
- // Marshal primary buffer
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.primary.Buf)))
- bb.Buf = append(bb.Buf, mp.primary.Buf...)
-
- // Marshal timestamps buffer
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.timestamps.Buf)))
- bb.Buf = append(bb.Buf, mp.timestamps.Buf...)
-
- // Marshal tagFamilies
- if mp.tagFamilies == nil {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
- } else {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFamilies)))
- for name, tf := range mp.tagFamilies {
- // Write name length and name
- nameBytes := []byte(name)
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
- bb.Buf = append(bb.Buf, nameBytes...)
- // Write buffer length and data
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tf.Buf)))
- bb.Buf = append(bb.Buf, tf.Buf...)
- }
- }
-
- // Marshal tagFamilyMetadata
- if mp.tagFamilyMetadata == nil {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
- } else {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFamilyMetadata)))
- for name, tfh := range mp.tagFamilyMetadata {
- // Write name length and name
- nameBytes := []byte(name)
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
- bb.Buf = append(bb.Buf, nameBytes...)
- // Write buffer length and data
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tfh.Buf)))
- bb.Buf = append(bb.Buf, tfh.Buf...)
- }
- }
-
- // Marshal tagFamilyFilter
- if mp.tagFamilyFilter == nil {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, 0)
- } else {
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(mp.tagFamilyFilter)))
- for name, tff := range mp.tagFamilyFilter {
- // Write name length and name
- nameBytes := []byte(name)
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(nameBytes)))
- bb.Buf = append(bb.Buf, nameBytes...)
- // Write buffer length and data
- bb.Buf = encoding.VarUint64ToBytes(bb.Buf, uint64(len(tff.Buf)))
- bb.Buf = append(bb.Buf, tff.Buf...)
- }
- }
-
- result := make([]byte, len(bb.Buf))
- copy(result, bb.Buf)
- return result, nil
-}
-
-// Unmarshal deserializes the memPart from []byte.
-func (mp *memPart) Unmarshal(data []byte) error {
- mp.reset()
-
- // Unmarshal partMetadata
- tail, metadataLen := encoding.BytesToVarUint64(data)
- if uint64(len(tail)) < metadataLen {
- return fmt.Errorf("insufficient data for partMetadata: need %d bytes, have %d", metadataLen, len(tail))
- }
- metadataBytes := tail[:metadataLen]
- tail = tail[metadataLen:]
- if err := json.Unmarshal(metadataBytes, &mp.partMetadata); err != nil {
- return fmt.Errorf("cannot unmarshal partMetadata: %w", err)
- }
-
- // Unmarshal meta buffer
- tail, metaLen := encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < metaLen {
- return fmt.Errorf("insufficient data for meta buffer: need %d bytes, have %d", metaLen, len(tail))
- }
- mp.meta.Buf = append(mp.meta.Buf[:0], tail[:metaLen]...)
- tail = tail[metaLen:]
-
- // Unmarshal primary buffer
- tail, primaryLen := encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < primaryLen {
- return fmt.Errorf("insufficient data for primary buffer: need %d bytes, have %d", primaryLen, len(tail))
- }
- mp.primary.Buf = append(mp.primary.Buf[:0], tail[:primaryLen]...)
- tail = tail[primaryLen:]
-
- // Unmarshal timestamps buffer
- tail, timestampsLen := encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < timestampsLen {
- return fmt.Errorf("insufficient data for timestamps buffer: need %d bytes, have %d", timestampsLen, len(tail))
- }
- mp.timestamps.Buf = append(mp.timestamps.Buf[:0], tail[:timestampsLen]...)
- tail = tail[timestampsLen:]
-
- var nameLen, bufLen uint64
- // Unmarshal tagFamilies
- tail, tagFamiliesCount := encoding.BytesToVarUint64(tail)
- if tagFamiliesCount > 0 {
- mp.tagFamilies = make(map[string]*bytes.Buffer)
- for i := uint64(0); i < tagFamiliesCount; i++ {
- // Read name length and name
- tail, nameLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < nameLen {
- return fmt.Errorf("insufficient data for tag family name: need %d bytes, have %d", nameLen, len(tail))
- }
- name := string(tail[:nameLen])
- tail = tail[nameLen:]
-
- // Read buffer length and data
- tail, bufLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < bufLen {
- return fmt.Errorf("insufficient data for tag family buffer: need %d bytes, have %d", bufLen, len(tail))
- }
- tf := &bytes.Buffer{}
- tf.Buf = append(tf.Buf[:0], tail[:bufLen]...)
- mp.tagFamilies[name] = tf
- tail = tail[bufLen:]
- }
- }
-
- // Unmarshal tagFamilyMetadata
- tail, tagFamilyMetadataCount := encoding.BytesToVarUint64(tail)
- if tagFamilyMetadataCount > 0 {
- mp.tagFamilyMetadata = make(map[string]*bytes.Buffer)
- for i := uint64(0); i < tagFamilyMetadataCount; i++ {
- // Read name length and name
- tail, nameLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < nameLen {
- return fmt.Errorf("insufficient data for tag family metadata name: need %d bytes, have %d", nameLen, len(tail))
- }
- name := string(tail[:nameLen])
- tail = tail[nameLen:]
-
- // Read buffer length and data
- tail, bufLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < bufLen {
- return fmt.Errorf("insufficient data for tag family metadata buffer: need %d bytes, have %d", bufLen, len(tail))
- }
- tfh := &bytes.Buffer{}
- tfh.Buf = append(tfh.Buf[:0], tail[:bufLen]...)
- mp.tagFamilyMetadata[name] = tfh
- tail = tail[bufLen:]
- }
- }
-
- // Unmarshal tagFamilyFilter
- tail, tagFamilyFilterCount := encoding.BytesToVarUint64(tail)
- if tagFamilyFilterCount > 0 {
- mp.tagFamilyFilter = make(map[string]*bytes.Buffer)
- for i := uint64(0); i < tagFamilyFilterCount; i++ {
- // Read name length and name
- tail, nameLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < nameLen {
- return fmt.Errorf("insufficient data for tag family filter name: need %d bytes, have %d", nameLen, len(tail))
- }
- name := string(tail[:nameLen])
- tail = tail[nameLen:]
-
- // Read buffer length and data
- tail, bufLen = encoding.BytesToVarUint64(tail)
- if uint64(len(tail)) < bufLen {
- return fmt.Errorf("insufficient data for tag family filter buffer: need %d bytes, have %d", bufLen, len(tail))
- }
- tff := &bytes.Buffer{}
- tff.Buf = append(tff.Buf[:0], tail[:bufLen]...)
- mp.tagFamilyFilter[name] = tff
- tail = tail[bufLen:]
- }
- }
-
- if len(tail) > 0 {
- return fmt.Errorf("unexpected trailing data: %d bytes", len(tail))
- }
-
- return nil
-}
-
-func (mp *memPart) mustInitFromPart(p *part) {
- mp.reset()
-
- // Copy part metadata
- mp.partMetadata = p.partMetadata
-
- // Read primary data
- sr := p.primary.SequentialRead()
- data, err := io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read primary data from %s: %s", p.primary.Path(), err)
- }
- fs.MustClose(sr)
- mp.primary.Buf = append(mp.primary.Buf[:0], data...)
-
- // Read timestamps data
- sr = p.timestamps.SequentialRead()
- data, err = io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read timestamps data from %s: %s", p.timestamps.Path(), err)
- }
- fs.MustClose(sr)
- mp.timestamps.Buf = append(mp.timestamps.Buf[:0], data...)
-
- // Read tag families data
- if p.tagFamilies != nil {
- mp.tagFamilies = make(map[string]*bytes.Buffer)
- mp.tagFamilyMetadata = make(map[string]*bytes.Buffer)
- mp.tagFamilyFilter = make(map[string]*bytes.Buffer)
-
- for name, reader := range p.tagFamilies {
- sr = reader.SequentialRead()
- data, err = io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read tag family data from %s: %s", reader.Path(), err)
- }
- fs.MustClose(sr)
-
- mp.tagFamilies[name] = &bytes.Buffer{}
- mp.tagFamilies[name].Buf = append(mp.tagFamilies[name].Buf[:0], data...)
- }
-
- for name, reader := range p.tagFamilyMetadata {
- sr = reader.SequentialRead()
- data, err = io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read tag family metadata from %s: %s", reader.Path(), err)
- }
- fs.MustClose(sr)
-
- mp.tagFamilyMetadata[name] = &bytes.Buffer{}
- mp.tagFamilyMetadata[name].Buf = append(mp.tagFamilyMetadata[name].Buf[:0], data...)
- }
-
- for name, reader := range p.tagFamilyFilter {
- sr = reader.SequentialRead()
- data, err = io.ReadAll(sr)
- if err != nil {
- logger.Panicf("cannot read tag family filter from %s: %s", reader.Path(), err)
- }
- fs.MustClose(sr)
-
- mp.tagFamilyFilter[name] = &bytes.Buffer{}
- mp.tagFamilyFilter[name].Buf = append(mp.tagFamilyFilter[name].Buf[:0], data...)
- }
- }
-
- // Encode primaryBlockMetadata to memPart.meta
- // This is the reverse process of mustReadPrimaryBlockMetadata
- bb := bigValuePool.Generate()
- defer bigValuePool.Release(bb)
-
- // Marshal all primaryBlockMetadata entries
- for _, pbm := range p.primaryBlockMetadata {
- bb.Buf = pbm.marshal(bb.Buf)
- }
-
- // Compress the marshaled data
- mp.meta.Buf = zstd.Compress(mp.meta.Buf[:0], bb.Buf, 1)
-}
-
func (mp *memPart) mustInitFromElements(es *elements) {
mp.reset()
diff --git a/banyand/stream/part_test.go b/banyand/stream/part_test.go
index 926041a2..f4443998 100644
--- a/banyand/stream/part_test.go
+++ b/banyand/stream/part_test.go
@@ -24,7 +24,6 @@ import (
"github.com/stretchr/testify/require"
"github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/pkg/bytes"
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/fs"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
@@ -181,321 +180,3 @@ var es = &elements{
{}, // empty tagFamilies for seriesID 3
},
}
-
-func TestMustInitFromPart(t *testing.T) {
- // Step 1: Load the global variable "es" to a memPart
- originalMemPart := &memPart{}
- originalMemPart.mustInitFromElements(es)
-
- // Step 2: Flush the memPart
- tmpPath, defFn := test.Space(require.New(t))
- defer defFn()
- epoch := uint64(1)
- path := partPath(tmpPath, epoch)
- fileSystem := fs.NewLocalFileSystem()
- originalMemPart.mustFlush(fileSystem, path)
-
- // Step 3: Init a part from the files the memPart flushed
- part := mustOpenFilePart(epoch, tmpPath, fileSystem)
- defer part.close()
-
- // Step 4: Convert the part to a new memPart
- newMemPart := &memPart{}
- newMemPart.mustInitFromPart(part)
-
- // Step 5: Compare the new memPart with the original memPart
- // Compare part metadata
- assert.NotEqual(t, originalMemPart.partMetadata.ID, newMemPart.partMetadata.ID)
- assert.Equal(t, originalMemPart.partMetadata.BlocksCount, newMemPart.partMetadata.BlocksCount)
- assert.Equal(t, originalMemPart.partMetadata.MinTimestamp, newMemPart.partMetadata.MinTimestamp)
- assert.Equal(t, originalMemPart.partMetadata.MaxTimestamp, newMemPart.partMetadata.MaxTimestamp)
- assert.Equal(t, originalMemPart.partMetadata.TotalCount, newMemPart.partMetadata.TotalCount)
-
- // Compare primary data
- assert.Equal(t, originalMemPart.primary.Buf, newMemPart.primary.Buf)
-
- // Compare timestamps data
- assert.Equal(t, originalMemPart.timestamps.Buf, newMemPart.timestamps.Buf)
-
- // Compare meta data (primaryBlockMetadata)
- assert.Equal(t, originalMemPart.meta.Buf, newMemPart.meta.Buf)
-
- // Compare tag families
- assert.Equal(t, len(originalMemPart.tagFamilies), len(newMemPart.tagFamilies))
- for name, originalBuffer := range originalMemPart.tagFamilies {
- newBuffer, exists := newMemPart.tagFamilies[name]
- assert.True(t, exists, "Tag family %s not found in new memPart", name)
- assert.Equal(t, originalBuffer.Buf, newBuffer.Buf)
- }
-
- // Compare tag family metadata
- assert.Equal(t, len(originalMemPart.tagFamilyMetadata), len(newMemPart.tagFamilyMetadata))
- for name, originalBuffer := range originalMemPart.tagFamilyMetadata {
- newBuffer, exists := newMemPart.tagFamilyMetadata[name]
- assert.True(t, exists, "Tag family metadata %s not found in new memPart", name)
- assert.Equal(t, originalBuffer.Buf, newBuffer.Buf)
- }
-
- // Compare tag family filters
- assert.Equal(t, len(originalMemPart.tagFamilyFilter), len(newMemPart.tagFamilyFilter))
- for name, originalBuffer := range originalMemPart.tagFamilyFilter {
- newBuffer, exists := newMemPart.tagFamilyFilter[name]
- assert.True(t, exists, "Tag family filter %s not found in new memPart", name)
- assert.Equal(t, originalBuffer.Buf, newBuffer.Buf)
- }
-}
-
-func Test_memPart_Marshal_Unmarshal(t *testing.T) {
- // Create a test memPart with some data
- mp := generateMemPart()
- defer releaseMemPart(mp)
-
- // Initialize with test elements
- esData := &elements{
- timestamps: []int64{1, 2, 3},
- elementIDs: []uint64{100, 200, 300},
- seriesIDs: []common.SeriesID{1, 1, 2},
- tagFamilies: [][]tagValues{
- {
- {
- tag: "service",
- values: []*tagValue{
- {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("test-service")},
- },
- },
- },
- {
- {
- tag: "service",
- values: []*tagValue{
- {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("test-service-2")},
- },
- },
- },
- {
- {
- tag: "instance",
- values: []*tagValue{
- {tag: "name", valueType: pbv1.ValueTypeStr, value: []byte("test-instance")},
- },
- },
- },
- },
- }
-
- mp.mustInitFromElements(esData)
-
- // Test marshal
- marshaled, err := mp.Marshal()
- require.NoError(t, err)
- require.NotEmpty(t, marshaled)
-
- // Test unmarshal
- mp2 := generateMemPart()
- defer releaseMemPart(mp2)
-
- err = mp2.Unmarshal(marshaled)
- require.NoError(t, err)
-
- // Verify the unmarshaled data matches the original
- require.Equal(t, mp.partMetadata.ID, mp2.partMetadata.ID)
- require.Equal(t, mp.partMetadata.TotalCount, mp2.partMetadata.TotalCount)
- require.Equal(t, mp.partMetadata.BlocksCount, mp2.partMetadata.BlocksCount)
- require.Equal(t, mp.partMetadata.MinTimestamp, mp2.partMetadata.MinTimestamp)
- require.Equal(t, mp.partMetadata.MaxTimestamp, mp2.partMetadata.MaxTimestamp)
-
- // Verify buffers
- require.Equal(t, mp.meta.Buf, mp2.meta.Buf)
- require.Equal(t, mp.primary.Buf, mp2.primary.Buf)
- require.Equal(t, mp.timestamps.Buf, mp2.timestamps.Buf)
-
- // Verify tag families
- require.Equal(t, len(mp.tagFamilies), len(mp2.tagFamilies))
- for name, tf := range mp.tagFamilies {
- tf2, exists := mp2.tagFamilies[name]
- require.True(t, exists)
- require.Equal(t, tf.Buf, tf2.Buf)
- }
-
- // Verify tag family metadata
- require.Equal(t, len(mp.tagFamilyMetadata), len(mp2.tagFamilyMetadata))
- for name, tfh := range mp.tagFamilyMetadata {
- tfh2, exists := mp2.tagFamilyMetadata[name]
- require.True(t, exists)
- require.Equal(t, tfh.Buf, tfh2.Buf)
- }
-
- // Verify tag family filter
- require.Equal(t, len(mp.tagFamilyFilter), len(mp2.tagFamilyFilter))
- for name, tff := range mp.tagFamilyFilter {
- tff2, exists := mp2.tagFamilyFilter[name]
- require.True(t, exists)
- require.Equal(t, tff.Buf, tff2.Buf)
- }
-}
-
-func Test_memPart_Marshal_Unmarshal_Empty(t *testing.T) {
- // Test with empty memPart
- mp := generateMemPart()
- defer releaseMemPart(mp)
-
- // Test marshal
- marshaled, err := mp.Marshal()
- require.NoError(t, err)
- require.NotEmpty(t, marshaled)
-
- // Test unmarshal
- mp2 := generateMemPart()
- defer releaseMemPart(mp2)
-
- err = mp2.Unmarshal(marshaled)
- require.NoError(t, err)
-
- // Verify the unmarshaled data matches the original
- // partMetadata.ID is not serialized, so always expect 0 after unmarshal
- require.Equal(t, uint64(0), mp2.partMetadata.ID)
- require.Equal(t, mp.partMetadata.TotalCount, mp2.partMetadata.TotalCount)
- require.Equal(t, mp.partMetadata.BlocksCount, mp2.partMetadata.BlocksCount)
- require.Equal(t, mp.partMetadata.MinTimestamp, mp2.partMetadata.MinTimestamp)
- require.Equal(t, mp.partMetadata.MaxTimestamp, mp2.partMetadata.MaxTimestamp)
-
- // Verify buffers are empty
- require.Empty(t, mp2.meta.Buf)
- require.Empty(t, mp2.primary.Buf)
- require.Empty(t, mp2.timestamps.Buf)
- require.Empty(t, mp2.tagFamilies)
- require.Empty(t, mp2.tagFamilyMetadata)
- require.Empty(t, mp2.tagFamilyFilter)
-}
-
-func Test_memPart_Marshal_Unmarshal_Simple(t *testing.T) {
- // Create a test memPart manually without using mustInitFromElements
- mp := generateMemPart()
- defer releaseMemPart(mp)
-
- // Set some basic metadata
- mp.partMetadata.ID = 123
- mp.partMetadata.TotalCount = 10
- mp.partMetadata.BlocksCount = 2
- mp.partMetadata.MinTimestamp = 1000
- mp.partMetadata.MaxTimestamp = 2000
-
- // Set some buffer data
- mp.meta.Buf = []byte("test meta data")
- mp.primary.Buf = []byte("test primary data")
- mp.timestamps.Buf = []byte("test timestamps data")
-
- // Create some tag families
- mp.tagFamilies = make(map[string]*bytes.Buffer)
- mp.tagFamilies["service"] = &bytes.Buffer{Buf: []byte("service data")}
- mp.tagFamilies["instance"] = &bytes.Buffer{Buf: []byte("instance data")}
-
- // Create some tag family metadata
- mp.tagFamilyMetadata = make(map[string]*bytes.Buffer)
- mp.tagFamilyMetadata["service"] = &bytes.Buffer{Buf: []byte("service metadata")}
- mp.tagFamilyMetadata["instance"] = &bytes.Buffer{Buf: []byte("instance metadata")}
-
- // Create some tag family filters
- mp.tagFamilyFilter = make(map[string]*bytes.Buffer)
- mp.tagFamilyFilter["service"] = &bytes.Buffer{Buf: []byte("service filter")}
- mp.tagFamilyFilter["instance"] = &bytes.Buffer{Buf: []byte("instance filter")}
-
- // Test marshal
- marshaled, err := mp.Marshal()
- require.NoError(t, err)
- require.NotEmpty(t, marshaled)
-
- // Test unmarshal
- mp2 := generateMemPart()
- defer releaseMemPart(mp2)
-
- err = mp2.Unmarshal(marshaled)
- require.NoError(t, err)
-
- // Verify the unmarshaled data matches the original
- require.NotEqual(t, mp.partMetadata.ID, mp2.partMetadata.ID)
- require.Equal(t, mp.partMetadata.TotalCount, mp2.partMetadata.TotalCount)
- require.Equal(t, mp.partMetadata.BlocksCount, mp2.partMetadata.BlocksCount)
- require.Equal(t, mp.partMetadata.MinTimestamp, mp2.partMetadata.MinTimestamp)
- require.Equal(t, mp.partMetadata.MaxTimestamp, mp2.partMetadata.MaxTimestamp)
-
- // Verify buffers
- require.Equal(t, mp.meta.Buf, mp2.meta.Buf)
- require.Equal(t, mp.primary.Buf, mp2.primary.Buf)
- require.Equal(t, mp.timestamps.Buf, mp2.timestamps.Buf)
-
- // Verify tag families
- require.Equal(t, len(mp.tagFamilies), len(mp2.tagFamilies))
- for name, tf := range mp.tagFamilies {
- tf2, exists := mp2.tagFamilies[name]
- require.True(t, exists)
- require.Equal(t, tf.Buf, tf2.Buf)
- }
-
- // Verify tag family metadata
- require.Equal(t, len(mp.tagFamilyMetadata), len(mp2.tagFamilyMetadata))
- for name, tfh := range mp.tagFamilyMetadata {
- tfh2, exists := mp2.tagFamilyMetadata[name]
- require.True(t, exists)
- require.Equal(t, tfh.Buf, tfh2.Buf)
- }
-
- // Verify tag family filter
- require.Equal(t, len(mp.tagFamilyFilter), len(mp2.tagFamilyFilter))
- for name, tff := range mp.tagFamilyFilter {
- tff2, exists := mp2.tagFamilyFilter[name]
- require.True(t, exists)
- require.Equal(t, tff.Buf, tff2.Buf)
- }
-}
-
-func Test_memPart_Marshal_Unmarshal_Minimal(t *testing.T) {
- // Create a minimal test memPart
- mp := generateMemPart()
- defer releaseMemPart(mp)
-
- // Set only basic metadata
- mp.partMetadata.ID = 123
- mp.partMetadata.TotalCount = 10
-
- // Set only basic buffer data
- mp.meta.Buf = []byte("test")
- mp.primary.Buf = []byte("test")
- mp.timestamps.Buf = []byte("test")
-
- // No tag families, metadata, or filters
-
- // Test marshal
- marshaled, err := mp.Marshal()
- require.NoError(t, err)
- require.NotEmpty(t, marshaled)
-
- // Debug: Check what was marshaled
- t.Logf("Marshaled data length: %d", len(marshaled))
- t.Logf("Original partMetadata ID: %d", mp.partMetadata.ID)
-
- // Test unmarshal
- mp2 := generateMemPart()
- defer releaseMemPart(mp2)
-
- err = mp2.Unmarshal(marshaled)
- require.NoError(t, err)
-
- // Debug: Check what was unmarshaled
- t.Logf("Unmarshaled partMetadata ID: %d", mp2.partMetadata.ID)
-
- // Verify the unmarshaled data matches the original
- // partMetadata.ID is not serialized, so always expect 0 after unmarshal
- require.Equal(t, uint64(0), mp2.partMetadata.ID)
- require.Equal(t, mp.partMetadata.TotalCount, mp2.partMetadata.TotalCount)
-
- // Verify buffers
- require.Equal(t, mp.meta.Buf, mp2.meta.Buf)
- require.Equal(t, mp.primary.Buf, mp2.primary.Buf)
- require.Equal(t, mp.timestamps.Buf, mp2.timestamps.Buf)
-
- // Verify maps are empty (not nil, but empty)
- require.Empty(t, mp2.tagFamilies)
- require.Empty(t, mp2.tagFamilyMetadata)
- require.Empty(t, mp2.tagFamilyFilter)
-}
diff --git a/pkg/test/setup/setup.go b/pkg/test/setup/setup.go
index c303bafb..35c3254c 100644
--- a/pkg/test/setup/setup.go
+++ b/pkg/test/setup/setup.go
@@ -325,6 +325,8 @@ func LiaisonNodeWithHTTP(etcdEndpoint string, flags ...string) (string, string,
"--node-host", nodeHost,
"--stream-root-path="+path,
"--measure-root-path="+path,
+ "--stream-flush-timeout=0s",
+ "--measure-flush-timeout=0s",
)
closeFn := CMD(flags...)
gomega.Eventually(helpers.HTTPHealthCheck(httpAddr, ""), testflags.EventuallyTimeout).Should(gomega.Succeed())