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 7ae98e0b6 feat: implement global tag type metadata for stream parts
(#1184)
7ae98e0b6 is described below
commit 7ae98e0b6addfe216b32712c57c054eabee5a4c2
Author: Huang Youliang <[email protected]>
AuthorDate: Sat Jun 27 08:57:44 2026 +0800
feat: implement global tag type metadata for stream parts (#1184)
* feat: implement global tag type metadata for stream parts
* Fixup
---------
Co-authored-by: Gao Hongtao <[email protected]>
---
banyand/stream/block.go | 9 ++++
banyand/stream/block_metadata.go | 16 ++++++
banyand/stream/block_test.go | 15 ++++++
banyand/stream/block_writer.go | 11 +++-
banyand/stream/file_part_sync_test.go | 30 ++++++++++-
banyand/stream/merger.go | 14 +++---
banyand/stream/merger_test.go | 8 +++
banyand/stream/part.go | 22 +++++++-
banyand/stream/part_metadata.go | 95 +++++++++++++++++++++++++++++++++++
banyand/stream/part_test.go | 6 +++
banyand/stream/syncer.go | 19 +++++++
banyand/stream/write_data.go | 20 +++++---
12 files changed, 249 insertions(+), 16 deletions(-)
diff --git a/banyand/stream/block.go b/banyand/stream/block.go
index f167c603c..8c83f0d28 100644
--- a/banyand/stream/block.go
+++ b/banyand/stream/block.go
@@ -190,10 +190,19 @@ func (b *block) validate() {
func (b *block) marshalTagFamily(tf tagFamily, bm *blockMetadata, ww *writers)
{
hw, w, fw := ww.getWriters(tf.name)
cc := tf.tags
+ 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 := generateTagFamilyMetadata()
cmm := cfm.resizeTagMetadata(len(cc))
for i := range cc {
cc[i].mustWriteTo(&cmm[i], w, fw)
+ tagTypes[cc[i].name] = cc[i].valueType
}
bb := bigValuePool.Generate()
defer bigValuePool.Release(bb)
diff --git a/banyand/stream/block_metadata.go b/banyand/stream/block_metadata.go
index cd117c2f7..db44348f7 100644
--- a/banyand/stream/block_metadata.go
+++ b/banyand/stream/block_metadata.go
@@ -61,6 +61,7 @@ func (d *dataBlock) unmarshal(src []byte) []byte {
type blockMetadata struct {
tagFamilies map[string]*dataBlock
+ tagType tagType
tagProjection []model.TagProjection
timestamps timestampsMetadata
elementIDs elementIDsMetadata
@@ -75,6 +76,16 @@ func (bm *blockMetadata) copyFrom(src *blockMetadata) {
bm.count = src.count
bm.timestamps.copyFrom(&src.timestamps)
bm.elementIDs.copyFrom(&src.elementIDs)
+ 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)
@@ -102,6 +113,11 @@ func (bm *blockMetadata) reset() {
bm.count = 0
bm.timestamps.reset()
bm.elementIDs.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/stream/block_test.go b/banyand/stream/block_test.go
index a2c08b9a5..b5195d19e 100644
--- a/banyand/stream/block_test.go
+++ b/banyand/stream/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"
@@ -340,6 +341,20 @@ 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,
+ "floatTag": pbv1.ValueTypeFloat64,
+ },
+ }, bm.tagType)
tagFamilyMetadataReaders := make(map[string]fs.Reader)
tagFamilyReaders := make(map[string]fs.Reader)
diff --git a/banyand/stream/block_writer.go b/banyand/stream/block_writer.go
index fdc77df25..c870c773c 100644
--- a/banyand/stream/block_writer.go
+++ b/banyand/stream/block_writer.go
@@ -155,6 +155,7 @@ func (sw *writers) getWriters(tagName string) (*writer,
*writer, *writer) {
type blockWriter struct {
writers writers
+ tagType tagType
metaData []byte
primaryBlockData []byte
primaryBlockMetadata primaryBlockMetadata
@@ -173,6 +174,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
@@ -230,6 +236,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
@@ -271,7 +278,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
@@ -286,6 +293,7 @@ func (bw *blockWriter) Flush(pm *partMetadata) {
bigValuePool.Release(bb)
pm.CompressedSizeBytes = bw.writers.totalBytesWritten()
+ tt.copyFrom(bw.tagType)
bw.writers.MustClose()
bw.reset()
@@ -295,6 +303,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/stream/file_part_sync_test.go
b/banyand/stream/file_part_sync_test.go
index bc1c47076..0404d3342 100644
--- a/banyand/stream/file_part_sync_test.go
+++ b/banyand/stream/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"
@@ -93,8 +94,27 @@ func TestMustAddFilePart_FilesOnDisk(t *testing.T) {
partID := uint64(1)
destPath := partPath(tabDir, partID)
buildAndFlushStreamMemPart(t, fileSystem, es, 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")
@@ -178,11 +198,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/stream/merger.go b/banyand/stream/merger.go
index 934eb43cb..05dc9bd5f 100644
--- a/banyand/stream/merger.go
+++ b/banyand/stream/merger.go
@@ -249,7 +249,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 {
@@ -259,6 +259,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
return nil, err
}
pm.mustWriteMetadata(fileSystem, dstPath)
+ mergedTagType.mustWriteTagType(fileSystem, dstPath)
// No SyncPath: mustWriteMetadata goes through WriteAtomic which already
// fsyncs the parent directory after rename.
p := mustOpenFilePart(partID, root, fileSystem)
@@ -268,7 +269,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)
@@ -289,7 +290,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
@@ -333,15 +334,16 @@ func mergeBlocks(closeCh <-chan struct{}, bw
*blockWriter, br *blockReader) (*pa
pendingBlockIsEmpty = true
}
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/stream/merger_test.go b/banyand/stream/merger_test.go
index 11c6022e8..5062f8f03 100644
--- a/banyand/stream/merger_test.go
+++ b/banyand/stream/merger_test.go
@@ -19,6 +19,7 @@ package stream
import (
"errors"
+ "path/filepath"
"reflect"
"testing"
@@ -317,6 +318,13 @@ func Test_mergeParts(t *testing.T) {
); diff != "" {
t.Errorf("Unexpected blockMetadata
(-got +want):\n%s", diff)
}
+
+ 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/stream/part.go b/banyand/stream/part.go
index 7ee11504c..7fac7392f 100644
--- a/banyand/stream/part.go
+++ b/banyand/stream/part.go
@@ -45,6 +45,7 @@ const (
streamSeriesMetadataName = "smeta"
metadataFilename = "metadata.json"
+ tagTypeFilename = "tag.type"
primaryFilename = streamPrimaryName + ".bin"
metaFilename = streamMetaName + ".bin"
timestampsFilename = streamTimestampsName + ".bin"
@@ -118,6 +119,7 @@ type memPart struct {
tagFamilyMetadata map[string]*bytes.Buffer
tagFamilies map[string]*bytes.Buffer
tagFamilyFilter map[string]*bytes.Buffer
+ tagType tagType
meta bytes.Buffer
primary bytes.Buffer
timestamps bytes.Buffer
@@ -149,6 +151,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()
@@ -203,7 +210,7 @@ func (mp *memPart) mustInitFromElements(es *elements) {
uncompressedBlockSizeBytes += uncompressedElementSizeBytes(i,
es)
}
bsw.MustWriteElements(sidPrev, es.timestamps[indexPrev:],
es.elementIDs[indexPrev:], es.tagFamilies[indexPrev:])
- bsw.Flush(&mp.partMetadata)
+ bsw.Flush(&mp.partMetadata, &mp.tagType)
releaseBlockWriter(bsw)
}
@@ -231,6 +238,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
@@ -422,6 +430,18 @@ func CreatePartFileReaderFromPath(partPath string, lfs
fs.FileSystem) ([]queue.F
if e.IsDir() {
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(),
+ })
+ }
if filepath.Ext(e.Name()) == tagFamiliesMetadataFilenameExt {
tfmPath := path.Join(partPath, e.Name())
tfmReader, err := lfs.OpenFile(tfmPath)
diff --git a/banyand/stream/part_metadata.go b/banyand/stream/part_metadata.go
index 47cddb501..c4edcfdd3 100644
--- a/banyand/stream/part_metadata.go
+++ b/banyand/stream/part_metadata.go
@@ -19,14 +19,19 @@ package stream
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/stream/part_test.go b/banyand/stream/part_test.go
index 638bde1ea..da521b850 100644
--- a/banyand/stream/part_test.go
+++ b/banyand/stream/part_test.go
@@ -114,6 +114,12 @@ func TestMustInitFromElements(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/stream/syncer.go b/banyand/stream/syncer.go
index 33c10c13e..fc5dcfb2d 100644
--- a/banyand/stream/syncer.go
+++ b/banyand/stream/syncer.go
@@ -20,12 +20,14 @@ package stream
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/watcher"
)
@@ -102,6 +104,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
@@ -150,8 +153,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/stream/write_data.go b/banyand/stream/write_data.go
index af360c9c4..7a92aac25 100644
--- a/banyand/stream/write_data.go
+++ b/banyand/stream/write_data.go
@@ -20,6 +20,7 @@ package stream
import (
"context"
"fmt"
+ "path/filepath"
"strings"
"sync/atomic"
"time"
@@ -40,12 +41,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 {
@@ -57,6 +59,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.
@@ -89,6 +94,7 @@ func (s *syncPartContext) Close() error {
}
s.tsTable = nil
s.fileSystem = nil
+ s.tagTypeBuffer = nil
return nil
}
@@ -166,6 +172,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 == streamMetaName:
partCtx.writers.metaWriter.MustWrite(chunk)
case fileName == streamPrimaryName:
