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 59b33b531 Support multiple types for the same tag in a single measure
or stream part (#1126)
59b33b531 is described below
commit 59b33b531257c3909bfa17dfd26daca8802f6387
Author: Huang Youliang <[email protected]>
AuthorDate: Wed Jul 8 15:16:06 2026 +0800
Support multiple types for the same tag in a single measure or stream part
(#1126)
---
banyand/measure/block.go | 40 ++++++-
banyand/measure/block_test.go | 4 +-
banyand/measure/column.go | 44 +++++++
banyand/measure/column_test.go | 19 +++
banyand/measure/measure_suite_test.go | 2 +-
banyand/measure/merger.go | 72 +++++++++++-
banyand/measure/merger_test.go | 188 ++++++++++++++++++++++++++++++
banyand/measure/metadata_test.go | 77 +++++++++++-
banyand/measure/migration_copy.go | 2 +-
banyand/measure/part.go | 4 +
banyand/measure/part_metadata.go | 21 ++++
banyand/measure/svc_standalone.go | 1 +
banyand/stream/block.go | 40 ++++++-
banyand/stream/block_test.go | 4 +-
banyand/stream/merger.go | 72 +++++++++++-
banyand/stream/merger_test.go | 186 +++++++++++++++++++++++++++++
banyand/stream/metadata_test.go | 80 ++++++++++++-
banyand/stream/migration_copy.go | 2 +-
banyand/stream/migration_element_index.go | 2 +-
banyand/stream/part.go | 4 +
banyand/stream/part_metadata.go | 21 ++++
banyand/stream/stream_suite_test.go | 2 +-
banyand/stream/svc_standalone.go | 1 +
banyand/stream/tag.go | 44 +++++++
banyand/stream/tag_test.go | 19 +++
25 files changed, 916 insertions(+), 35 deletions(-)
diff --git a/banyand/measure/block.go b/banyand/measure/block.go
index cb88287c3..3164d076f 100644
--- a/banyand/measure/block.go
+++ b/banyand/measure/block.go
@@ -216,6 +216,7 @@ func (b *block) marshalTagFamily(tf columnFamily, bm
*blockMetadata, ww *writers
func (b *block) unmarshalTagFamily(decoder *encoding.BytesBlockDecoder,
tfIndex int, name string,
columnFamilyMetadataBlock *dataBlock, tagProjection []string,
metaReader, valueReader fs.Reader, count int,
+ schemaTagTypes map[string]pbv1.ValueType,
) {
if len(tagProjection) < 1 {
return
@@ -234,13 +235,28 @@ func (b *block) unmarshalTagFamily(decoder
*encoding.BytesBlockDecoder, tfIndex
cc := b.tagFamilies[tfIndex].resizeColumns(len(tagProjection))
NEXT:
for j := range tagProjection {
+ tp := tagProjection[j]
+ schemaType, hasSchemaType := schemaTagTypes[tp]
+ if hasSchemaType {
+ typedName := encodeTypedColumn(tp, schemaType)
+ for i := range cfm.columnMetadata {
+ if cfm.columnMetadata[i].name == typedName {
+ cc[j].mustReadValues(decoder,
valueReader, cfm.columnMetadata[i], uint64(b.Len()))
+ cc[j].name = tp
+ continue NEXT
+ }
+ }
+ }
for i := range cfm.columnMetadata {
- if tagProjection[j] == cfm.columnMetadata[i].name {
+ if cfm.columnMetadata[i].name == tp {
+ if hasSchemaType &&
cfm.columnMetadata[i].valueType != schemaType {
+ continue
+ }
cc[j].mustReadValues(decoder, valueReader,
cfm.columnMetadata[i], uint64(b.Len()))
continue NEXT
}
}
- cc[j].name = tagProjection[j]
+ cc[j].name = tp
cc[j].valueType = pbv1.ValueTypeUnknown
cc[j].resizeValues(count)
for k := range cc[j].values {
@@ -305,7 +321,7 @@ func (b *block) uncompressedSizeBytes() uint64 {
return n
}
-func (b *block) mustReadFrom(decoder *encoding.BytesBlockDecoder, p *part, bm
blockMetadata) {
+func (b *block) mustReadFrom(decoder *encoding.BytesBlockDecoder, p *part, bm
blockMetadata, schemaTagTypes map[string]pbv1.ValueType) {
b.reset()
b.timestamps, b.versions = mustReadTimestampsFrom(b.timestamps,
b.versions, &bm.timestamps, int(bm.count), p.timestamps)
@@ -334,7 +350,7 @@ func (b *block) mustReadFrom(decoder
*encoding.BytesBlockDecoder, p *part, bm bl
}
b.unmarshalTagFamily(decoder, i, name, block,
bm.tagProjection[i].Names, p.tagFamilyMetadata[name],
- p.tagFamilies[name], int(bm.count))
+ p.tagFamilies[name], int(bm.count), schemaTagTypes)
}
}
@@ -829,7 +845,7 @@ NEXT_FIELD:
}
}
bc.bm.tagFamilies = tf
- tmpBlock.mustReadFrom(&bc.columnValuesDecoder, bc.p, bc.bm)
+ tmpBlock.mustReadFrom(&bc.columnValuesDecoder, bc.p, bc.bm,
bc.schemaTagTypes)
start, end, ok := timestamp.FindRange(tmpBlock.timestamps,
bc.minTimestamp, bc.maxTimestamp)
if !ok {
@@ -952,16 +968,20 @@ func fastTagAppend(bi, b *blockPointer, offset int) error
{
if len(bi.tagFamilies) != len(b.tagFamilies) {
return fmt.Errorf("unexpected number of tag families: got %d;
want %d", len(b.tagFamilies), len(bi.tagFamilies))
}
+ existingDataSize := len(bi.timestamps)
for i := range bi.tagFamilies {
if bi.tagFamilies[i].name != b.tagFamilies[i].name {
+ rollbackFastTagAppend(bi, existingDataSize)
return fmt.Errorf("unexpected tag family name: got %q;
want %q", b.tagFamilies[i].name, bi.tagFamilies[i].name)
}
if len(bi.tagFamilies[i].columns) !=
len(b.tagFamilies[i].columns) {
+ rollbackFastTagAppend(bi, existingDataSize)
return fmt.Errorf("unexpected number of tags for tag
family %q: got %d; want %d",
bi.tagFamilies[i].name,
len(b.tagFamilies[i].columns), len(bi.tagFamilies[i].columns))
}
for j := range bi.tagFamilies[i].columns {
if bi.tagFamilies[i].columns[j].name !=
b.tagFamilies[i].columns[j].name {
+ rollbackFastTagAppend(bi, existingDataSize)
return fmt.Errorf("unexpected tag name for tag
family %q: got %q; want %q",
bi.tagFamilies[i].name,
b.tagFamilies[i].columns[j].name, bi.tagFamilies[i].columns[j].name)
}
@@ -972,6 +992,16 @@ func fastTagAppend(bi, b *blockPointer, offset int) error {
return nil
}
+func rollbackFastTagAppend(bi *blockPointer, existingDataSize int) {
+ for i := range bi.tagFamilies {
+ for j := range bi.tagFamilies[i].columns {
+ if len(bi.tagFamilies[i].columns[j].values) >
existingDataSize {
+ bi.tagFamilies[i].columns[j].values =
bi.tagFamilies[i].columns[j].values[:existingDataSize]
+ }
+ }
+ }
+}
+
func fullTagAppend(bi, b *blockPointer, offset int) {
existDataSize := len(bi.timestamps)
appendTagFamilies := func(tf columnFamily) {
diff --git a/banyand/measure/block_test.go b/banyand/measure/block_test.go
index 9414954c7..04b0bb61b 100644
--- a/banyand/measure/block_test.go
+++ b/banyand/measure/block_test.go
@@ -329,7 +329,7 @@ func Test_marshalAndUnmarshalTagFamily(t *testing.T) {
unmarshaled.timestamps = make([]int64, len(b.timestamps))
unmarshaled.resizeTagFamilies(1)
- unmarshaled.unmarshalTagFamily(decoder, tfIndex, name,
bm.getTagFamilyMetadata(name), tagProjection[name], metaBuffer, dataBuffer, 1)
+ unmarshaled.unmarshalTagFamily(decoder, tfIndex, name,
bm.getTagFamilyMetadata(name), tagProjection[name], metaBuffer, dataBuffer, 1,
nil)
if diff := cmp.Diff(unmarshaled.tagFamilies[0], b.tagFamilies[0],
cmp.AllowUnexported(columnFamily{}, column{}),
@@ -419,7 +419,7 @@ func Test_marshalAndUnmarshalBlock(t *testing.T) {
})
}
bm.tagProjection = tp
- unmarshaled.mustReadFrom(decoder, p, bm)
+ unmarshaled.mustReadFrom(decoder, p, bm, nil)
// blockMetadata is using a map, so the order of tag families is not
guaranteed
unmarshaled.sortTagFamilies()
diff --git a/banyand/measure/column.go b/banyand/measure/column.go
index c5bf2978b..795110b2e 100644
--- a/banyand/measure/column.go
+++ b/banyand/measure/column.go
@@ -18,6 +18,8 @@
package measure
import (
+ "strings"
+
"github.com/apache/skywalking-banyandb/pkg/bytes"
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/encoding"
@@ -27,6 +29,48 @@ import (
"github.com/apache/skywalking-banyandb/pkg/pool"
)
+const typedColumnSeparator = "#"
+
+var (
+ valueTypeToSuffix = map[pbv1.ValueType]string{
+ pbv1.ValueTypeStr: "str",
+ pbv1.ValueTypeInt64: "int",
+ pbv1.ValueTypeFloat64: "float",
+ pbv1.ValueTypeBinaryData: "bin",
+ pbv1.ValueTypeStrArr: "str_arr",
+ pbv1.ValueTypeInt64Arr: "int_arr",
+ pbv1.ValueTypeTimestamp: "ts",
+ pbv1.ValueTypeUnknown: "",
+ }
+ suffixToValueType = map[string]pbv1.ValueType{
+ "str": pbv1.ValueTypeStr,
+ "int": pbv1.ValueTypeInt64,
+ "float": pbv1.ValueTypeFloat64,
+ "bin": pbv1.ValueTypeBinaryData,
+ "str_arr": pbv1.ValueTypeStrArr,
+ "int_arr": pbv1.ValueTypeInt64Arr,
+ "ts": pbv1.ValueTypeTimestamp,
+ }
+)
+
+func encodeTypedColumn(name string, vt pbv1.ValueType) string {
+ suffix, ok := valueTypeToSuffix[vt]
+ if !ok || suffix == "" {
+ return name
+ }
+ return name + typedColumnSeparator + suffix
+}
+
+func decodeTypedColumn(key string) string {
+ for suffix := range suffixToValueType {
+ sepSuffix := typedColumnSeparator + suffix
+ if strings.HasSuffix(key, sepSuffix) {
+ return key[:len(key)-len(sepSuffix)]
+ }
+ }
+ return key
+}
+
func generateInt64Slice(length int) *[]int64 {
v := int64SlicePool.Get()
if v == nil {
diff --git a/banyand/measure/column_test.go b/banyand/measure/column_test.go
index 9f2e39ecd..e54bd3f6a 100644
--- a/banyand/measure/column_test.go
+++ b/banyand/measure/column_test.go
@@ -271,3 +271,22 @@ func getColumnEncodeTypeName(encType encoding.EncodeType)
string {
return fmt.Sprintf("Unknown(%d)", encType)
}
}
+
+func TestEncodeTypedColumn(t *testing.T) {
+ assert.Equal(t, "state#int", encodeTypedColumn("state",
pbv1.ValueTypeInt64))
+ assert.Equal(t, "state#str", encodeTypedColumn("state",
pbv1.ValueTypeStr))
+ assert.Equal(t, "http.status#int", encodeTypedColumn("http.status",
pbv1.ValueTypeInt64))
+ assert.Equal(t, "data#bin", encodeTypedColumn("data",
pbv1.ValueTypeBinaryData))
+ assert.Equal(t, "unknown", encodeTypedColumn("unknown",
pbv1.ValueTypeUnknown))
+}
+
+func TestDecodeTypedColumn(t *testing.T) {
+ assert.Equal(t, "state", decodeTypedColumn("state#int"))
+ assert.Equal(t, "state", decodeTypedColumn("state#str"))
+ assert.Equal(t, "http.status", decodeTypedColumn("http.status#int"))
+ assert.Equal(t, "http.status", decodeTypedColumn("http.status"))
+ assert.Equal(t, "simple", decodeTypedColumn("simple"))
+ assert.Equal(t, "my.str", decodeTypedColumn("my.str"))
+ assert.Equal(t, "arr.int", decodeTypedColumn("arr.int"))
+ assert.Equal(t, "data.bin", decodeTypedColumn("data.bin"))
+}
diff --git a/banyand/measure/measure_suite_test.go
b/banyand/measure/measure_suite_test.go
index 7b5b08596..8d307eb68 100644
--- a/banyand/measure/measure_suite_test.go
+++ b/banyand/measure/measure_suite_test.go
@@ -97,7 +97,7 @@ func setUp() (*services, func()) {
)
rootPath, deferFunc, err := test.NewSpace()
gomega.Expect(err).NotTo(gomega.HaveOccurred())
- flags = append(flags, "--measure-root-path="+rootPath)
+ flags = append(flags, "--measure-root-path="+rootPath,
"--measure-max-merge-parts=2")
moduleDeferFunc := test.SetupModules(
flags,
pipeline,
diff --git a/banyand/measure/merger.go b/banyand/measure/merger.go
index 26251f4e6..de3693289 100644
--- a/banyand/measure/merger.go
+++ b/banyand/measure/merger.go
@@ -30,6 +30,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/cgroups"
"github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/fs"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
@@ -237,6 +238,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
return nil, errNoPartToMerge
}
dstPath := partPath(root, partID)
+ conflictColumns := collectConflictColumns(parts)
var totalSize int64
pii := make([]*partMergeIter, 0, len(parts))
for i := range parts {
@@ -251,7 +253,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
bw := generateBlockWriter()
bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
- pm, mergedTagType, err := mergeBlocks(closeCh, bw, br)
+ pm, mergedTagType, err := mergeBlocks(closeCh, bw, br, conflictColumns)
releaseBlockWriter(bw)
releaseBlockReader(br)
for i := range pii {
@@ -270,7 +272,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, *tagType, error) {
+func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader,
conflictColumns map[string]map[string]struct{}) (*partMetadata, *tagType,
error) {
pendingBlockIsEmpty := true
pendingBlock := generateBlockPointer()
defer releaseBlockPointer(pendingBlock)
@@ -288,6 +290,10 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
decoder = nil
}
}
+ loadAndRename := func() {
+ br.loadBlockData(getDecoder())
+ renameConflictColumns(&br.block.block, conflictColumns)
+ }
for br.nextBlockMetadata() {
select {
case <-closeCh:
@@ -297,7 +303,7 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
b := br.block
if pendingBlockIsEmpty {
- br.loadBlockData(getDecoder())
+ loadAndRename()
pendingBlock.copyFrom(b)
pendingBlockIsEmpty = false
continue
@@ -307,7 +313,7 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
(pendingBlock.isFull() &&
pendingBlock.bm.timestamps.max <= b.bm.timestamps.min) {
bw.mustWriteBlock(pendingBlock.bm.seriesID,
&pendingBlock.block)
releaseDecoder()
- br.loadBlockData(getDecoder())
+ loadAndRename()
pendingBlock.copyFrom(b)
continue
}
@@ -318,7 +324,7 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
}
tmpBlock.reset()
tmpBlock.bm.seriesID = b.bm.seriesID
- br.loadBlockData(getDecoder())
+ loadAndRename()
mergeTwoBlocks(tmpBlock, pendingBlock, b)
if len(tmpBlock.timestamps) <= maxBlockLength &&
tmpBlock.uncompressedSizeBytes() <= maxUncompressedBlockSize {
if len(tmpBlock.timestamps) == 0 {
@@ -361,6 +367,62 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
return &result, &mergedTagType, nil
}
+func renameConflictColumns(b *block, conflictColumns
map[string]map[string]struct{}) {
+ if len(conflictColumns) == 0 {
+ return
+ }
+ for i := range b.tagFamilies {
+ columns := conflictColumns[b.tagFamilies[i].name]
+ if columns == nil {
+ continue
+ }
+ cc := b.tagFamilies[i].columns
+ for j := range cc {
+ if _, ok := columns[cc[j].name]; ok {
+ cc[j].name = encodeTypedColumn(cc[j].name,
cc[j].valueType)
+ }
+ }
+ }
+}
+
+func collectConflictColumns(parts []*partWrapper)
map[string]map[string]struct{} {
+ familyColumnTypes :=
make(map[string]map[string]map[pbv1.ValueType]struct{})
+ for _, pw := range parts {
+ for cf, cc := range pw.p.tagType {
+ columnTypes := familyColumnTypes[cf]
+ if columnTypes == nil {
+ columnTypes =
make(map[string]map[pbv1.ValueType]struct{})
+ familyColumnTypes[cf] = columnTypes
+ }
+ for name, vt := range cc {
+ decoded := decodeTypedColumn(name)
+ valueTypes := columnTypes[decoded]
+ if valueTypes == nil {
+ valueTypes =
make(map[pbv1.ValueType]struct{})
+ columnTypes[decoded] = valueTypes
+ }
+ valueTypes[vt] = struct{}{}
+ }
+ }
+ }
+ var conflictColumns map[string]map[string]struct{}
+ for cf, columnTypes := range familyColumnTypes {
+ for name, valueTypes := range columnTypes {
+ if len(valueTypes) <= 1 {
+ continue
+ }
+ if conflictColumns == nil {
+ conflictColumns =
make(map[string]map[string]struct{})
+ }
+ if conflictColumns[cf] == nil {
+ conflictColumns[cf] = make(map[string]struct{})
+ }
+ conflictColumns[cf][name] = struct{}{}
+ }
+ }
+ return conflictColumns
+}
+
func mergeTwoBlocks(target, left, right *blockPointer) {
appendIfEmpty := func(ib1, ib2 *blockPointer) bool {
if ib1.idx >= len(ib1.timestamps) {
diff --git a/banyand/measure/merger_test.go b/banyand/measure/merger_test.go
index 39d92142c..9ded3ac64 100644
--- a/banyand/measure/merger_test.go
+++ b/banyand/measure/merger_test.go
@@ -1523,3 +1523,191 @@ func Test_mergeParts_fileBased(t *testing.T) {
"should have at least one block per seriesID")
})
}
+
+func Test_mergePartsWithConflictingTagTypes(t *testing.T) {
+ type wantColumn struct {
+ value []byte
+ valueType pbv1.ValueType
+ }
+ type wantBlock struct {
+ columns map[string]wantColumn
+ seriesID common.SeriesID
+ }
+ tests := []struct {
+ name string
+ dpsList []*dataPoints
+ wantBlocks []wantBlock
+ }{
+ {
+ name: "two parts with same tag name and conflicting
valueTypes",
+ dpsList: []*dataPoints{
+ {
+ seriesIDs: []common.SeriesID{1},
+ timestamps: []int64{1},
+ versions: []int64{1},
+ tagFamilies: [][]nameValues{
+ {
+ {name: "searchable",
values: []*nameValue{
+ {name:
"extra_tag", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)},
+ }},
+ },
+ },
+ fields: []nameValues{{}},
+ },
+ {
+ seriesIDs: []common.SeriesID{2},
+ timestamps: []int64{2},
+ versions: []int64{1},
+ tagFamilies: [][]nameValues{
+ {
+ {name: "searchable",
values: []*nameValue{
+ {name:
"extra_tag", valueType: pbv1.ValueTypeStr, value: []byte("active")},
+ }},
+ },
+ },
+ fields: []nameValues{{}},
+ },
+ },
+ wantBlocks: []wantBlock{
+ {
+ seriesID: 1,
+ columns: map[string]wantColumn{
+ "searchable/extra_tag#int":
{valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)},
+ },
+ },
+ {
+ seriesID: 2,
+ columns: map[string]wantColumn{
+ "searchable/extra_tag#str":
{valueType: pbv1.ValueTypeStr, value: []byte("active")},
+ },
+ },
+ },
+ },
+ {
+ name: "merge typed-suffix part with plain part",
+ dpsList: []*dataPoints{
+ {
+ seriesIDs: []common.SeriesID{1},
+ timestamps: []int64{1},
+ versions: []int64{1},
+ tagFamilies: [][]nameValues{
+ {
+ {name: "searchable",
values: []*nameValue{
+ {name:
"extra_tag#int", valueType: pbv1.ValueTypeInt64, value:
convert.Int64ToBytes(100)},
+ }},
+ },
+ },
+ fields: []nameValues{{}},
+ },
+ {
+ seriesIDs: []common.SeriesID{2},
+ timestamps: []int64{2},
+ versions: []int64{1},
+ tagFamilies: [][]nameValues{
+ {
+ {name: "searchable",
values: []*nameValue{
+ {name:
"extra_tag", valueType: pbv1.ValueTypeStr, value: []byte("pending")},
+ }},
+ },
+ },
+ fields: []nameValues{{}},
+ },
+ },
+ wantBlocks: []wantBlock{
+ {
+ seriesID: 1,
+ columns: map[string]wantColumn{
+ "searchable/extra_tag#int":
{valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)},
+ },
+ },
+ {
+ seriesID: 2,
+ columns: map[string]wantColumn{
+ "searchable/extra_tag#str":
{valueType: pbv1.ValueTypeStr, value: []byte("pending")},
+ },
+ },
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+
+ var fpp []*partWrapper
+ defer func() {
+ for _, pw := range fpp {
+ pw.decRef()
+ }
+ }()
+ for i, dps := range tt.dpsList {
+ mp := generateMemPart()
+ mp.mustInitFromDataPoints(dps)
+ mp.mustFlush(fileSystem, partPath(tmpPath,
uint64(i+1)))
+ filePW := newPartWrapper(nil,
mustOpenFilePart(uint64(i+1), tmpPath, fileSystem))
+ filePW.p.partMetadata.ID = uint64(i + 1)
+ fpp = append(fpp, filePW)
+ releaseMemPart(mp)
+ }
+
+ conflictColumns := collectConflictColumns(fpp)
+ require.NotNil(t, conflictColumns, "expected conflict
columns to be collected")
+ require.Contains(t, conflictColumns, "searchable")
+ require.Contains(t, conflictColumns["searchable"],
"extra_tag")
+
+ closeCh := make(chan struct{})
+ defer close(closeCh)
+ tst := &tsTable{pm: protector.Nop{}}
+ merged, err := tst.mergeParts(fileSystem, closeCh, fpp,
uint64(len(tt.dpsList)+1), tmpPath)
+ require.NoError(t, err)
+ defer merged.decRef()
+
+ pmi := &partMergeIter{}
+ pmi.mustInitFromPart(merged.p)
+ reader := &blockReader{}
+ reader.init([]*partMergeIter{pmi})
+ decoder := generateColumnValuesDecoder()
+ defer releaseColumnValuesDecoder(decoder)
+
+ gotBlocks :=
make(map[common.SeriesID]map[string]wantColumn)
+ for reader.nextBlockMetadata() {
+ reader.loadBlockData(decoder)
+ bp := reader.block
+ cc := make(map[string]wantColumn)
+ for i := range bp.tagFamilies {
+ tf := &bp.tagFamilies[i]
+ for j := range tf.columns {
+ c := &tf.columns[j]
+ require.NotEmpty(t, c.values,
"column %q in family %q should have values", c.name, tf.name)
+ cc[tf.name+"/"+c.name] =
wantColumn{
+ valueType: c.valueType,
+ value: c.values[0],
+ }
+ }
+ }
+ gotBlocks[bp.bm.seriesID] = cc
+ }
+ require.NoError(t, reader.error())
+
+ require.Len(t, gotBlocks, len(tt.wantBlocks))
+ for _, wb := range tt.wantBlocks {
+ got, ok := gotBlocks[wb.seriesID]
+ require.True(t, ok, "missing block for seriesID
%d", wb.seriesID)
+ for key, want := range wb.columns {
+ gotColumn, exists := got[key]
+ require.True(t, exists, "seriesID %d:
missing column %q", wb.seriesID, key)
+ require.Equal(t, want.valueType,
gotColumn.valueType,
+ "seriesID %d: valueType
mismatch for %q", wb.seriesID, key)
+ require.True(t,
reflect.DeepEqual(want.value, gotColumn.value),
+ "seriesID %d: value mismatch
for %q: got %v, want %v",
+ wb.seriesID, key,
gotColumn.value, want.value)
+ }
+ _, hasPlain := got["searchable/extra_tag"]
+ require.False(t, hasPlain,
+ "seriesID %d: plain-named conflict
column should not exist after merge", wb.seriesID)
+ }
+ })
+ }
+}
diff --git a/banyand/measure/metadata_test.go b/banyand/measure/metadata_test.go
index 501580bbe..ef30f0c1a 100644
--- a/banyand/measure/metadata_test.go
+++ b/banyand/measure/metadata_test.go
@@ -20,6 +20,7 @@ package measure_test
import (
"context"
"errors"
+ "fmt"
"strconv"
"time"
@@ -408,7 +409,7 @@ var _ = Describe("Schema Change", func() {
for _, dp := range dataPoints {
for _, tf := range dp.TagFamilies {
- if tf.Name == "default" {
+ if tf.Name == defaultTagFamily {
for _, tag := range
tf.Tags {
innerGm.Expect(tag.Key).NotTo(Equal("extra_tag"),
"deleted tag should not be returned in query results")
@@ -441,7 +442,7 @@ var _ = Describe("Schema Change", func() {
newDataCount := 0
for _, dp := range dataPoints {
for _, tf := range dp.TagFamilies {
- if tf.Name == "default" {
+ if tf.Name == defaultTagFamily {
for _, tag := range
tf.Tags {
if tag.Key ==
"extra_tag" {
if
tag.Value.GetInt() != nil {
@@ -481,7 +482,7 @@ var _ = Describe("Schema Change", func() {
stringCount := 0
for _, dp := range dataPoints {
for _, tf := range dp.TagFamilies {
- if tf.Name == "default" {
+ if tf.Name == defaultTagFamily {
for _, tag := range
tf.Tags {
if tag.Key ==
"extra_tag" {
switch
tag.Value.GetValue().(type) {
@@ -503,6 +504,59 @@ var _ = Describe("Schema Change", func() {
})
})
+ Context("Measure schema with changed tag type after merge", func() {
+ It("querying data should return correct values after parts with
different types are merged", func() {
+ measureName := "schema_change_tag_type_merge"
+ now := timestamp.NowMilli()
+
+ env := setupSchemaChangeMeasure(svcs, measureName,
measureSetupOptions{withExtraTag: true})
+ writeSchemaChangeMeasureData(svcs, measureName,
now.Add(-2*time.Hour), 5,
+ measureWriteDataOptions{withExtraTag: true})
+ changeExtraMeasureTagType(svcs, measureName)
+ writeSchemaChangeMeasureData(svcs, measureName,
now.Add(-1*time.Hour), 3,
+ measureWriteDataOptions{withExtraTagString:
true, entityIDPrefix: "entity_new_"})
+ partCountBeforeMerge, partCountErr :=
getTotalMeasurePartCount(svcs, groupName)
+ Expect(partCountErr).ShouldNot(HaveOccurred())
+ Eventually(func(innerGm Gomega) int64 {
+ currentPartCount, currentPartCountErr :=
getTotalMeasurePartCount(svcs, groupName)
+
innerGm.Expect(currentPartCountErr).ShouldNot(HaveOccurred())
+ return currentPartCount
+ }, flags.EventuallyTimeout).Should(BeNumerically("<",
partCountBeforeMerge))
+
+ Eventually(func(innerGm Gomega) {
+ dataPoints :=
querySchemaChangeMeasureData(svcs, measureName,
+ now.Add(-3*time.Hour), now,
+ []string{"id", "entity_id",
"extra_tag"}, []string{"total"})
+ innerGm.Expect(dataPoints).To(HaveLen(8))
+
+ nullCount := 0
+ stringCount := 0
+ for _, dp := range dataPoints {
+ for _, tf := range dp.TagFamilies {
+ if tf.Name == defaultTagFamily {
+ for _, tag := range
tf.Tags {
+ if tag.Key ==
"extra_tag" {
+ switch
tag.Value.GetValue().(type) {
+ case
*modelv1.TagValue_Null:
+
nullCount++
+ case
*modelv1.TagValue_Str:
+
stringCount++
+ }
+ }
+ }
+ }
+ }
+ }
+ innerGm.Expect(nullCount).To(Equal(5),
+ "old data with int type should return
null after schema changed to STRING and parts merged")
+ innerGm.Expect(stringCount).To(Equal(3),
+ "new data should have string extra_tag
values after merge")
+ }, flags.EventuallyTimeout).Should(Succeed())
+
+ env.cleanup()
+ })
+ })
+
Context("Measure schema with deleted tag family", func() {
It("querying data should succeed after a tag family is
deleted", func() {
measureName := "schema_change_deleted_family"
@@ -1037,3 +1091,20 @@ func queryMeasureWithDeletedFieldProjection(svcs
*services, measureName string,
return errors.New("unexpected data type")
}
}
+
+func getTotalMeasurePartCount(svcs *services, group string) (int64, error) {
+ dataInfo, err := svcs.measure.CollectDataInfo(context.TODO(), group)
+ if err != nil {
+ return 0, fmt.Errorf("collect measure data info: %w", err)
+ }
+ if dataInfo == nil {
+ return 0, errors.New("measure data info is nil")
+ }
+ var total int64
+ for _, seg := range dataInfo.SegmentInfo {
+ for _, shard := range seg.ShardInfo {
+ total += shard.PartCount
+ }
+ }
+ return total, nil
+}
diff --git a/banyand/measure/migration_copy.go
b/banyand/measure/migration_copy.go
index df4c6b5bc..2ba4ae43b 100644
--- a/banyand/measure/migration_copy.go
+++ b/banyand/measure/migration_copy.go
@@ -829,7 +829,7 @@ func slowCopyOnePart(in processPartInput, srcPartID uint64)
(processPartResult,
bm := &bms[j]
bm.tagProjection = in.tagProjection
b := generateBlock()
- b.mustReadFrom(in.decoder, p, *bm)
+ b.mustReadFrom(in.decoder, p, *bm, nil)
liveBlocks = append(liveBlocks, b)
for k := uint64(0); k < bm.count; k++ {
diff --git a/banyand/measure/part.go b/banyand/measure/part.go
index d67488fe8..75cec6aed 100644
--- a/banyand/measure/part.go
+++ b/banyand/measure/part.go
@@ -62,6 +62,7 @@ type part struct {
fileSystem fs.FileSystem
tagFamilyMetadata map[string]fs.Reader
tagFamilies map[string]fs.Reader
+ tagType tagType
seriesMetadata fs.Reader // Optional: series metadata reader
cache storage.Cache
path string
@@ -92,6 +93,7 @@ func (p *part) String() string {
func openMemPart(mp *memPart) *part {
var p part
p.partMetadata = mp.partMetadata
+ p.tagType = mp.tagType
p.primaryBlockMetadata =
mustReadPrimaryBlockMetadata(p.primaryBlockMetadata[:0], &mp.meta)
@@ -325,6 +327,8 @@ func mustOpenFilePart(id uint64, root string, fileSystem
fs.FileSystem) *part {
fs.CleanupLeftoverTmp(fileSystem, partPath)
p.partMetadata.mustReadMetadata(fileSystem, partPath)
p.partMetadata.ID = id
+ p.tagType = make(tagType)
+ p.tagType.mustReadTagType(fileSystem, partPath)
metaPath := path.Join(partPath, metaFilename)
pr := mustOpenReader(metaPath, fileSystem)
diff --git a/banyand/measure/part_metadata.go b/banyand/measure/part_metadata.go
index cf6f0c661..fdad43001 100644
--- a/banyand/measure/part_metadata.go
+++ b/banyand/measure/part_metadata.go
@@ -186,6 +186,27 @@ func (tt tagType) unmarshal(src []byte) error {
return nil
}
+func (tt tagType) mustReadTagType(fileSystem fs.FileSystem, partPath string) {
+ tt.reset()
+ tagTypePath := filepath.Join(partPath, tagTypeFilename)
+ data, err := fileSystem.Read(tagTypePath)
+ if err != nil {
+ var fsErr *fs.FileSystemError
+ if errors.As(err, &fsErr) && fsErr.Code == fs.IsNotExistError {
+ return
+ }
+ logger.Panicf("cannot read %s: %s", tagTypePath, err)
+ return
+ }
+ if len(data) == 0 {
+ return
+ }
+ if err = tt.unmarshal(data); err != nil {
+ logger.Panicf("cannot parse %q: %s", tagTypePath, err)
+ return
+ }
+}
+
func (tt tagType) mustWriteTagType(fileSystem fs.FileSystem, partPath string) {
if len(tt) == 0 {
return
diff --git a/banyand/measure/svc_standalone.go
b/banyand/measure/svc_standalone.go
index 8be02a81c..54d145054 100644
--- a/banyand/measure/svc_standalone.go
+++ b/banyand/measure/svc_standalone.go
@@ -180,6 +180,7 @@ func (s *standalone) FlagSet() *run.FlagSet {
flagS.DurationVar(&s.option.memWaitTimeout,
"measure-lifecycle-receive-mem-wait-timeout", 5*time.Minute,
"max time the migration receiver waits for memory to recover
before introducing an external segment")
s.option.mergePolicy = newDefaultMergePolicy()
+ flagS.IntVar(&s.option.mergePolicy.maxParts, "measure-max-merge-parts",
s.option.mergePolicy.maxParts, "the maximum number of parts to merge at once")
flagS.VarP(&s.option.mergePolicy.maxFanOutSize,
"measure-max-fan-out-size", "", "the upper bound of a single file size after
merge of measure")
s.option.seriesCacheMaxSize = run.Bytes(32 << 20)
flagS.VarP(&s.option.seriesCacheMaxSize,
"measure-series-cache-max-size", "", "the max size of series cache in each
group")
diff --git a/banyand/stream/block.go b/banyand/stream/block.go
index 8c83f0d28..f51fcdefa 100644
--- a/banyand/stream/block.go
+++ b/banyand/stream/block.go
@@ -219,6 +219,7 @@ func (b *block) marshalTagFamily(tf tagFamily, bm
*blockMetadata, ww *writers) {
func (b *block) unmarshalTagFamily(decoder *encoding.BytesBlockDecoder,
tfIndex int, name string,
tagFamilyMetadataBlock *dataBlock, tagProjection []string, metaReader,
valueReader fs.Reader, count int,
+ schemaTagTypes map[string]pbv1.ValueType,
) {
if len(tagProjection) < 1 {
return
@@ -237,13 +238,28 @@ func (b *block) unmarshalTagFamily(decoder
*encoding.BytesBlockDecoder, tfIndex
cc := b.tagFamilies[tfIndex].resizeTags(len(tagProjection))
NEXT:
for j := range tagProjection {
+ tp := tagProjection[j]
+ schemaType, hasSchemaType := schemaTagTypes[tp]
+ if hasSchemaType {
+ typedName := encodeTypedTag(tp, schemaType)
+ for i := range tfm.tagMetadata {
+ if tfm.tagMetadata[i].name == typedName {
+ cc[j].mustReadValues(decoder,
valueReader, tfm.tagMetadata[i], uint64(b.Len()))
+ cc[j].name = tp
+ continue NEXT
+ }
+ }
+ }
for i := range tfm.tagMetadata {
- if tagProjection[j] == tfm.tagMetadata[i].name {
+ if tfm.tagMetadata[i].name == tp {
+ if hasSchemaType &&
tfm.tagMetadata[i].valueType != schemaType {
+ continue
+ }
cc[j].mustReadValues(decoder, valueReader,
tfm.tagMetadata[i], uint64(b.Len()))
continue NEXT
}
}
- cc[j].name = tagProjection[j]
+ cc[j].name = tp
cc[j].valueType = pbv1.ValueTypeUnknown
cc[j].resizeValues(count)
for k := range cc[j].values {
@@ -297,7 +313,7 @@ func (b *block) uncompressedSizeBytes() uint64 {
return n
}
-func (b *block) mustReadFrom(decoder *encoding.BytesBlockDecoder, p *part, bm
blockMetadata) {
+func (b *block) mustReadFrom(decoder *encoding.BytesBlockDecoder, p *part, bm
blockMetadata, schemaTagTypes map[string]pbv1.ValueType) {
b.reset()
b.timestamps, b.elementIDs = mustReadTimestampsFrom(b.timestamps,
b.elementIDs, &bm.timestamps, int(bm.count), p.timestamps)
@@ -321,7 +337,7 @@ func (b *block) mustReadFrom(decoder
*encoding.BytesBlockDecoder, p *part, bm bl
}
b.unmarshalTagFamily(decoder, i, name, block,
bm.tagProjection[i].Names, p.tagFamilyMetadata[name],
- p.tagFamilies[name], int(bm.count))
+ p.tagFamilies[name], int(bm.count), schemaTagTypes)
}
}
@@ -576,7 +592,7 @@ func (bc *blockCursor) loadData(tmpBlock *block) bool {
}
bc.bm.tagFamilies = tf
- tmpBlock.mustReadFrom(&bc.tagValuesDecoder, bc.p, bc.bm)
+ tmpBlock.mustReadFrom(&bc.tagValuesDecoder, bc.p, bc.bm,
bc.schemaTagTypes)
if len(tmpBlock.timestamps) == 0 {
return false
}
@@ -704,16 +720,20 @@ func fastTagAppend(bi, b *blockPointer, offset int) error
{
if len(bi.tagFamilies) != len(b.tagFamilies) {
return fmt.Errorf("unexpected number of tag families: got %d;
want %d", len(b.tagFamilies), len(bi.tagFamilies))
}
+ existingDataSize := len(bi.timestamps)
for i := range bi.tagFamilies {
if bi.tagFamilies[i].name != b.tagFamilies[i].name {
+ rollbackFastTagAppend(bi, existingDataSize)
return fmt.Errorf("unexpected tag family name: got %q;
want %q", b.tagFamilies[i].name, bi.tagFamilies[i].name)
}
if len(bi.tagFamilies[i].tags) != len(b.tagFamilies[i].tags) {
+ rollbackFastTagAppend(bi, existingDataSize)
return fmt.Errorf("unexpected number of tags for tag
family %q: got %d; want %d",
bi.tagFamilies[i].name,
len(b.tagFamilies[i].tags), len(bi.tagFamilies[i].tags))
}
for j := range bi.tagFamilies[i].tags {
if bi.tagFamilies[i].tags[j].name !=
b.tagFamilies[i].tags[j].name {
+ rollbackFastTagAppend(bi, existingDataSize)
return fmt.Errorf("unexpected tag name for tag
family %q: got %q; want %q",
bi.tagFamilies[i].name,
b.tagFamilies[i].tags[j].name, bi.tagFamilies[i].tags[j].name)
}
@@ -724,6 +744,16 @@ func fastTagAppend(bi, b *blockPointer, offset int) error {
return nil
}
+func rollbackFastTagAppend(bi *blockPointer, existingDataSize int) {
+ for i := range bi.tagFamilies {
+ for j := range bi.tagFamilies[i].tags {
+ if len(bi.tagFamilies[i].tags[j].values) >
existingDataSize {
+ bi.tagFamilies[i].tags[j].values =
bi.tagFamilies[i].tags[j].values[:existingDataSize]
+ }
+ }
+ }
+}
+
func fullTagAppend(bi, b *blockPointer, offset int) {
existDataSize := len(bi.timestamps)
appendTagFamilies := func(tf tagFamily) {
diff --git a/banyand/stream/block_test.go b/banyand/stream/block_test.go
index b5195d19e..5976fc004 100644
--- a/banyand/stream/block_test.go
+++ b/banyand/stream/block_test.go
@@ -288,7 +288,7 @@ func Test_marshalAndUnmarshalTagFamily(t *testing.T) {
unmarshaled.timestamps = make([]int64, len(b.timestamps))
unmarshaled.resizeTagFamilies(1)
- unmarshaled.unmarshalTagFamily(decoder, tfIndex, name,
bm.getTagFamilyMetadata(name), tagProjection[name], metaBuffer, dataBuffer, 1)
+ unmarshaled.unmarshalTagFamily(decoder, tfIndex, name,
bm.getTagFamilyMetadata(name), tagProjection[name], metaBuffer, dataBuffer, 1,
nil)
if diff := cmp.Diff(unmarshaled.tagFamilies[0], b.tagFamilies[0],
cmp.AllowUnexported(tagFamily{}, tag{}, tagFilter{}),
@@ -377,7 +377,7 @@ func Test_marshalAndUnmarshalBlock(t *testing.T) {
})
}
bm.tagProjection = tp
- unmarshaled.mustReadFrom(decoder, p, bm)
+ unmarshaled.mustReadFrom(decoder, p, bm, nil)
// blockMetadata is using a map, so the order of tag families is not
guaranteed
unmarshaled.sortTagFamilies()
diff --git a/banyand/stream/merger.go b/banyand/stream/merger.go
index 8d64fa7d8..320308a90 100644
--- a/banyand/stream/merger.go
+++ b/banyand/stream/merger.go
@@ -28,6 +28,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/cgroups"
"github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/fs"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
@@ -235,6 +236,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
return nil, errNoPartToMerge
}
dstPath := partPath(root, partID)
+ conflictTags := collectConflictTags(parts)
var totalSize int64
pii := make([]*partMergeIter, 0, len(parts))
for i := range parts {
@@ -249,7 +251,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem,
closeCh <-chan struct{}
bw := generateBlockWriter()
bw.mustInitForFilePart(fileSystem, dstPath, shouldCache)
- pm, mergedTagType, err := mergeBlocks(closeCh, bw, br)
+ pm, mergedTagType, err := mergeBlocks(closeCh, bw, br, conflictTags)
releaseBlockWriter(bw)
releaseBlockReader(br)
for i := range pii {
@@ -269,7 +271,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, *tagType, error) {
+func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter, br *blockReader,
conflictTags map[string]map[string]struct{}) (*partMetadata, *tagType, error) {
pendingBlockIsEmpty := true
pendingBlock := generateBlockPointer()
defer releaseBlockPointer(pendingBlock)
@@ -287,6 +289,10 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
decoder = nil
}
}
+ loadAndRename := func() {
+ br.loadBlockData(getDecoder())
+ renameConflictTags(&br.block.block, conflictTags)
+ }
for br.nextBlockMetadata() {
select {
case <-closeCh:
@@ -296,7 +302,7 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
b := br.block
if pendingBlockIsEmpty {
- br.loadBlockData(getDecoder())
+ loadAndRename()
pendingBlock.copyFrom(b)
pendingBlockIsEmpty = false
continue
@@ -307,7 +313,7 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
bw.mustWriteBlock(pendingBlock.bm.seriesID,
&pendingBlock.block)
releaseDecoder()
pendingBlock.reset()
- br.loadBlockData(getDecoder())
+ loadAndRename()
pendingBlock.copyFrom(b)
continue
}
@@ -318,7 +324,7 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
}
tmpBlock.reset()
tmpBlock.bm.seriesID = b.bm.seriesID
- br.loadBlockData(getDecoder())
+ loadAndRename()
mergeTwoBlocks(tmpBlock, pendingBlock, b)
if tmpBlock.uncompressedSizeBytes() <= maxUncompressedBlockSize
{
if len(tmpBlock.timestamps) == 0 {
@@ -346,6 +352,62 @@ func mergeBlocks(closeCh <-chan struct{}, bw *blockWriter,
br *blockReader) (*pa
return &result, &mergedTagType, nil
}
+func renameConflictTags(b *block, conflictTags map[string]map[string]struct{})
{
+ if len(conflictTags) == 0 {
+ return
+ }
+ for i := range b.tagFamilies {
+ tags := conflictTags[b.tagFamilies[i].name]
+ if tags == nil {
+ continue
+ }
+ tt := b.tagFamilies[i].tags
+ for j := range tt {
+ if _, ok := tags[tt[j].name]; ok {
+ tt[j].name = encodeTypedTag(tt[j].name,
tt[j].valueType)
+ }
+ }
+ }
+}
+
+func collectConflictTags(parts []*partWrapper) map[string]map[string]struct{} {
+ familyTagTypes :=
make(map[string]map[string]map[pbv1.ValueType]struct{})
+ for _, pw := range parts {
+ for tf, tt := range pw.p.tagType {
+ tagTypes := familyTagTypes[tf]
+ if tagTypes == nil {
+ tagTypes =
make(map[string]map[pbv1.ValueType]struct{})
+ familyTagTypes[tf] = tagTypes
+ }
+ for name, vt := range tt {
+ decoded := decodeTypedTag(name)
+ valueTypes := tagTypes[decoded]
+ if valueTypes == nil {
+ valueTypes =
make(map[pbv1.ValueType]struct{})
+ tagTypes[decoded] = valueTypes
+ }
+ valueTypes[vt] = struct{}{}
+ }
+ }
+ }
+ var conflictTags map[string]map[string]struct{}
+ for tf, tagTypes := range familyTagTypes {
+ for name, valueTypes := range tagTypes {
+ if len(valueTypes) <= 1 {
+ continue
+ }
+ if conflictTags == nil {
+ conflictTags =
make(map[string]map[string]struct{})
+ }
+ if conflictTags[tf] == nil {
+ conflictTags[tf] = make(map[string]struct{})
+ }
+ conflictTags[tf][name] = struct{}{}
+ }
+ }
+ return conflictTags
+}
+
func mergeTwoBlocks(target, left, right *blockPointer) {
appendIfEmpty := func(ib1, ib2 *blockPointer) bool {
if ib1.idx >= len(ib1.timestamps) {
diff --git a/banyand/stream/merger_test.go b/banyand/stream/merger_test.go
index 5062f8f03..627dcee7f 100644
--- a/banyand/stream/merger_test.go
+++ b/banyand/stream/merger_test.go
@@ -27,7 +27,9 @@ import (
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/stretchr/testify/require"
+ "github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/fs"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/test"
@@ -368,3 +370,187 @@ func Test_mergeParts(t *testing.T) {
})
}
}
+
+func Test_mergePartsWithConflictingTagTypes(t *testing.T) {
+ type wantTag struct {
+ value []byte
+ valueType pbv1.ValueType
+ }
+ type wantBlock struct {
+ tags map[string]wantTag
+ seriesID common.SeriesID
+ }
+ tests := []struct {
+ name string
+ esList []*elements
+ wantBlocks []wantBlock
+ }{
+ {
+ name: "two parts with same tag name and conflicting
valueTypes",
+ esList: []*elements{
+ {
+ seriesIDs: []common.SeriesID{1},
+ timestamps: []int64{1},
+ elementIDs: []uint64{1},
+ tagFamilies: [][]tagValues{
+ {
+ {tag: "searchable",
values: []*tagValue{
+ {tag:
"extra_tag", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)},
+ }},
+ },
+ },
+ },
+ {
+ seriesIDs: []common.SeriesID{2},
+ timestamps: []int64{2},
+ elementIDs: []uint64{2},
+ tagFamilies: [][]tagValues{
+ {
+ {tag: "searchable",
values: []*tagValue{
+ {tag:
"extra_tag", valueType: pbv1.ValueTypeStr, value: []byte("active")},
+ }},
+ },
+ },
+ },
+ },
+ wantBlocks: []wantBlock{
+ {
+ seriesID: 1,
+ tags: map[string]wantTag{
+ "searchable/extra_tag#int":
{valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)},
+ },
+ },
+ {
+ seriesID: 2,
+ tags: map[string]wantTag{
+ "searchable/extra_tag#str":
{valueType: pbv1.ValueTypeStr, value: []byte("active")},
+ },
+ },
+ },
+ },
+ {
+ name: "merge typed-suffix part with plain part",
+ esList: []*elements{
+ {
+ seriesIDs: []common.SeriesID{1},
+ timestamps: []int64{1},
+ elementIDs: []uint64{1},
+ tagFamilies: [][]tagValues{
+ {
+ {tag: "searchable",
values: []*tagValue{
+ {tag:
"extra_tag#int", valueType: pbv1.ValueTypeInt64, value:
convert.Int64ToBytes(100)},
+ }},
+ },
+ },
+ },
+ {
+ seriesIDs: []common.SeriesID{2},
+ timestamps: []int64{2},
+ elementIDs: []uint64{2},
+ tagFamilies: [][]tagValues{
+ {
+ {tag: "searchable",
values: []*tagValue{
+ {tag:
"extra_tag", valueType: pbv1.ValueTypeStr, value: []byte("pending")},
+ }},
+ },
+ },
+ },
+ },
+ wantBlocks: []wantBlock{
+ {
+ seriesID: 1,
+ tags: map[string]wantTag{
+ "searchable/extra_tag#int":
{valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100)},
+ },
+ },
+ {
+ seriesID: 2,
+ tags: map[string]wantTag{
+ "searchable/extra_tag#str":
{valueType: pbv1.ValueTypeStr, value: []byte("pending")},
+ },
+ },
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ tmpPath, defFn := test.Space(require.New(t))
+ defer defFn()
+ fileSystem := fs.NewLocalFileSystem()
+
+ var fpp []*partWrapper
+ defer func() {
+ for _, pw := range fpp {
+ pw.decRef()
+ }
+ }()
+ for i, es := range tt.esList {
+ mp := generateMemPart()
+ mp.mustInitFromElements(es)
+ mp.mustFlush(fileSystem, partPath(tmpPath,
uint64(i+1)))
+ filePW := newPartWrapper(nil,
mustOpenFilePart(uint64(i+1), tmpPath, fileSystem))
+ filePW.p.partMetadata.ID = uint64(i + 1)
+ fpp = append(fpp, filePW)
+ releaseMemPart(mp)
+ }
+
+ conflictTags := collectConflictTags(fpp)
+ require.NotNil(t, conflictTags, "expected conflict tags
to be collected")
+ require.Contains(t, conflictTags, "searchable")
+ require.Contains(t, conflictTags["searchable"],
"extra_tag")
+
+ closeCh := make(chan struct{})
+ defer close(closeCh)
+ tst := &tsTable{pm: protector.Nop{}}
+ merged, err := tst.mergeParts(fileSystem, closeCh, fpp,
uint64(len(tt.esList)+1), tmpPath)
+ require.NoError(t, err)
+ defer merged.decRef()
+
+ pmi := &partMergeIter{}
+ pmi.mustInitFromPart(merged.p)
+ reader := &blockReader{}
+ reader.init([]*partMergeIter{pmi})
+ decoder := generateColumnValuesDecoder()
+ defer releaseColumnValuesDecoder(decoder)
+
+ gotBlocks :=
make(map[common.SeriesID]map[string]wantTag)
+ for reader.nextBlockMetadata() {
+ reader.loadBlockData(decoder)
+ bp := reader.block
+ tt := make(map[string]wantTag)
+ for i := range bp.tagFamilies {
+ tf := &bp.tagFamilies[i]
+ for j := range tf.tags {
+ tg := &tf.tags[j]
+ require.NotEmpty(t, tg.values,
"tag %q in family %q should have values", tg.name, tf.name)
+ tt[tf.name+"/"+tg.name] =
wantTag{
+ valueType: tg.valueType,
+ value: tg.values[0],
+ }
+ }
+ }
+ gotBlocks[bp.bm.seriesID] = tt
+ }
+ require.NoError(t, reader.error())
+
+ require.Len(t, gotBlocks, len(tt.wantBlocks))
+ for _, wb := range tt.wantBlocks {
+ got, ok := gotBlocks[wb.seriesID]
+ require.True(t, ok, "missing block for seriesID
%d", wb.seriesID)
+ for key, want := range wb.tags {
+ gotTag, exists := got[key]
+ require.True(t, exists, "seriesID %d:
missing tag %q", wb.seriesID, key)
+ require.Equal(t, want.valueType,
gotTag.valueType,
+ "seriesID %d: valueType
mismatch for %q", wb.seriesID, key)
+ require.True(t,
reflect.DeepEqual(want.value, gotTag.value),
+ "seriesID %d: value mismatch
for %q: got %v, want %v",
+ wb.seriesID, key, gotTag.value,
want.value)
+ }
+ _, hasPlain := got["searchable/extra_tag"]
+ require.False(t, hasPlain,
+ "seriesID %d: plain-named conflict tag
should not exist after merge", wb.seriesID)
+ }
+ })
+ }
+}
diff --git a/banyand/stream/metadata_test.go b/banyand/stream/metadata_test.go
index f813df7b2..87b9bb0a8 100644
--- a/banyand/stream/metadata_test.go
+++ b/banyand/stream/metadata_test.go
@@ -19,6 +19,8 @@ package stream_test
import (
"context"
+ "errors"
+ "fmt"
"strconv"
"time"
@@ -403,7 +405,7 @@ var _ = Describe("Schema Change", func() {
for _, elem := range elements {
for _, tf := range elem.TagFamilies {
- if tf.Name == "searchable" {
+ if tf.Name ==
searchableTagFamily {
for _, tag := range
tf.Tags {
innerGm.Expect(tag.Key).NotTo(Equal("extra_tag"),
"deleted tag should not be returned in query results")
@@ -436,7 +438,7 @@ var _ = Describe("Schema Change", func() {
newDataCount := 0
for _, elem := range elements {
for _, tf := range elem.TagFamilies {
- if tf.Name == "searchable" {
+ if tf.Name ==
searchableTagFamily {
for _, tag := range
tf.Tags {
if tag.Key ==
"extra_tag" {
if
tag.Value.GetInt() != nil {
@@ -476,7 +478,7 @@ var _ = Describe("Schema Change", func() {
stringCount := 0
for _, elem := range elements {
for _, tf := range elem.TagFamilies {
- if tf.Name == "searchable" {
+ if tf.Name ==
searchableTagFamily {
for _, tag := range
tf.Tags {
if tag.Key ==
"extra_tag" {
switch
tag.Value.GetValue().(type) {
@@ -498,6 +500,59 @@ var _ = Describe("Schema Change", func() {
})
})
+ Context("Stream schema with changed tag type after merge", func() {
+ It("querying data should return correct values after parts with
different types are merged", func() {
+ streamName := "schema_change_tag_type_merge"
+ now := timestamp.NowMilli()
+
+ env := setupSchemaChangeStream(svcs, streamName,
groupName, streamSetupOptions{withExtraTag: true})
+ writeSchemaChangeData(svcs, streamName, groupName,
now.Add(-2*time.Hour), 5,
+ writeDataOptions{extraTag: extraTagInt})
+ changeExtraTagType(svcs, streamName, groupName)
+ writeSchemaChangeData(svcs, streamName, groupName,
now.Add(-1*time.Hour), 3,
+ writeDataOptions{extraTag: extraTagString,
traceIDPrefix: "trace_new_", elementIDOffset: 5})
+ partCountBeforeMerge, partCountErr :=
getTotalStreamPartCount(svcs, groupName)
+ Expect(partCountErr).ShouldNot(HaveOccurred())
+ Eventually(func(innerGm Gomega) int64 {
+ currentPartCount, currentPartCountErr :=
getTotalStreamPartCount(svcs, groupName)
+
innerGm.Expect(currentPartCountErr).ShouldNot(HaveOccurred())
+ return currentPartCount
+ }, flags.EventuallyTimeout).Should(BeNumerically("<",
partCountBeforeMerge))
+
+ Eventually(func(innerGm Gomega) {
+ elements := querySchemaChangeData(svcs,
streamName, groupName,
+ now.Add(-3*time.Hour), now,
+ []string{"trace_id", "service_id",
"duration", "extra_tag"}, nil)
+ innerGm.Expect(elements).To(HaveLen(8))
+
+ nullCount := 0
+ stringCount := 0
+ for _, elem := range elements {
+ for _, tf := range elem.TagFamilies {
+ if tf.Name ==
searchableTagFamily {
+ for _, tag := range
tf.Tags {
+ if tag.Key ==
"extra_tag" {
+ switch
tag.Value.GetValue().(type) {
+ case
*modelv1.TagValue_Null:
+
nullCount++
+ case
*modelv1.TagValue_Str:
+
stringCount++
+ }
+ }
+ }
+ }
+ }
+ }
+ innerGm.Expect(nullCount).To(Equal(5),
+ "old data with int type should return
null after schema changed to STRING and parts merged")
+ innerGm.Expect(stringCount).To(Equal(3),
+ "new data should have string extra_tag
values after merge")
+ }, flags.EventuallyTimeout).Should(Succeed())
+
+ env.cleanup()
+ })
+ })
+
Context("Stream schema with deleted tag in query", func() {
It("querying data should fail if the condition includes a
deleted tag", func() {
streamName := "schema_change_filter_deleted"
@@ -609,6 +664,8 @@ var _ = Describe("Schema Change", func() {
})
})
+const searchableTagFamily = "searchable"
+
type extraTagType int
const (
@@ -909,3 +966,20 @@ func querySchemaChangeData(svcs *services, name, group
string, begin, end time.T
}).WithTimeout(flags.EventuallyTimeout).Should(BeTrue())
return resp.Elements
}
+
+func getTotalStreamPartCount(svcs *services, group string) (int64, error) {
+ dataInfo, err := svcs.stream.CollectDataInfo(context.TODO(), group)
+ if err != nil {
+ return 0, fmt.Errorf("collect stream data info: %w", err)
+ }
+ if dataInfo == nil {
+ return 0, errors.New("stream data info is nil")
+ }
+ var total int64
+ for _, seg := range dataInfo.SegmentInfo {
+ for _, shard := range seg.ShardInfo {
+ total += shard.PartCount
+ }
+ }
+ return total, nil
+}
diff --git a/banyand/stream/migration_copy.go b/banyand/stream/migration_copy.go
index f1c158e48..f85b43ba5 100644
--- a/banyand/stream/migration_copy.go
+++ b/banyand/stream/migration_copy.go
@@ -767,7 +767,7 @@ func slowCopyOneStreamPart(in streamProcessPartInput,
srcPartID uint64) (streamP
bm := &bms[j]
bm.tagProjection = in.tagProjection
b := generateBlock()
- b.mustReadFrom(in.decoder, p, *bm)
+ b.mustReadFrom(in.decoder, p, *bm, nil)
liveBlocks = append(liveBlocks, b)
for k := uint64(0); k < bm.count; k++ {
diff --git a/banyand/stream/migration_element_index.go
b/banyand/stream/migration_element_index.go
index 615d92b71..3a4d426ec 100644
--- a/banyand/stream/migration_element_index.go
+++ b/banyand/stream/migration_element_index.go
@@ -561,7 +561,7 @@ func rebuildOneStreamPartIntoDocs(r rebuildOnePartInput)
(missingRows, missingSe
}
bm.tagProjection = in.tagProjection
b := generateBlock()
- b.mustReadFrom(in.decoder, p, *bm)
+ b.mustReadFrom(in.decoder, p, *bm, nil)
for k := uint64(0); k < bm.count; k++ {
appendStreamElementIndexDoc(streamElementIndexRowInput{
in: in,
diff --git a/banyand/stream/part.go b/banyand/stream/part.go
index 7fac7392f..84b94c3a0 100644
--- a/banyand/stream/part.go
+++ b/banyand/stream/part.go
@@ -63,6 +63,7 @@ type part struct {
tagFamilyMetadata map[string]fs.Reader
tagFamilies map[string]fs.Reader
tagFamilyFilter map[string]fs.Reader
+ tagType tagType
seriesMetadata fs.Reader // Optional: series metadata reader
path string
primaryBlockMetadata []primaryBlockMetadata
@@ -93,6 +94,7 @@ func (p *part) String() string {
func openMemPart(mp *memPart) *part {
var p part
p.partMetadata = mp.partMetadata
+ p.tagType = mp.tagType
p.primaryBlockMetadata =
mustReadPrimaryBlockMetadata(p.primaryBlockMetadata[:0], &mp.meta)
@@ -326,6 +328,8 @@ func mustOpenFilePart(id uint64, root string, fileSystem
fs.FileSystem) *part {
fs.CleanupLeftoverTmp(fileSystem, partPath)
p.partMetadata.mustReadMetadata(fileSystem, partPath)
p.partMetadata.ID = id
+ p.tagType = make(tagType)
+ p.tagType.mustReadTagType(fileSystem, partPath)
metaPath := path.Join(partPath, metaFilename)
pr := mustOpenReader(metaPath, fileSystem)
diff --git a/banyand/stream/part_metadata.go b/banyand/stream/part_metadata.go
index c4edcfdd3..d6dde20e0 100644
--- a/banyand/stream/part_metadata.go
+++ b/banyand/stream/part_metadata.go
@@ -186,6 +186,27 @@ func (tt tagType) unmarshal(src []byte) error {
return nil
}
+func (tt tagType) mustReadTagType(fileSystem fs.FileSystem, partPath string) {
+ tt.reset()
+ tagTypePath := filepath.Join(partPath, tagTypeFilename)
+ data, err := fileSystem.Read(tagTypePath)
+ if err != nil {
+ var fsErr *fs.FileSystemError
+ if errors.As(err, &fsErr) && fsErr.Code == fs.IsNotExistError {
+ return
+ }
+ logger.Panicf("cannot read %s: %s", tagTypePath, err)
+ return
+ }
+ if len(data) == 0 {
+ return
+ }
+ if err = tt.unmarshal(data); err != nil {
+ logger.Panicf("cannot parse %q: %s", tagTypePath, err)
+ return
+ }
+}
+
func (tt tagType) mustWriteTagType(fileSystem fs.FileSystem, partPath string) {
if len(tt) == 0 {
return
diff --git a/banyand/stream/stream_suite_test.go
b/banyand/stream/stream_suite_test.go
index 5c02d073c..5875c4571 100644
--- a/banyand/stream/stream_suite_test.go
+++ b/banyand/stream/stream_suite_test.go
@@ -97,7 +97,7 @@ func setUp() (*services, func()) {
)
rootPath, deferFunc, err := test.NewSpace()
gomega.Expect(err).NotTo(gomega.HaveOccurred())
- flags = append(flags, "--stream-root-path="+rootPath)
+ flags = append(flags, "--stream-root-path="+rootPath,
"--stream-max-merge-parts=2")
moduleDeferFunc := test.SetupModules(
flags,
pipeline,
diff --git a/banyand/stream/svc_standalone.go b/banyand/stream/svc_standalone.go
index 86172aa1d..4bf812554 100644
--- a/banyand/stream/svc_standalone.go
+++ b/banyand/stream/svc_standalone.go
@@ -180,6 +180,7 @@ func (s *standalone) FlagSet() *run.FlagSet {
flagS.DurationVar(&s.option.memWaitTimeout,
"stream-lifecycle-receive-mem-wait-timeout", 5*time.Minute,
"max time the migration receiver waits for memory to recover
before introducing an external segment")
s.option.mergePolicy = newDefaultMergePolicy()
+ flagS.IntVar(&s.option.mergePolicy.maxParts, "stream-max-merge-parts",
s.option.mergePolicy.maxParts, "the maximum number of parts to merge at once")
flagS.VarP(&s.option.mergePolicy.maxFanOutSize,
"stream-max-fan-out-size", "", "the upper bound of a single file size after
merge of stream")
s.option.seriesCacheMaxSize = run.Bytes(32 << 20)
flagS.VarP(&s.option.seriesCacheMaxSize,
"stream-series-cache-max-size", "", "the max size of series cache in each
group")
diff --git a/banyand/stream/tag.go b/banyand/stream/tag.go
index 769c67c99..31a3a9e4f 100644
--- a/banyand/stream/tag.go
+++ b/banyand/stream/tag.go
@@ -18,6 +18,8 @@
package stream
import (
+ "strings"
+
internalencoding
"github.com/apache/skywalking-banyandb/banyand/internal/encoding"
"github.com/apache/skywalking-banyandb/pkg/bytes"
"github.com/apache/skywalking-banyandb/pkg/convert"
@@ -28,6 +30,48 @@ import (
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)
+const typedTagSeparator = "#"
+
+var (
+ valueTypeToSuffix = map[pbv1.ValueType]string{
+ pbv1.ValueTypeStr: "str",
+ pbv1.ValueTypeInt64: "int",
+ pbv1.ValueTypeFloat64: "float",
+ pbv1.ValueTypeBinaryData: "bin",
+ pbv1.ValueTypeStrArr: "str_arr",
+ pbv1.ValueTypeInt64Arr: "int_arr",
+ pbv1.ValueTypeTimestamp: "ts",
+ pbv1.ValueTypeUnknown: "",
+ }
+ suffixToValueType = map[string]pbv1.ValueType{
+ "str": pbv1.ValueTypeStr,
+ "int": pbv1.ValueTypeInt64,
+ "float": pbv1.ValueTypeFloat64,
+ "bin": pbv1.ValueTypeBinaryData,
+ "str_arr": pbv1.ValueTypeStrArr,
+ "int_arr": pbv1.ValueTypeInt64Arr,
+ "ts": pbv1.ValueTypeTimestamp,
+ }
+)
+
+func encodeTypedTag(name string, vt pbv1.ValueType) string {
+ suffix, ok := valueTypeToSuffix[vt]
+ if !ok || suffix == "" {
+ return name
+ }
+ return name + typedTagSeparator + suffix
+}
+
+func decodeTypedTag(key string) string {
+ for suffix := range suffixToValueType {
+ sepSuffix := typedTagSeparator + suffix
+ if strings.HasSuffix(key, sepSuffix) {
+ return key[:len(key)-len(sepSuffix)]
+ }
+ }
+ return key
+}
+
type tag struct {
uniqueValues map[string]struct{}
min []byte
diff --git a/banyand/stream/tag_test.go b/banyand/stream/tag_test.go
index acb0e08f2..1f2e544a2 100644
--- a/banyand/stream/tag_test.go
+++ b/banyand/stream/tag_test.go
@@ -261,3 +261,22 @@ func TestTag_HighCardinalityStringEncoding(t *testing.T) {
})
}
}
+
+func TestEncodeTypedTag(t *testing.T) {
+ assert.Equal(t, "state#int", encodeTypedTag("state",
pbv1.ValueTypeInt64))
+ assert.Equal(t, "state#str", encodeTypedTag("state", pbv1.ValueTypeStr))
+ assert.Equal(t, "http.status#int", encodeTypedTag("http.status",
pbv1.ValueTypeInt64))
+ assert.Equal(t, "data#bin", encodeTypedTag("data",
pbv1.ValueTypeBinaryData))
+ assert.Equal(t, "unknown", encodeTypedTag("unknown",
pbv1.ValueTypeUnknown))
+}
+
+func TestDecodeTypedTag(t *testing.T) {
+ assert.Equal(t, "state", decodeTypedTag("state#int"))
+ assert.Equal(t, "state", decodeTypedTag("state#str"))
+ assert.Equal(t, "http.status", decodeTypedTag("http.status#int"))
+ assert.Equal(t, "http.status", decodeTypedTag("http.status"))
+ assert.Equal(t, "simple", decodeTypedTag("simple"))
+ assert.Equal(t, "my.str", decodeTypedTag("my.str"))
+ assert.Equal(t, "arr.int", decodeTypedTag("arr.int"))
+ assert.Equal(t, "data.bin", decodeTypedTag("data.bin"))
+}
