This is an automated email from the ASF dual-hosted git repository.
hanahmily 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 4c674abc4 feat: implement global tag type metadata for measure parts
(#1196)
4c674abc4 is described below
commit 4c674abc429e4350f236fc2762d907c02704d521
Author: Huang Youliang <[email protected]>
AuthorDate: Thu Jul 2 08:59:36 2026 +0800
feat: implement global tag type metadata for measure parts (#1196)
---
banyand/measure/block.go | 9 ++++
banyand/measure/block_metadata.go | 16 ++++++
banyand/measure/block_metadata_test.go | 3 ++
banyand/measure/block_reader_test.go | 1 +
banyand/measure/block_test.go | 14 +++++
banyand/measure/block_writer.go | 11 +++-
banyand/measure/file_part_sync_test.go | 30 ++++++++++-
banyand/measure/merger.go | 14 ++---
banyand/measure/merger_test.go | 40 +++++++++-----
banyand/measure/part.go | 23 +++++++-
banyand/measure/part_iter_test.go | 2 +
banyand/measure/part_metadata.go | 95 ++++++++++++++++++++++++++++++++++
banyand/measure/part_test.go | 6 +++
banyand/measure/syncer.go | 19 +++++++
banyand/measure/tstable_test.go | 1 +
banyand/measure/write_data.go | 20 ++++---
banyand/stream/merger.go | 2 +-
banyand/trace/merger.go | 6 +--
18 files changed, 280 insertions(+), 32 deletions(-)
diff --git a/banyand/measure/block.go b/banyand/measure/block.go
index 79e927446..cb88287c3 100644
--- a/banyand/measure/block.go
+++ b/banyand/measure/block.go
@@ -187,10 +187,19 @@ func (b *block) validate() {
func (b *block) marshalTagFamily(tf columnFamily, bm *blockMetadata, ww
*writers) {
hw, w := ww.getColumnMetadataWriterAndColumnWriter(tf.name)
cc := tf.columns
+ if bm.tagType == nil {
+ bm.tagType = make(tagType)
+ }
+ tagTypes := bm.tagType[tf.name]
+ if tagTypes == nil {
+ tagTypes = make(map[string]pbv1.ValueType, len(cc))
+ bm.tagType[tf.name] = tagTypes
+ }
cfm := generateColumnFamilyMetadata()
cmm := cfm.resizeColumnMetadata(len(cc))
for i := range cc {
cc[i].mustWriteTo(&cmm[i], w)
+ tagTypes[cc[i].name] = cc[i].valueType
}
bb := bigValuePool.Generate()
defer bigValuePool.Release(bb)
diff --git a/banyand/measure/block_metadata.go
b/banyand/measure/block_metadata.go
index 65253dcd0..75a042cf8 100644
--- a/banyand/measure/block_metadata.go
+++ b/banyand/measure/block_metadata.go
@@ -62,6 +62,7 @@ func (b *dataBlock) unmarshal(src []byte) []byte {
type blockMetadata struct {
tagFamilies map[string]*dataBlock
+ tagType tagType
field columnFamilyMetadata
tagProjection []model.TagProjection
timestamps timestampsMetadata
@@ -75,6 +76,16 @@ func (bm *blockMetadata) copyFrom(src *blockMetadata) {
bm.uncompressedSizeBytes = src.uncompressedSizeBytes
bm.count = src.count
bm.timestamps.copyFrom(&src.timestamps)
+ if src.tagType == nil {
+ bm.tagType = nil
+ } else {
+ if bm.tagType == nil {
+ bm.tagType = make(tagType, len(src.tagType))
+ } else {
+ bm.tagType.reset()
+ }
+ bm.tagType.copyFrom(src.tagType)
+ }
for k, db := range src.tagFamilies {
if bm.tagFamilies == nil {
bm.tagFamilies = make(map[string]*dataBlock)
@@ -103,6 +114,11 @@ func (bm *blockMetadata) reset() {
bm.count = 0
bm.timestamps.reset()
bm.field.reset()
+ if bm.tagType == nil {
+ bm.tagType = make(tagType)
+ } else {
+ bm.tagType.reset()
+ }
for k := range bm.tagFamilies {
bm.tagFamilies[k].reset()
delete(bm.tagFamilies, k)
diff --git a/banyand/measure/block_metadata_test.go
b/banyand/measure/block_metadata_test.go
index dbb514113..9f32c935e 100644
--- a/banyand/measure/block_metadata_test.go
+++ b/banyand/measure/block_metadata_test.go
@@ -342,6 +342,7 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
original := []blockMetadata{
{
seriesID: common.SeriesID(1),
+ tagType: make(tagType),
timestamps: timestampsMetadata{
dataBlock: dataBlock{
offset: 1,
@@ -354,6 +355,7 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
},
{
seriesID: common.SeriesID(2),
+ tagType: make(tagType),
timestamps: timestampsMetadata{
dataBlock: dataBlock{
offset: 2,
@@ -381,6 +383,7 @@ func Test_unmarshalBlockMetadata(t *testing.T) {
{
seriesID: common.SeriesID(1),
tagFamilies: make(map[string]*dataBlock),
+ tagType: make(tagType),
timestamps: timestampsMetadata{
dataBlock: dataBlock{
offset: 1,
diff --git a/banyand/measure/block_reader_test.go
b/banyand/measure/block_reader_test.go
index 4432aa047..8e05e3cbc 100644
--- a/banyand/measure/block_reader_test.go
+++ b/banyand/measure/block_reader_test.go
@@ -117,6 +117,7 @@ func Test_blockReader_nextBlock(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{},
"timestamps"),
cmpopts.IgnoreFields(blockMetadata{},
"field"),
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/measure/block_test.go b/banyand/measure/block_test.go
index ef5630caa..9414954c7 100644
--- a/banyand/measure/block_test.go
+++ b/banyand/measure/block_test.go
@@ -22,6 +22,7 @@ import (
"testing"
"github.com/google/go-cmp/cmp"
+ "github.com/stretchr/testify/require"
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/pkg/bytes"
@@ -383,6 +384,19 @@ func Test_marshalAndUnmarshalBlock(t *testing.T) {
bm := blockMetadata{}
b.mustWriteTo(sid, &bm, ww)
+ require.Equal(t, tagType{
+ "arrTag": {
+ "strArrTag": pbv1.ValueTypeStrArr,
+ "intArrTag": pbv1.ValueTypeInt64Arr,
+ },
+ "binaryTag": {
+ "binaryTag": pbv1.ValueTypeBinaryData,
+ },
+ "singleTag": {
+ "strTag": pbv1.ValueTypeStr,
+ "intTag": pbv1.ValueTypeInt64,
+ },
+ }, bm.tagType)
tagFamilyMetadataReaders := make(map[string]fs.Reader)
tagFamilyReaders := make(map[string]fs.Reader)
diff --git a/banyand/measure/block_writer.go b/banyand/measure/block_writer.go
index a6238b3b0..b441ad7d0 100644
--- a/banyand/measure/block_writer.go
+++ b/banyand/measure/block_writer.go
@@ -144,6 +144,7 @@ func (sw *writers)
getColumnMetadataWriterAndColumnWriter(columnName string) (*w
type blockWriter struct {
writers writers
+ tagType tagType
metaData []byte
primaryBlockData []byte
primaryBlockMetadata primaryBlockMetadata
@@ -162,6 +163,11 @@ type blockWriter struct {
func (bw *blockWriter) reset() {
bw.writers.reset()
+ if bw.tagType == nil {
+ bw.tagType = make(tagType)
+ } else {
+ bw.tagType.reset()
+ }
bw.sidLast = 0
bw.sidFirst = 0
bw.minTimestampLast = 0
@@ -220,6 +226,7 @@ func (bw *blockWriter) mustWriteBlock(sid common.SeriesID,
b *block) {
bm := generateBlockMetadata()
b.mustWriteTo(sid, bm, &bw.writers)
+ bw.tagType.copyFrom(bm.tagType)
tm := &bm.timestamps
if bw.totalCount == 0 || tm.min < bw.totalMinTimestamp {
bw.totalMinTimestamp = tm.min
@@ -261,7 +268,7 @@ func (bw *blockWriter) mustFlushPrimaryBlock(data []byte) {
bw.sidFirst = 0
}
-func (bw *blockWriter) Flush(pm *partMetadata) {
+func (bw *blockWriter) Flush(pm *partMetadata, tt *tagType) {
pm.UncompressedSizeBytes = bw.totalUncompressedSizeBytes
pm.TotalCount = bw.totalCount
pm.BlocksCount = bw.totalBlocksCount
@@ -276,6 +283,7 @@ func (bw *blockWriter) Flush(pm *partMetadata) {
bigValuePool.Release(bb)
pm.CompressedSizeBytes = bw.writers.totalBytesWritten()
+ tt.copyFrom(bw.tagType)
bw.writers.MustClose()
bw.reset()
@@ -285,6 +293,7 @@ func generateBlockWriter() *blockWriter {
v := blockWriterPool.Get()
if v == nil {
return &blockWriter{
+ tagType: make(tagType),
writers: writers{
tagFamilyMetadataWriters:
make(map[string]*writer),
tagFamilyWriters:
make(map[string]*writer),
diff --git a/banyand/measure/file_part_sync_test.go
b/banyand/measure/file_part_sync_test.go
index eabe55627..ef284e153 100644
--- a/banyand/measure/file_part_sync_test.go
+++ b/banyand/measure/file_part_sync_test.go
@@ -26,6 +26,7 @@ import (
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -106,8 +107,27 @@ func TestMustAddFilePart_FilesOnDisk(t *testing.T) {
partID := uint64(1)
destPath := partPath(tabDir, partID)
buildAndFlushMeasureMemPart(t, fileSystem, dps, destPath)
-
- tst.mustAddFilePart(partID)
+ var metadata partMetadata
+ metadata.mustReadMetadata(fileSystem, destPath)
+ partCtx := &syncPartContext{
+ tsTable: tst,
+ fileSystem: fileSystem,
+ partPath: destPath,
+ partMeta: metadata,
+ partID: partID,
+ }
+ chunkCtx := &queue.ChunkedSyncPartContext{Handler: partCtx, FileName:
tagTypeFilename}
+ wantTagType := tagType{"singleTag": {"strTag":
pbv1.ValueTypeStr}}.marshal()
+ callback := &syncCallback{l: logger.GetLogger("test")}
+ require.NoError(t, callback.HandleFileChunk(chunkCtx,
wantTagType[:len(wantTagType)/2]))
+ require.NoError(t, callback.HandleFileChunk(chunkCtx,
wantTagType[len(wantTagType)/2:]))
+ require.Equal(t, wantTagType, partCtx.tagTypeBuffer)
+ require.NoError(t, partCtx.FinishSync())
+ require.Nil(t, partCtx.tagTypeBuffer)
+
+ gotTagType, readErr := fileSystem.Read(filepath.Join(destPath,
tagTypeFilename))
+ require.NoError(t, readErr)
+ require.Equal(t, wantTagType, gotTagType)
snp := tst.currentSnapshot()
require.NotNil(t, snp, "snapshot must not be nil after mustAddFilePart")
@@ -190,11 +210,17 @@ func TestFilePart_ErrorCleanup(t *testing.T) {
spc.writers = w
spc.writers.metaWriter.MustWrite([]byte("partial"))
+ tagTypeData := tagType{"singleTag": {"strTag":
pbv1.ValueTypeStr}}.marshal()
+ callback := &syncCallback{l: logger.GetLogger("test")}
+ chunkCtx := &queue.ChunkedSyncPartContext{Handler: spc, FileName:
tagTypeFilename}
+ require.NoError(t, callback.HandleFileChunk(chunkCtx, tagTypeData))
+ require.Equal(t, tagTypeData, spc.tagTypeBuffer)
_, statErr := os.Stat(destPath)
require.NoError(t, statErr, "part directory must exist before Close")
require.NoError(t, spc.Close())
+ require.Nil(t, spc.tagTypeBuffer)
_, statErr = os.Stat(destPath)
require.True(t, os.IsNotExist(statErr), "incomplete part directory must
be removed after Close without FinishSync")
diff --git a/banyand/measure/merger.go b/banyand/measure/merger.go
index 067c26c80..26251f4e6 100644
--- a/banyand/measure/merger.go
+++ b/banyand/measure/merger.go
@@ -251,7 +251,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
bw := generateBlockWriter()
bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
- pm, err := mergeBlocks(closeCh, bw, br)
+ pm, mergedTagType, err := mergeBlocks(closeCh, bw, br)
releaseBlockWriter(bw)
releaseBlockReader(br)
for i := range pii {
@@ -260,6 +260,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
if err != nil {
return nil, err
}
+ mergedTagType.mustWriteTagType(fileSystem, dstPath)
pm.mustWriteMetadata(fileSystem, dstPath)
// No SyncPath: mustWriteMetadata goes through WriteAtomic which already
// fsyncs the parent directory after rename.
@@ -269,7 +270,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
var errClosed = fmt.Errorf("the merger is closed")
-func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader)
(*partMetadata, error) {
+func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader)
(*partMetadata, *tagType, error) {
pendingBlockIsEmpty := true
pendingBlock := generateBlockPointer()
defer releaseBlockPointer(pendingBlock)
@@ -290,7 +291,7 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
for br.nextBlockMetadata() {
select {
case <-closeCh:
- return nil, errClosed
+ return nil, nil, errClosed
default:
}
b := br.block
@@ -348,15 +349,16 @@ func mergeBlocks(closeCh <-chan struct{}, bw
*blockWriter, br *blockReader) (*pa
releaseDecoder()
}
if err := br.error(); err != nil {
- return nil, fmt.Errorf("cannot read block to merge: %w", err)
+ return nil, nil, fmt.Errorf("cannot read block to merge: %w",
err)
}
if !pendingBlockIsEmpty {
bw.mustWriteBlock(pendingBlock.bm.seriesID, &pendingBlock.block)
}
releaseDecoder()
var result partMetadata
- bw.Flush(&result)
- return &result, nil
+ mergedTagType := make(tagType)
+ bw.Flush(&result, &mergedTagType)
+ return &result, &mergedTagType, nil
}
func mergeTwoBlocks(target, left, right *blockPointer) {
diff --git a/banyand/measure/merger_test.go b/banyand/measure/merger_test.go
index 37b1d7fc7..39d92142c 100644
--- a/banyand/measure/merger_test.go
+++ b/banyand/measure/merger_test.go
@@ -19,6 +19,7 @@ package measure
import (
"errors"
+ "path/filepath"
"reflect"
"testing"
"time"
@@ -972,10 +973,11 @@ var mergedTopNBlock4 = block{
func Test_mergeParts(t *testing.T) {
tests := []struct {
- wantErr error
- name string
- dpsList []*dataPoints
- want []blockMetadata
+ wantErr error
+ name string
+ dpsList []*dataPoints
+ want []blockMetadata
+ wantTagType bool
}{
{
name: "Test with no data point",
@@ -983,8 +985,9 @@ func Test_mergeParts(t *testing.T) {
wantErr: errNoPartToMerge,
},
{
- name: "Test with single part",
- dpsList: []*dataPoints{dpsTS1},
+ name: "Test with single part",
+ dpsList: []*dataPoints{dpsTS1},
+ wantTagType: true,
want: []blockMetadata{
{seriesID: 1, count: 1, uncompressedSizeBytes:
1654},
{seriesID: 2, count: 1, uncompressedSizeBytes:
47},
@@ -992,8 +995,9 @@ func Test_mergeParts(t *testing.T) {
},
},
{
- name: "Test with multiple parts with different ts",
- dpsList: []*dataPoints{dpsTS1, dpsTS2, dpsTS2},
+ name: "Test with multiple parts with different
ts",
+ dpsList: []*dataPoints{dpsTS1, dpsTS2, dpsTS2},
+ wantTagType: true,
want: []blockMetadata{
{seriesID: 1, count: 2, uncompressedSizeBytes:
3245},
{seriesID: 2, count: 2, uncompressedSizeBytes:
71},
@@ -1001,8 +1005,9 @@ func Test_mergeParts(t *testing.T) {
},
},
{
- name: "Test with multiple parts with same ts",
- dpsList: []*dataPoints{dpsTS11, dpsTS1},
+ name: "Test with multiple parts with same ts",
+ dpsList: []*dataPoints{dpsTS11, dpsTS1},
+ wantTagType: true,
want: []blockMetadata{
{seriesID: 1, count: 1, uncompressedSizeBytes:
1654},
{seriesID: 2, count: 1, uncompressedSizeBytes:
47},
@@ -1010,8 +1015,9 @@ func Test_mergeParts(t *testing.T) {
},
},
{
- name: "Test with multiple parts with a large
quantity of different ts",
- dpsList: []*dataPoints{generateHugeDatapoints(1, 5000,
1), generateHugeDatapoints(5001, 10000, 2)},
+ name: "Test with multiple parts with a large
quantity of different ts",
+ dpsList: []*dataPoints{generateHugeDatapoints(1,
5000, 1), generateHugeDatapoints(5001, 10000, 2)},
+ wantTagType: true,
want: []blockMetadata{
{seriesID: 1, count: 2530,
uncompressedSizeBytes: 4025293},
{seriesID: 1, count: 2470,
uncompressedSizeBytes: 3929833},
@@ -1061,10 +1067,20 @@ func Test_mergeParts(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{},
"timestamps"),
cmpopts.IgnoreFields(blockMetadata{},
"field"),
cmpopts.IgnoreFields(blockMetadata{},
"tagFamilies"),
+ cmpopts.IgnoreFields(blockMetadata{},
"tagType"),
cmp.AllowUnexported(blockMetadata{}),
); diff != "" {
t.Errorf("Unexpected blockMetadata
(-got +want):\n%s", diff)
}
+
+ if tt.wantTagType {
+ tagTypeData, readErr :=
fileSystem.Read(filepath.Join(partPath(root, partID), tagTypeFilename))
+ require.NoError(t, readErr)
+ mergedTagType := make(tagType)
+ require.NoError(t,
mergedTagType.unmarshal(tagTypeData))
+ require.Equal(t, pbv1.ValueTypeStrArr,
mergedTagType["arrTag"]["strArrTag"])
+ require.Equal(t, pbv1.ValueTypeInt64,
mergedTagType["singleTag"]["intTag"])
+ }
}
t.Run("memory parts", func(t *testing.T) {
diff --git a/banyand/measure/part.go b/banyand/measure/part.go
index 93f5b33c5..d67488fe8 100644
--- a/banyand/measure/part.go
+++ b/banyand/measure/part.go
@@ -45,6 +45,7 @@ const (
measureSeriesMetadataName = "smeta"
metadataFilename = "metadata.json"
+ tagTypeFilename = "tag.type"
primaryFilename = measurePrimaryName + ".bin"
metaFilename = measureMetaName + ".bin"
timestampsFilename = measureTimestampsName + ".bin"
@@ -112,6 +113,7 @@ func openMemPart(mp *memPart) *part {
type memPart struct {
tagFamilyMetadata map[string]*bytes.Buffer
tagFamilies map[string]*bytes.Buffer
+ tagType tagType
meta bytes.Buffer
primary bytes.Buffer
timestamps bytes.Buffer
@@ -140,6 +142,11 @@ func (mp *memPart) mustCreateMemTagFamilyWriters(name
string) (fs.Writer, fs.Wri
func (mp *memPart) reset() {
mp.partMetadata.reset()
+ if mp.tagType == nil {
+ mp.tagType = make(tagType)
+ } else {
+ mp.tagType.reset()
+ }
mp.meta.Reset()
mp.primary.Reset()
mp.timestamps.Reset()
@@ -200,7 +207,7 @@ func (mp *memPart) mustInitFromDataPoints(dps *dataPoints) {
uncompressedBlockSizeBytes += uncompressedDataPointSizeBytes(i,
dps)
}
bsw.MustWriteDataPoints(sidPrev, dps.timestamps[indexPrev:],
dps.versions[indexPrev:], dps.tagFamilies[indexPrev:], dps.fields[indexPrev:])
- bsw.Flush(&mp.partMetadata)
+ bsw.Flush(&mp.partMetadata, &mp.tagType)
releaseBlockWriter(bsw)
}
@@ -226,6 +233,7 @@ func (mp *memPart) mustFlush(fileSystem fs.FileSystem, path
string) {
fs.MustFlushAtomic(fileSystem, mp.seriesMetadata.Buf,
filepath.Join(path, seriesMetadataFilename), storage.FilePerm)
}
+ mp.tagType.mustWriteTagType(fileSystem, path)
mp.partMetadata.mustWriteMetadata(fileSystem, path)
// No SyncPath: mustWriteMetadata goes through WriteAtomic which already
// fsyncs the parent directory after rename, covering the dirent changes
@@ -415,6 +423,19 @@ func CreatePartFileReaderFromPath(partPath string, lfs
fs.FileSystem) ([]queue.F
continue
}
+ if e.Name() == tagTypeFilename {
+ tagTypePath := path.Join(partPath, e.Name())
+ tagTypeReader, openErr := lfs.OpenFile(tagTypePath)
+ if openErr != nil {
+ logger.Panicf("cannot open tag type file %q:
%s", tagTypePath, openErr)
+ }
+ readers = append(readers, tagTypeReader)
+ files = append(files, queue.FileInfo{
+ Name: tagTypeFilename,
+ Reader: tagTypeReader.SequentialRead(),
+ })
+ }
+
// Tag family metadata files (.tfm)
if filepath.Ext(e.Name()) == tagFamiliesMetadataFilenameExt {
tfmPath := path.Join(partPath, e.Name())
diff --git a/banyand/measure/part_iter_test.go
b/banyand/measure/part_iter_test.go
index 8f4fc674b..4a50d5e09 100644
--- a/banyand/measure/part_iter_test.go
+++ b/banyand/measure/part_iter_test.go
@@ -136,6 +136,7 @@ func Test_partIter_nextBlock(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{},
"timestamps"),
cmpopts.IgnoreFields(blockMetadata{},
"field"),
cmpopts.IgnoreFields(blockMetadata{},
"tagFamilies"),
+ cmpopts.IgnoreFields(blockMetadata{},
"tagType"),
cmp.AllowUnexported(blockMetadata{}),
); diff != "" {
t.Errorf("Unexpected blockMetadata
(-got +want):\n%s", diff)
@@ -203,6 +204,7 @@ func Test_partMergeIter_nextBlock(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{},
"timestamps"),
cmpopts.IgnoreFields(blockMetadata{},
"field"),
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/measure/part_metadata.go b/banyand/measure/part_metadata.go
index e7458c07d..cf6f0c661 100644
--- a/banyand/measure/part_metadata.go
+++ b/banyand/measure/part_metadata.go
@@ -19,14 +19,19 @@ package measure
import (
"encoding/json"
+ "fmt"
"path/filepath"
+ "sort"
"github.com/pkg/errors"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)
type partMetadata struct {
@@ -107,6 +112,96 @@ func (pm *partMetadata) mustWriteMetadata(fileSystem
fs.FileSystem, partPath str
}
}
+type tagType map[string]map[string]pbv1.ValueType
+
+func (tt tagType) reset() {
+ clear(tt)
+}
+
+func (tt tagType) copyFrom(source tagType) {
+ for familyName, sourceTags := range source {
+ tags := tt[familyName]
+ if tags == nil {
+ tags = make(map[string]pbv1.ValueType, len(sourceTags))
+ tt[familyName] = tags
+ }
+ for tagName, valueType := range sourceTags {
+ tags[tagName] = valueType
+ }
+ }
+}
+
+func (tt tagType) marshal() []byte {
+ var dst []byte
+ dst = encoding.VarUint64ToBytes(dst, uint64(len(tt)))
+ familyNames := make([]string, 0, len(tt))
+ for familyName := range tt {
+ familyNames = append(familyNames, familyName)
+ }
+ sort.Strings(familyNames)
+ for _, familyName := range familyNames {
+ dst = encoding.EncodeBytes(dst,
convert.StringToBytes(familyName))
+ tags := tt[familyName]
+ dst = encoding.VarUint64ToBytes(dst, uint64(len(tags)))
+ tagNames := make([]string, 0, len(tags))
+ for tagName := range tags {
+ tagNames = append(tagNames, tagName)
+ }
+ sort.Strings(tagNames)
+ for _, tagName := range tagNames {
+ dst = encoding.EncodeBytes(dst,
convert.StringToBytes(tagName))
+ dst = append(dst, byte(tags[tagName]))
+ }
+ }
+ return dst
+}
+
+func (tt tagType) unmarshal(src []byte) error {
+ tt.reset()
+ remaining, familyCount := encoding.BytesToVarUint64(src)
+ for i := uint64(0); i < familyCount; i++ {
+ var familyNameBytes []byte
+ var decodeErr error
+ remaining, familyNameBytes, decodeErr =
encoding.DecodeBytes(remaining)
+ if decodeErr != nil {
+ return fmt.Errorf("cannot decode tag family name: %w",
decodeErr)
+ }
+ var tagCount uint64
+ remaining, tagCount = encoding.BytesToVarUint64(remaining)
+ tags := make(map[string]pbv1.ValueType, tagCount)
+ for j := uint64(0); j < tagCount; j++ {
+ var tagNameBytes []byte
+ remaining, tagNameBytes, decodeErr =
encoding.DecodeBytes(remaining)
+ if decodeErr != nil {
+ return fmt.Errorf("cannot decode tag name: %w",
decodeErr)
+ }
+ if len(remaining) < 1 {
+ return errors.New("insufficient data for tag
value type")
+ }
+ tags[string(tagNameBytes)] =
pbv1.ValueType(remaining[0])
+ remaining = remaining[1:]
+ }
+ tt[string(familyNameBytes)] = tags
+ }
+ return nil
+}
+
+func (tt tagType) mustWriteTagType(fileSystem fs.FileSystem, partPath string) {
+ if len(tt) == 0 {
+ return
+ }
+ data := tt.marshal()
+ tagTypePath := filepath.Join(partPath, tagTypeFilename)
+ written, writeErr := fileSystem.WriteAtomic(data, tagTypePath,
storage.FilePerm)
+ if writeErr != nil {
+ logger.Panicf("cannot write tag type: %s", writeErr)
+ return
+ }
+ if written != len(data) {
+ logger.Panicf("unexpected number of bytes written to %s; got
%d; want %d", tagTypePath, written, len(data))
+ }
+}
+
// ParsePartMetadata parses the part metadata from the metadata.json file.
func ParsePartMetadata(fileSystem fs.FileSystem, partPath string)
(queue.StreamingPartData, error) {
metadataPath := filepath.Join(partPath, metadataFilename)
diff --git a/banyand/measure/part_test.go b/banyand/measure/part_test.go
index 026d6956a..6fe8ff297 100644
--- a/banyand/measure/part_test.go
+++ b/banyand/measure/part_test.go
@@ -122,6 +122,12 @@ func TestMustInitFromDataPoints(t *testing.T) {
_, ok = p.tagFamilyMetadata[k]
require.True(t, ok,
"p.tagFamilyMetadata %s not found", k)
}
+ tagTypeData, readErr :=
fileSystem.Read(filepath.Join(path, tagTypeFilename))
+ require.NoError(t, readErr)
+ gotTagType := make(tagType)
+ require.NoError(t,
gotTagType.unmarshal(tagTypeData))
+ require.Equal(t, pbv1.ValueTypeStrArr,
gotTagType["arrTag"]["strArrTag"])
+ require.Equal(t, pbv1.ValueTypeInt64Arr,
gotTagType["arrTag"]["intArrTag"])
}
})
}
diff --git a/banyand/measure/syncer.go b/banyand/measure/syncer.go
index 31594289e..a2bd45bf3 100644
--- a/banyand/measure/syncer.go
+++ b/banyand/measure/syncer.go
@@ -20,12 +20,14 @@ package measure
import (
"context"
"fmt"
+ "path/filepath"
"strconv"
"time"
"github.com/apache/skywalking-banyandb/api/data"
"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/pool"
"github.com/apache/skywalking-banyandb/pkg/watcher"
@@ -132,6 +134,7 @@ func (tst *tsTable) syncLoop(syncCh chan *syncIntroduction,
flusherNotifier watc
func createPartFileReaders(part *part) ([]queue.FileInfo, func()) {
var files []queue.FileInfo
+ var buffersToRelease []*bytes.Buffer
buf := bigValuePool.Generate()
// Stream metadata.
@@ -177,8 +180,24 @@ func createPartFileReaders(part *part) ([]queue.FileInfo,
func()) {
}
}
+ if part.fileSystem != nil {
+ tagTypePath := filepath.Join(part.path, tagTypeFilename)
+ if tagTypeData, readErr := part.fileSystem.Read(tagTypePath);
readErr == nil && len(tagTypeData) > 0 {
+ tagTypeBuffer := bigValuePool.Generate()
+ tagTypeBuffer.Buf = append(tagTypeBuffer.Buf[:0],
tagTypeData...)
+ buffersToRelease = append(buffersToRelease,
tagTypeBuffer)
+ files = append(files, queue.FileInfo{
+ Name: tagTypeFilename,
+ Reader: tagTypeBuffer.SequentialRead(),
+ })
+ }
+ }
+
return files, func() {
bigValuePool.Release(bb)
+ for _, buffer := range buffersToRelease {
+ bigValuePool.Release(buffer)
+ }
}
}
diff --git a/banyand/measure/tstable_test.go b/banyand/measure/tstable_test.go
index 68c39d223..f01a7f6f0 100644
--- a/banyand/measure/tstable_test.go
+++ b/banyand/measure/tstable_test.go
@@ -172,6 +172,7 @@ func Test_tstIter(t *testing.T) {
cmpopts.IgnoreFields(blockMetadata{}, "timestamps"),
cmpopts.IgnoreFields(blockMetadata{}, "field"),
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/measure/write_data.go b/banyand/measure/write_data.go
index 301449a82..5bc993640 100644
--- a/banyand/measure/write_data.go
+++ b/banyand/measure/write_data.go
@@ -20,6 +20,7 @@ package measure
import (
"context"
"fmt"
+ "path/filepath"
"strings"
"sync/atomic"
"time"
@@ -41,12 +42,13 @@ type syncPartContext struct {
// can drive refCount to 0; releasing earlier lets each sync call
bottom out
// there, so the next session's incRef re-runs initTSTable and deletes
the
// in-flight part.
- segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
- fileSystem fs.FileSystem
- writers *writers
- partPath string
- partMeta partMetadata
- partID uint64
+ segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
+ fileSystem fs.FileSystem
+ writers *writers
+ partPath string
+ tagTypeBuffer []byte
+ partMeta partMetadata
+ partID uint64
}
func (s *syncPartContext) NewPartType(_ *queue.ChunkedSyncPartContext) error {
@@ -58,6 +60,9 @@ func (s *syncPartContext) FinishSync() error {
// Close writers first so file data is flushed before we write metadata.
s.releaseCoreWriters()
+ if len(s.tagTypeBuffer) > 0 {
+ fs.MustFlushAtomic(s.fileSystem, s.tagTypeBuffer,
filepath.Join(s.partPath, tagTypeFilename), storage.FilePerm)
+ }
s.partMeta.mustWriteMetadata(s.fileSystem, s.partPath)
// No SyncPath: mustWriteMetadata goes through WriteAtomic which already
// fsyncs the parent directory after rename.
@@ -90,6 +95,7 @@ func (s *syncPartContext) Close() error {
}
s.tsTable = nil
s.fileSystem = nil
+ s.tagTypeBuffer = nil
return nil
}
@@ -167,6 +173,8 @@ func (s *syncCallback) HandleFileChunk(ctx
*queue.ChunkedSyncPartContext, chunk
// Select the appropriate writer based on the filename and write the
chunk.
fileName := ctx.FileName
switch {
+ case fileName == tagTypeFilename:
+ partCtx.tagTypeBuffer = append(partCtx.tagTypeBuffer, chunk...)
case fileName == measureMetaName:
partCtx.writers.metaWriter.MustWrite(chunk)
case fileName == measurePrimaryName:
diff --git a/banyand/stream/merger.go b/banyand/stream/merger.go
index 05dc9bd5f..8d64fa7d8 100644
--- a/banyand/stream/merger.go
+++ b/banyand/stream/merger.go
@@ -258,8 +258,8 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
if err != nil {
return nil, err
}
- pm.mustWriteMetadata(fileSystem, dstPath)
mergedTagType.mustWriteTagType(fileSystem, dstPath)
+ pm.mustWriteMetadata(fileSystem, dstPath)
// No SyncPath: mustWriteMetadata goes through WriteAtomic which already
// fsyncs the parent directory after rename.
p := mustOpenFilePart(partID, root, fileSystem)
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index aa75cb343..d9c4e2b9c 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -595,12 +595,12 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
if err != nil {
return nil, nil, err
}
- pm.MinTimestamp = minTimestamp
- pm.MaxTimestamp = maxTimestamp
- pm.mustWriteMetadata(fileSystem, dstPath)
tf.mustWriteTraceIDFilter(fileSystem, dstPath)
tf.reset()
tt.mustWriteTagType(fileSystem, dstPath)
+ pm.MinTimestamp = minTimestamp
+ pm.MaxTimestamp = maxTimestamp
+ pm.mustWriteMetadata(fileSystem, dstPath)
// No SyncPath here: each mustWrite* helper goes through
fileSystem.WriteAtomic
// which already fsyncs the parent directory after rename. The last
atomic
// metadata write covers all prior dirent changes (data file creations).
