This is an automated email from the ASF dual-hosted git repository.
ButterBright pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 152fa41f8 Fix block metadata reset before unmarshal for stream and
trace (#1195)
152fa41f8 is described below
commit 152fa41f89aca1a97c73e692587f37d6ee96d68e
Author: Huang Youliang <[email protected]>
AuthorDate: Wed Jul 1 20:49:58 2026 +0800
Fix block metadata reset before unmarshal for stream and trace (#1195)
* Fix block metadata reset before unmarshal
---
CHANGES.md | 1 +
banyand/measure/block_metadata_test.go | 32 +++++++++++++++++++++++++++++++
banyand/stream/block_metadata.go | 2 ++
banyand/stream/block_metadata_test.go | 35 ++++++++++++++++++++++++++++++++++
banyand/stream/block_scanner_test.go | 1 +
banyand/stream/part_iter_test.go | 2 ++
banyand/stream/query_by_idx_test.go | 1 +
banyand/stream/tstable_test.go | 1 +
banyand/trace/block_metadata.go | 2 ++
banyand/trace/block_metadata_test.go | 34 +++++++++++++++++++++++++++++++++
10 files changed, 111 insertions(+)
diff --git a/CHANGES.md b/CHANGES.md
index 6182ca9df..fb123ea32 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -124,6 +124,7 @@ Release Notes.
- Fix lifecycle row-replay OOM on large measure parts by streaming the dump
reader, pooling size-classed marshal buffers, and bounding in-flight batch
bytes (default 32 MiB); peak heap drops ~1.5 GB→~296 MB.
- Consolidate lifecycle migration report errors into a single flat list of
structured, stage-aware entries.
- Fix backup container OOM from overlapping scheduled runs; serialize runs and
upload small snapshot files concurrently.
+- Fix block metadata reset before unmarshal for stream and trace.
### Document
diff --git a/banyand/measure/block_metadata_test.go
b/banyand/measure/block_metadata_test.go
index 3ea72f491..dbb514113 100644
--- a/banyand/measure/block_metadata_test.go
+++ b/banyand/measure/block_metadata_test.go
@@ -376,6 +376,38 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
require.Equal(t, original, unmarshaled)
})
+ t.Run("reuse destination clears stale tag families", func(t *testing.T)
{
+ original := []blockMetadata{
+ {
+ seriesID: common.SeriesID(1),
+ tagFamilies: make(map[string]*dataBlock),
+ timestamps: timestampsMetadata{
+ dataBlock: dataBlock{
+ offset: 1,
+ size: 1,
+ },
+ min: 1,
+ max: 1,
+ encodeType: encoding.EncodeTypeConst,
+ },
+ },
+ }
+
+ var marshaled []byte
+ for _, bm := range original {
+ marshaled = bm.marshal(marshaled)
+ }
+
+ unmarshaled, err := unmarshalBlockMetadata(nil, marshaled)
+ require.NoError(t, err)
+ require.Len(t, unmarshaled, 1)
+
+ unmarshaled[0].tagFamilies = map[string]*dataBlock{"stale":
{offset: 99, size: 99}}
+ reused, reuseErr := unmarshalBlockMetadata(unmarshaled[:0],
marshaled)
+ require.NoError(t, reuseErr)
+ require.Equal(t, original, reused)
+ })
+
t.Run("unmarshal invalid blockMetadata", func(t *testing.T) {
original := []blockMetadata{
{
diff --git a/banyand/stream/block_metadata.go b/banyand/stream/block_metadata.go
index db44348f7..a8c8c3a9a 100644
--- a/banyand/stream/block_metadata.go
+++ b/banyand/stream/block_metadata.go
@@ -310,6 +310,7 @@ func unmarshalBlockMetadata(dst []blockMetadata, src
[]byte) ([]blockMetadata, e
dst = append(dst, blockMetadata{})
}
bm := &dst[len(dst)-1]
+ bm.reset()
tail, err := bm.unmarshal(src)
if err != nil {
return dstOrig, fmt.Errorf("cannot unmarshal
blockMetadata entries: %w", err)
@@ -391,6 +392,7 @@ func unmarshalBlockMetadataFiltered(dst []blockMetadata,
src []byte, wanted []co
dst = append(dst, blockMetadata{})
}
bm := &dst[len(dst)-1]
+ bm.reset()
tail, uErr := bm.unmarshal(src)
if uErr != nil {
return dstOrig, fmt.Errorf("cannot unmarshal
blockMetadata entries: %w", uErr)
diff --git a/banyand/stream/block_metadata_test.go
b/banyand/stream/block_metadata_test.go
index 64831fdfb..bb22c35d5 100644
--- a/banyand/stream/block_metadata_test.go
+++ b/banyand/stream/block_metadata_test.go
@@ -246,6 +246,7 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
original := []blockMetadata{
{
seriesID: common.SeriesID(1),
+ tagType: make(tagType),
timestamps: timestampsMetadata{
dataBlock: dataBlock{
offset: 1,
@@ -258,6 +259,7 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
},
{
seriesID: common.SeriesID(2),
+ tagType: make(tagType),
timestamps: timestampsMetadata{
dataBlock: dataBlock{
offset: 2,
@@ -280,6 +282,39 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
require.Equal(t, original, unmarshaled)
})
+ t.Run("reuse destination clears stale tag families", func(t *testing.T)
{
+ original := []blockMetadata{
+ {
+ seriesID: common.SeriesID(1),
+ tagFamilies: make(map[string]*dataBlock),
+ tagType: make(tagType),
+ timestamps: timestampsMetadata{
+ dataBlock: dataBlock{
+ offset: 1,
+ size: 1,
+ },
+ min: 1,
+ max: 1,
+ encodeType: encoding.EncodeTypeConst,
+ },
+ },
+ }
+
+ var marshaled []byte
+ for _, bm := range original {
+ marshaled = bm.marshal(marshaled)
+ }
+
+ unmarshaled, err := unmarshalBlockMetadata(nil, marshaled)
+ require.NoError(t, err)
+ require.Len(t, unmarshaled, 1)
+
+ unmarshaled[0].tagFamilies = map[string]*dataBlock{"stale":
{offset: 99, size: 99}}
+ reused, reuseErr := unmarshalBlockMetadata(unmarshaled[:0],
marshaled)
+ require.NoError(t, reuseErr)
+ require.Equal(t, original, reused)
+ })
+
t.Run("unmarshal invalid blockMetadata", func(t *testing.T) {
original := []blockMetadata{
{
diff --git a/banyand/stream/block_scanner_test.go
b/banyand/stream/block_scanner_test.go
index ac592b9d7..25f9e490b 100644
--- a/banyand/stream/block_scanner_test.go
+++ b/banyand/stream/block_scanner_test.go
@@ -187,6 +187,7 @@ func TestBlockScanner_QuotaExceeded(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{},
"timestamps"),
cmpopts.IgnoreFields(blockMetadata{},
"elementIDs"),
cmpopts.IgnoreFields(blockMetadata{},
"tagFamilies"),
+ cmpopts.IgnoreFields(blockMetadata{},
"tagType"),
cmp.AllowUnexported(blockMetadata{}),
); diff != "" {
t.Errorf("Unexpected blockMetadata (-got
+want):\n%s", diff)
diff --git a/banyand/stream/part_iter_test.go b/banyand/stream/part_iter_test.go
index 6eb63da01..dc8c59e25 100644
--- a/banyand/stream/part_iter_test.go
+++ b/banyand/stream/part_iter_test.go
@@ -136,6 +136,7 @@ func Test_partIter_nextBlock(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{},
"timestamps"),
cmpopts.IgnoreFields(blockMetadata{},
"elementIDs"),
cmpopts.IgnoreFields(blockMetadata{},
"tagFamilies"),
+ cmpopts.IgnoreFields(blockMetadata{},
"tagType"),
cmp.AllowUnexported(blockMetadata{}),
); diff != "" {
t.Errorf("Unexpected blockMetadata
(-got +want):\n%s", diff)
@@ -199,6 +200,7 @@ func Test_partMergeIter_nextBlock(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{},
"timestamps"),
cmpopts.IgnoreFields(blockMetadata{},
"elementIDs"),
cmpopts.IgnoreFields(blockMetadata{},
"tagFamilies"),
+ cmpopts.IgnoreFields(blockMetadata{},
"tagType"),
cmp.AllowUnexported(blockMetadata{}),
); diff != "" {
t.Errorf("Unexpected blockMetadata
(-got +want):\n%s", diff)
diff --git a/banyand/stream/query_by_idx_test.go
b/banyand/stream/query_by_idx_test.go
index 4a695bd8e..05af46e7f 100644
--- a/banyand/stream/query_by_idx_test.go
+++ b/banyand/stream/query_by_idx_test.go
@@ -127,6 +127,7 @@ func TestQueryResult_QuotaExceeded(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{},
"timestamps"),
cmpopts.IgnoreFields(blockMetadata{},
"elementIDs"),
cmpopts.IgnoreFields(blockMetadata{},
"tagFamilies"),
+ cmpopts.IgnoreFields(blockMetadata{},
"tagType"),
cmp.AllowUnexported(blockMetadata{}),
); diff != "" {
t.Errorf("Unexpected blockMetadata (-got
+want):\n%s", diff)
diff --git a/banyand/stream/tstable_test.go b/banyand/stream/tstable_test.go
index ffbd31e35..7ce2dffd8 100644
--- a/banyand/stream/tstable_test.go
+++ b/banyand/stream/tstable_test.go
@@ -167,6 +167,7 @@ func Test_tstIter(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{}, "timestamps"),
cmpopts.IgnoreFields(blockMetadata{}, "elementIDs"),
cmpopts.IgnoreFields(blockMetadata{}, "tagFamilies"),
+ cmpopts.IgnoreFields(blockMetadata{}, "tagType"),
cmp.AllowUnexported(blockMetadata{}),
); diff != "" {
t.Errorf("Unexpected blockMetadata (-got +want):\n%s",
diff)
diff --git a/banyand/trace/block_metadata.go b/banyand/trace/block_metadata.go
index 09f884d62..ad9660717 100644
--- a/banyand/trace/block_metadata.go
+++ b/banyand/trace/block_metadata.go
@@ -266,6 +266,7 @@ func unmarshalBlockMetadata(dst []blockMetadata, src
[]byte, tagType map[string]
dst = append(dst, blockMetadata{})
}
bm := &dst[len(dst)-1]
+ bm.reset()
tail, err := bm.unmarshal(src, tagType)
if err != nil {
return dstOrig, fmt.Errorf("cannot unmarshal
blockMetadata entries: %w", err)
@@ -334,6 +335,7 @@ func unmarshalBlockMetadataFiltered(dst []blockMetadata,
src []byte, tagType map
dst = append(dst, blockMetadata{})
}
bm := &dst[len(dst)-1]
+ bm.reset()
tail, uErr := bm.unmarshal(src, tagType)
if uErr != nil {
return dstOrig, fmt.Errorf("cannot unmarshal
blockMetadata entries: %w", uErr)
diff --git a/banyand/trace/block_metadata_test.go
b/banyand/trace/block_metadata_test.go
index a8dc829de..d100d6e47 100644
--- a/banyand/trace/block_metadata_test.go
+++ b/banyand/trace/block_metadata_test.go
@@ -207,6 +207,7 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
wanted := []blockMetadata{
{
traceID: "trace1",
+ tags: make(map[string]*dataBlock),
tagType: make(map[string]pbv1.ValueType),
spans: &dataBlock{
offset: 10,
@@ -218,6 +219,7 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
},
{
traceID: "trace2",
+ tags: make(map[string]*dataBlock),
tagType: make(map[string]pbv1.ValueType),
spans: &dataBlock{
offset: 30,
@@ -240,6 +242,38 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
require.Equal(t, wanted, unmarshaled)
})
+ t.Run("reuse destination clears stale tags", func(t *testing.T) {
+ original := []blockMetadata{
+ {
+ traceID: "trace1",
+ tags: make(map[string]*dataBlock),
+ tagType: make(map[string]pbv1.ValueType),
+ spans: &dataBlock{
+ offset: 10,
+ size: 20,
+ },
+ timestamps: timestampsMetadata{},
+ uncompressedSpanSizeBytes: 100,
+ count: 1,
+ },
+ }
+
+ var marshaled []byte
+ for _, bm := range original {
+ marshaled = bm.marshal(marshaled)
+ }
+
+ tagType := make(map[string]pbv1.ValueType)
+ unmarshaled, err := unmarshalBlockMetadata(nil, marshaled,
tagType)
+ require.NoError(t, err)
+ require.Equal(t, original, unmarshaled)
+
+ unmarshaled[0].tags["stale"] = &dataBlock{offset: 99, size: 99}
+ reused, reuseErr := unmarshalBlockMetadata(unmarshaled[:0],
marshaled, tagType)
+ require.NoError(t, reuseErr)
+ require.Equal(t, original, reused)
+ })
+
t.Run("unmarshal invalid blockMetadata", func(t *testing.T) {
original := []blockMetadata{
{
