You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/04/17 02:11:18 UTC
(skywalking-banyandb) branch main updated: Fix OOM issue cause during migration when a group contains a large amount of data (#1072)
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 998ce3122 Fix OOM issue cause during migration when a group contains a large amount of data (#1072)
998ce3122 is described below
commit 998ce31226f94836234bd3d93bd25abade9c9f65
Author: mrproliu <74...@qq.com>
AuthorDate: Fri Apr 17 10:11:11 2026 +0800
Fix OOM issue cause during migration when a group contains a large amount of data (#1072)
---
CHANGES.md | 1 +
banyand/internal/sidx/block_writer.go | 31 +-
banyand/internal/sidx/file_part_sync_test.go | 134 ++++
banyand/internal/sidx/interfaces.go | 2 +
banyand/internal/sidx/metadata.go | 19 +
banyand/internal/sidx/part.go | 62 +-
banyand/internal/sidx/sidx.go | 14 +
banyand/measure/block_writer.go | 26 +-
banyand/measure/file_part_sync_test.go | 207 +++++
banyand/measure/introducer.go | 20 +-
banyand/measure/part_metadata.go | 9 +
banyand/measure/tstable.go | 27 +-
banyand/measure/write_data.go | 82 +-
banyand/stream/block_writer.go | 29 +-
banyand/stream/file_part_sync_test.go | 195 +++++
banyand/stream/introducer.go | 21 +-
banyand/stream/part_metadata.go | 9 +
banyand/stream/tstable.go | 27 +-
banyand/stream/write_data.go | 73 +-
banyand/trace/block_writer.go | 26 +-
banyand/trace/file_part_sync_test.go | 549 +++++++++++++
banyand/trace/introducer.go | 41 +-
banyand/trace/migration_test.go | 1106 ++++++++++++++++++++++++++
banyand/trace/part_metadata.go | 9 +
banyand/trace/streaming_pipeline_test.go | 8 +
banyand/trace/tstable.go | 28 +-
banyand/trace/write_data.go | 127 +--
27 files changed, 2655 insertions(+), 227 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 252694890..e960ab720 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -90,6 +90,7 @@ Release Notes.
- Fix duplicate query execution in distributed measure Agg+TopN queries by enabling push-down aggregation, removing the wasteful double-query pattern.
- Fix nil pointer panic in segment collectMetrics during shutdown.
- Fix entity tag handling in trace filter to prevent TagIdx index mismatch when filtering with both entity and non-entity tags.
+- Fix OOM issue cause during migration when a group contains a large amount of data.
### Document
diff --git a/banyand/internal/sidx/block_writer.go b/banyand/internal/sidx/block_writer.go
index 75645b4fd..607844eea 100644
--- a/banyand/internal/sidx/block_writer.go
+++ b/banyand/internal/sidx/block_writer.go
@@ -116,6 +116,21 @@ func (sw *writers) MustInitForMemPart(mp *memPart) {
sw.keysWriter.init(&mp.keys)
}
+// MustInitForFilePart initializes writers for file-backed part.
+func (sw *writers) MustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool) {
+ sw.reset()
+ fileSystem.MkdirPanicIfExist(path, storage.DirPerm)
+ sw.mustCreateTagWriters = func(name string) (fs.Writer, fs.Writer, fs.Writer) {
+ return fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagMetadataExtension), storage.FilePerm, shouldCache),
+ fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagDataExtension), storage.FilePerm, shouldCache),
+ fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagFilterExtension), storage.FilePerm, shouldCache)
+ }
+ sw.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), storage.FilePerm, shouldCache))
+ sw.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), storage.FilePerm, shouldCache))
+ sw.dataWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, dataFilename), storage.FilePerm, shouldCache))
+ sw.keysWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, keysFilename), storage.FilePerm, shouldCache))
+}
+
// totalBytesWritten returns total bytes written across all writers.
func (sw *writers) totalBytesWritten() uint64 {
n := sw.metaWriter.bytesWritten + sw.primaryWriter.bytesWritten +
@@ -244,21 +259,7 @@ func (bw *blockWriter) MustInitForMemPart(mp *memPart) {
// mustInitForFilePart initializes block writer for file part.
func (bw *blockWriter) mustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool) {
bw.reset()
- fileSystem.MkdirPanicIfExist(path, storage.DirPerm)
-
- bw.writers.mustCreateTagWriters = func(name string) (fs.Writer, fs.Writer, fs.Writer) {
- metaPath := filepath.Join(path, name+tagMetadataExtension)
- dataPath := filepath.Join(path, name+tagDataExtension)
- filterPath := filepath.Join(path, name+tagFilterExtension)
- return fs.MustCreateFile(fileSystem, metaPath, storage.FilePerm, shouldCache),
- fs.MustCreateFile(fileSystem, dataPath, storage.FilePerm, shouldCache),
- fs.MustCreateFile(fileSystem, filterPath, storage.FilePerm, shouldCache)
- }
-
- bw.writers.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), storage.FilePerm, shouldCache))
- bw.writers.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), storage.FilePerm, shouldCache))
- bw.writers.dataWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, dataFilename), storage.FilePerm, shouldCache))
- bw.writers.keysWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, keysFilename), storage.FilePerm, shouldCache))
+ bw.writers.MustInitForFilePart(fileSystem, path, shouldCache)
}
// MustWriteElements writes elements to the block writer.
diff --git a/banyand/internal/sidx/file_part_sync_test.go b/banyand/internal/sidx/file_part_sync_test.go
new file mode 100644
index 000000000..890a4dce6
--- /dev/null
+++ b/banyand/internal/sidx/file_part_sync_test.go
@@ -0,0 +1,134 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package sidx
+
+import (
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+)
+
+// TestSyncPartContext_SetForFile_WritesFiles verifies that SetForFile creates the expected
+// files on disk when data is written through the writers and then flushed.
+func TestSyncPartContext_SetForFile_WritesFiles(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ destPath := filepath.Join(tmpPath, "sidx-part-1")
+
+ spc := NewSyncPartContext()
+ spc.SetForFile("test-part", fileSystem, destPath, &queue.ChunkedSyncPartContext{}, false)
+
+ w := spc.GetWriters()
+ require.NotNil(t, w, "writers must be initialized after SetForFile")
+
+ // Write some minimal data to the core writers to produce non-empty files.
+ w.SidxMetaWriter().MustWrite([]byte("meta-data"))
+ w.SidxPrimaryWriter().MustWrite([]byte("primary-data"))
+ w.SidxDataWriter().MustWrite([]byte("data-content"))
+ w.SidxKeysWriter().MustWrite([]byte("keys-content"))
+
+ finishedPath := spc.Finish()
+ require.Equal(t, destPath, finishedPath, "Finish should return the part path")
+
+ // All core files must exist on disk.
+ for _, name := range []string{metaFilename, primaryFilename, dataFilename, keysFilename} {
+ _, err := os.Stat(filepath.Join(destPath, name))
+ require.NoError(t, err, "file %s must exist on disk after Finish", name)
+ }
+}
+
+// TestPrepareFilePart_OpensWrittenPart verifies that PrepareFilePart successfully opens a
+// previously flushed SIDX part and introduces it into the snapshot with mp == nil.
+func TestPrepareFilePart_OpensWrittenPart(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ opts := NewDefaultOptions()
+ opts.Memory = protector.NewMemory(observability.NewBypassRegistry())
+ opts.Path = tmpPath
+ opts.AvailablePartIDs = nil
+
+ sidxInstance, err := NewSIDX(fileSystem, opts)
+ require.NoError(t, err)
+ defer func() {
+ require.NoError(t, sidxInstance.Close())
+ }()
+
+ // Write data and flush it to disk to get a valid on-disk part.
+ reqs := []WriteRequest{
+ createTestWriteRequest(common.SeriesID(1), 100, "payload-1",
+ Tag{Name: testTagStatus, Value: []byte("ok"), ValueType: pbv1.ValueTypeStr}),
+ createTestWriteRequest(common.SeriesID(2), 200, "payload-2",
+ Tag{Name: testTagStatus, Value: []byte("ok"), ValueType: pbv1.ValueTypeStr}),
+ }
+ mp, convErr := sidxInstance.ConvertToMemPart(reqs, 1, nil, nil)
+ require.NoError(t, convErr)
+ require.NotNil(t, mp)
+
+ partID := uint64(1)
+ destPath := partPath(tmpPath, partID)
+ mp.MustFlush(fileSystem, destPath)
+ ReleaseMemPart(mp)
+
+ // Verify core files exist before calling PrepareFilePart.
+ for _, name := range []string{primaryFilename, dataFilename, keysFilename, metaFilename} {
+ _, statErr := os.Stat(filepath.Join(destPath, name))
+ require.NoError(t, statErr, "core file %s must exist on disk", name)
+ }
+
+ // Apply PrepareFilePart by building and replacing the snapshot.
+ transitionFn := sidxInstance.PrepareFilePart(partID, destPath)
+ cur := sidxInstance.CurrentSnapshot()
+ next := transitionFn(cur)
+ if cur != nil {
+ cur.DecRef()
+ }
+ sidxInstance.ReplaceSnapshot(next)
+
+ // Verify: snapshot contains the part with nil mp (file-backed).
+ snp := sidxInstance.CurrentSnapshot()
+ require.NotNil(t, snp, "snapshot must not be nil after PrepareFilePart")
+ defer snp.DecRef()
+
+ require.Equal(t, 1, snp.getPartCount(), "snapshot must contain exactly one part")
+
+ // The part must be file-backed (mp == nil).
+ for _, pw := range snp.parts {
+ require.Nil(t, pw.mp, "part introduced via PrepareFilePart must have nil memPart")
+ require.Equal(t, partID, pw.p.partMetadata.ID, "part ID must match")
+ }
+}
diff --git a/banyand/internal/sidx/interfaces.go b/banyand/internal/sidx/interfaces.go
index e15d545de..058c8626f 100644
--- a/banyand/internal/sidx/interfaces.go
+++ b/banyand/internal/sidx/interfaces.go
@@ -37,6 +37,8 @@ import (
type SnapshotTransactionSupport interface {
// PrepareMemPart prepares a transition for introducing a memory part.
PrepareMemPart(partID uint64, mp *MemPart) func(cur *Snapshot) *Snapshot
+ // PrepareFilePart prepares a transition for introducing a file-backed part.
+ PrepareFilePart(partID uint64, partPath string) func(cur *Snapshot) *Snapshot
// PrepareFlushed prepares a transition for introducing flushed parts.
PrepareFlushed(intro *FlusherIntroduction) func(cur *Snapshot) *Snapshot
// PrepareMerged prepares a transition for introducing merged parts.
diff --git a/banyand/internal/sidx/metadata.go b/banyand/internal/sidx/metadata.go
index 58e0c252c..40b2cf521 100644
--- a/banyand/internal/sidx/metadata.go
+++ b/banyand/internal/sidx/metadata.go
@@ -28,6 +28,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/encoding"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -61,6 +62,11 @@ func validatePartMetadata(fileSystem fs.FileSystem, partPath string) error {
return nil
}
+// ValidatePartMetadata reads and validates the manifest.json in the given part directory.
+func ValidatePartMetadata(fileSystem fs.FileSystem, partPath string) error {
+ return validatePartMetadata(fileSystem, partPath)
+}
+
// blockMetadata contains metadata for a block within a part.
type blockMetadata struct {
tagsBlocks map[string]dataBlock
@@ -142,6 +148,19 @@ func (pm *partMetadata) reset() {
pm.ID = 0
}
+func (pm *partMetadata) fillFromSyncContext(ctx *queue.ChunkedSyncPartContext) {
+ pm.CompressedSizeBytes = ctx.CompressedSizeBytes
+ pm.UncompressedSizeBytes = ctx.UncompressedSizeBytes
+ pm.TotalCount = ctx.TotalCount
+ pm.BlocksCount = ctx.BlocksCount
+ pm.MinKey = ctx.MinKey
+ pm.MaxKey = ctx.MaxKey
+ minTS := ctx.MinTimestamp
+ maxTS := ctx.MaxTimestamp
+ pm.MinTimestamp = &minTS
+ pm.MaxTimestamp = &maxTS
+}
+
// reset clears blockMetadata for reuse in object pool.
func (bm *blockMetadata) reset() {
bm.seriesID = 0
diff --git a/banyand/internal/sidx/part.go b/banyand/internal/sidx/part.go
index 98aa364af..f0372af6a 100644
--- a/banyand/internal/sidx/part.go
+++ b/banyand/internal/sidx/part.go
@@ -869,11 +869,13 @@ func partName(epoch uint64) string {
return fmt.Sprintf("%016x", epoch)
}
-// SyncPartContext manages multiple sidx memParts.
+// SyncPartContext manages a file-backed sidx part during streaming sync.
type SyncPartContext struct {
- memPart *memPart
- writers *writers
- name string
+ fileSystem fs.FileSystem
+ writers *writers
+ name string
+ partPath string
+ partMeta partMetadata
}
// NewSyncPartContext creates a new sidx part context.
@@ -881,11 +883,35 @@ func NewSyncPartContext() *SyncPartContext {
return &SyncPartContext{}
}
-// Set sets the memory part and writers.
-func (spc *SyncPartContext) Set(name string, memPart *memPart, writers *writers) {
+// SetForFile initializes file-backed writers for direct file writing.
+// The metadata fields are stored and written as manifest.json during Finish().
+func (spc *SyncPartContext) SetForFile(name string, fileSystem fs.FileSystem, partPath string, ctx *queue.ChunkedSyncPartContext, shouldCache bool) {
spc.name = name
- spc.memPart = memPart
- spc.writers = writers
+ spc.fileSystem = fileSystem
+ spc.partPath = partPath
+ spc.partMeta.fillFromSyncContext(ctx)
+
+ w := GenerateWriters()
+ w.MustInitForFilePart(fileSystem, partPath, shouldCache)
+ spc.writers = w
+}
+
+// Finish closes writers, writes manifest.json, and returns the on-disk part path.
+// It clears spc.partPath so a subsequent Close will not remove the files.
+func (spc *SyncPartContext) Finish() string {
+ if spc.writers != nil {
+ spc.writers.MustClose()
+ ReleaseWriters(spc.writers)
+ spc.writers = nil
+ }
+ if spc.fileSystem != nil && spc.partPath != "" {
+ spc.partMeta.mustWriteMetadata(spc.fileSystem, spc.partPath)
+ spc.fileSystem.SyncPath(spc.partPath)
+ }
+ p := spc.partPath
+ spc.partPath = ""
+ spc.fileSystem = nil
+ return p
}
// Name gets the name.
@@ -893,22 +919,22 @@ func (spc *SyncPartContext) Name() string {
return spc.name
}
-// GetMemPart gets the memory part.
-func (spc *SyncPartContext) GetMemPart() *MemPart {
- return spc.memPart
-}
-
// GetWriters gets the writers.
func (spc *SyncPartContext) GetWriters() *Writers {
return spc.writers
}
-// Close closes the sidx part context.
+// Close closes the sidx part context and cleans up any incomplete file parts on error.
func (spc *SyncPartContext) Close() {
- spc.writers.MustClose()
- ReleaseWriters(spc.writers)
- spc.writers = nil
- spc.memPart = nil
+ if spc.writers != nil {
+ spc.writers.MustClose()
+ ReleaseWriters(spc.writers)
+ spc.writers = nil
+ }
+ if spc.partPath != "" && spc.fileSystem != nil {
+ spc.fileSystem.MustRMAll(spc.partPath)
+ spc.partPath = ""
+ }
}
func removeExt(nameWithExt, ext string) string {
diff --git a/banyand/internal/sidx/sidx.go b/banyand/internal/sidx/sidx.go
index 910d85b4a..244ffbedd 100644
--- a/banyand/internal/sidx/sidx.go
+++ b/banyand/internal/sidx/sidx.go
@@ -298,6 +298,20 @@ func (s *sidx) PrepareMemPart(partID uint64, mp *MemPart) func(cur *Snapshot) *S
}
}
+// PrepareFilePart prepares a transition for introducing a file-backed part.
+func (s *sidx) PrepareFilePart(partID uint64, partPath string) func(cur *Snapshot) *Snapshot {
+ return func(cur *Snapshot) *Snapshot {
+ if cur == nil {
+ cur = &Snapshot{}
+ }
+ next := cur.copyAllTo()
+ part := mustOpenPart(partID, partPath, s.fileSystem)
+ pw := newPartWrapper(nil, part)
+ next.parts = append(next.parts, pw)
+ return next
+ }
+}
+
// PrepareFlushed prepares a transition for introducing flushed parts.
func (s *sidx) PrepareFlushed(intro *FlusherIntroduction) func(cur *Snapshot) *Snapshot {
return func(cur *Snapshot) *Snapshot {
diff --git a/banyand/measure/block_writer.go b/banyand/measure/block_writer.go
index 948e80572..a6238b3b0 100644
--- a/banyand/measure/block_writer.go
+++ b/banyand/measure/block_writer.go
@@ -113,6 +113,19 @@ func (sw *writers) MustClose() {
}
}
+func (sw *writers) mustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool) {
+ sw.reset()
+ fileSystem.MkdirPanicIfExist(path, storage.DirPerm)
+ sw.mustCreateTagFamilyWriters = func(name string) (fs.Writer, fs.Writer) {
+ return fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagFamiliesMetadataFilenameExt), storage.FilePerm, shouldCache),
+ fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagFamiliesFilenameExt), storage.FilePerm, shouldCache)
+ }
+ sw.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), storage.FilePerm, shouldCache))
+ sw.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), storage.FilePerm, shouldCache))
+ sw.timestampsWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, timestampsFilename), storage.FilePerm, shouldCache))
+ sw.fieldValuesWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, fieldValuesFilename), storage.FilePerm, shouldCache))
+}
+
func (sw *writers) getColumnMetadataWriterAndColumnWriter(columnName string) (*writer, *writer) {
chw, ok := sw.tagFamilyMetadataWriters[columnName]
cw := sw.tagFamilyWriters[columnName]
@@ -176,18 +189,7 @@ func (bw *blockWriter) MustInitForMemPart(mp *memPart) {
func (bw *blockWriter) mustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool) {
bw.reset()
- fileSystem.MkdirPanicIfExist(path, storage.DirPerm)
- bw.writers.mustCreateTagFamilyWriters = func(name string) (fs.Writer, fs.Writer) {
- metaPath := filepath.Join(path, name+tagFamiliesMetadataFilenameExt)
- dataPath := filepath.Join(path, name+tagFamiliesFilenameExt)
- return fs.MustCreateFile(fileSystem, metaPath, storage.FilePerm, shouldCache),
- fs.MustCreateFile(fileSystem, dataPath, storage.FilePerm, shouldCache)
- }
-
- bw.writers.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), storage.FilePerm, shouldCache))
- bw.writers.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), storage.FilePerm, shouldCache))
- bw.writers.timestampsWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, timestampsFilename), storage.FilePerm, shouldCache))
- bw.writers.fieldValuesWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, fieldValuesFilename), storage.FilePerm, shouldCache))
+ bw.writers.mustInitForFilePart(fileSystem, path, shouldCache)
}
func (bw *blockWriter) MustWriteDataPoints(sid common.SeriesID, timestamps, versions []int64, tagFamilies [][]nameValues, fields []nameValues) {
diff --git a/banyand/measure/file_part_sync_test.go b/banyand/measure/file_part_sync_test.go
new file mode 100644
index 000000000..eabe55627
--- /dev/null
+++ b/banyand/measure/file_part_sync_test.go
@@ -0,0 +1,207 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package measure
+
+import (
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+)
+
+func buildAndFlushMeasureMemPart(t *testing.T, fileSystem fs.FileSystem, dps *dataPoints, destPath string) {
+ t.Helper()
+ mp := generateMemPart()
+ mp.mustInitFromDataPoints(dps)
+ mp.mustFlush(fileSystem, destPath)
+ releaseMemPart(mp)
+}
+
+func makeTestDataPoints(now int64) *dataPoints {
+ return &dataPoints{
+ seriesIDs: []common.SeriesID{1, 2},
+ timestamps: []int64{now, now + 1000},
+ versions: []int64{1, 2},
+ tagFamilies: [][]nameValues{
+ {
+ {
+ name: "singleTag",
+ values: []*nameValue{
+ {name: "strTag", valueType: pbv1.ValueTypeStr, value: []byte("val1"), valueArr: nil},
+ },
+ },
+ },
+ {
+ {
+ name: "singleTag",
+ values: []*nameValue{
+ {name: "strTag", valueType: pbv1.ValueTypeStr, value: []byte("val2"), valueArr: nil},
+ },
+ },
+ },
+ },
+ fields: []nameValues{
+ {
+ name: "intField",
+ values: []*nameValue{
+ {name: "intField", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(100), valueArr: nil},
+ },
+ },
+ {
+ name: "intField",
+ values: []*nameValue{
+ {name: "intField", valueType: pbv1.ValueTypeInt64, value: convert.Int64ToBytes(200), valueArr: nil},
+ },
+ },
+ },
+ }
+}
+
+// TestMustAddFilePart_FilesOnDisk verifies that mustAddFilePart introduces a file-backed part
+// into the snapshot with mp == nil.
+func TestMustAddFilePart_FilesOnDisk(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _ := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), testSnapshotOption(), nil)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ now := int64(1_000_000_000)
+ dps := makeTestDataPoints(now)
+
+ partID := uint64(1)
+ destPath := partPath(tabDir, partID)
+ buildAndFlushMeasureMemPart(t, fileSystem, dps, destPath)
+
+ tst.mustAddFilePart(partID)
+
+ snp := tst.currentSnapshot()
+ require.NotNil(t, snp, "snapshot must not be nil after mustAddFilePart")
+ defer snp.decRef()
+
+ require.Len(t, snp.parts, 1, "snapshot must contain exactly one part")
+ require.Equal(t, partID, snp.parts[0].p.partMetadata.ID, "part ID must match")
+ require.Nil(t, snp.parts[0].mp, "file-backed part must have nil memPart")
+}
+
+// TestIntroduceMemPart_NilMpGuard verifies that mustAddFilePart does not panic when
+// introducing a file-backed part (mp == nil) through the introducer.
+func TestIntroduceMemPart_NilMpGuard(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _ := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), testSnapshotOption(), nil)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ now := int64(2_000_000_000)
+ dps := makeTestDataPoints(now)
+
+ partID := uint64(7)
+ destPath := partPath(tabDir, partID)
+ buildAndFlushMeasureMemPart(t, fileSystem, dps, destPath)
+
+ require.NotPanics(t, func() {
+ tst.mustAddFilePart(partID)
+ })
+
+ snp := tst.currentSnapshot()
+ require.NotNil(t, snp)
+ defer snp.decRef()
+ require.Len(t, snp.parts, 1)
+ require.Nil(t, snp.parts[0].mp, "file-backed part must have nil memPart")
+}
+
+// TestFilePart_ErrorCleanup verifies that when a syncPartContext is closed without calling
+// FinishSync, the incomplete part directory is removed from disk.
+func TestFilePart_ErrorCleanup(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _ := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), testSnapshotOption(), nil)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ partID := uint64(300)
+ destPath := partPath(tabDir, partID)
+ fileSystem.MkdirPanicIfExist(destPath, storage.DirPerm)
+
+ spc := &syncPartContext{
+ tsTable: tst,
+ fileSystem: fileSystem,
+ partPath: destPath,
+ partID: partID,
+ }
+ w := generateWriters()
+ w.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, metaFilename), storage.FilePerm, false))
+ w.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, primaryFilename), storage.FilePerm, false))
+ w.timestampsWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, timestampsFilename), storage.FilePerm, false))
+ w.fieldValuesWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, fieldValuesFilename), storage.FilePerm, false))
+ spc.writers = w
+
+ spc.writers.metaWriter.MustWrite([]byte("partial"))
+
+ _, statErr := os.Stat(destPath)
+ require.NoError(t, statErr, "part directory must exist before Close")
+
+ require.NoError(t, spc.Close())
+
+ _, statErr = os.Stat(destPath)
+ require.True(t, os.IsNotExist(statErr), "incomplete part directory must be removed after Close without FinishSync")
+
+ snp := tst.currentSnapshot()
+ if snp != nil {
+ defer snp.decRef()
+ require.Empty(t, snp.parts, "no parts must be in snapshot after error cleanup")
+ }
+}
diff --git a/banyand/measure/introducer.go b/banyand/measure/introducer.go
index 8aed38c3d..92dc02c4e 100644
--- a/banyand/measure/introducer.go
+++ b/banyand/measure/introducer.go
@@ -23,12 +23,12 @@ import (
)
type introduction struct {
- memPart *partWrapper
+ part *partWrapper
applied chan struct{}
}
func (i *introduction) reset() {
- i.memPart = nil
+ i.part = nil
i.applied = nil
}
@@ -120,7 +120,7 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
return
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
- tst.introduceMemPart(next, epoch)
+ tst.introducePart(next, epoch)
tst.incTotalIntroduceLoopFinished("mem")
epoch++
case next := <-flushCh:
@@ -154,7 +154,7 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, me
return
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
- tst.introduceMemPart(next, epoch)
+ tst.introducePart(next, epoch)
tst.incTotalIntroduceLoopFinished("mem")
epoch++
case next := <-flushCh:
@@ -183,7 +183,7 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, me
}
}
-func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint64) {
+func (tst *tsTable) introducePart(nextIntroduction *introduction, epoch uint64) {
cur := tst.currentSnapshot()
if cur != nil {
defer cur.decRef()
@@ -191,12 +191,16 @@ func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint6
cur = new(snapshot)
}
- next := nextIntroduction.memPart
- tst.addPendingDataCount(-int64(next.mp.partMetadata.TotalCount))
+ next := nextIntroduction.part
+ if next.mp != nil {
+ tst.addPendingDataCount(-int64(next.mp.partMetadata.TotalCount))
+ }
nextSnp := cur.copyAllTo(epoch)
nextSnp.parts = append(nextSnp.parts, next)
nextSnp.creator = snapshotCreatorMemPart
- tst.replaceSnapshot(&nextSnp, false)
+ // Persist snapshot for file-backed introductions so the part survives restart.
+ // memPart introductions skip persist because the flusher persists after mem→file conversion.
+ tst.replaceSnapshot(&nextSnp, next.mp == nil)
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
}
diff --git a/banyand/measure/part_metadata.go b/banyand/measure/part_metadata.go
index dec52f160..dbd41b5f6 100644
--- a/banyand/measure/part_metadata.go
+++ b/banyand/measure/part_metadata.go
@@ -49,6 +49,15 @@ func (pm *partMetadata) reset() {
pm.ID = 0
}
+func (pm *partMetadata) fillFromSyncContext(ctx *queue.ChunkedSyncPartContext) {
+ pm.CompressedSizeBytes = ctx.CompressedSizeBytes
+ pm.UncompressedSizeBytes = ctx.UncompressedSizeBytes
+ pm.TotalCount = ctx.TotalCount
+ pm.BlocksCount = ctx.BlocksCount
+ pm.MinTimestamp = ctx.MinTimestamp
+ pm.MaxTimestamp = ctx.MaxTimestamp
+}
+
func validatePartMetadata(fileSystem fs.FileSystem, partPath string) error {
metadataPath := filepath.Join(partPath, metadataFilename)
metadata, err := fileSystem.Read(metadataPath)
diff --git a/banyand/measure/tstable.go b/banyand/measure/tstable.go
index c9054d541..6c3fff58e 100644
--- a/banyand/measure/tstable.go
+++ b/banyand/measure/tstable.go
@@ -349,14 +349,35 @@ func (tst *tsTable) mustAddDataPointsWithSegmentID(dps *dataPoints, segmentID in
tst.mustAddMemPart(mp)
}
+func (tst *tsTable) mustAddFilePart(partID uint64) {
+ p := mustOpenFilePart(partID, tst.root, tst.fileSystem)
+ p.partMetadata.ID = partID
+
+ ind := generateIntroduction()
+ defer releaseIntroduction(ind)
+ ind.applied = make(chan struct{})
+ ind.part = newPartWrapper(nil, p)
+
+ select {
+ case tst.introductions <- ind:
+ case <-tst.loopCloser.CloseNotify():
+ ind.part.decRef()
+ return
+ }
+ select {
+ case <-ind.applied:
+ case <-tst.loopCloser.CloseNotify():
+ }
+}
+
func (tst *tsTable) mustAddMemPart(mp *memPart) {
p := openMemPart(mp)
ind := generateIntroduction()
defer releaseIntroduction(ind)
ind.applied = make(chan struct{})
- ind.memPart = newPartWrapper(mp, p)
- ind.memPart.p.partMetadata.ID = atomic.AddUint64(&tst.curPartID, 1)
+ ind.part = newPartWrapper(mp, p)
+ ind.part.p.partMetadata.ID = atomic.AddUint64(&tst.curPartID, 1)
startTime := time.Now()
totalCount := mp.partMetadata.TotalCount
tst.addPendingDataCount(int64(totalCount))
@@ -364,7 +385,7 @@ func (tst *tsTable) mustAddMemPart(mp *memPart) {
case tst.introductions <- ind:
case <-tst.loopCloser.CloseNotify():
tst.addPendingDataCount(-int64(totalCount))
- ind.memPart.decRef()
+ ind.part.decRef()
return
}
select {
diff --git a/banyand/measure/write_data.go b/banyand/measure/write_data.go
index cc5b855b7..2f848f111 100644
--- a/banyand/measure/write_data.go
+++ b/banyand/measure/write_data.go
@@ -20,21 +20,26 @@ package measure
import (
"fmt"
"strings"
+ "sync/atomic"
"time"
"github.com/apache/skywalking-banyandb/api/common"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/pool"
)
type syncPartContext struct {
- tsTable *tsTable
- writers *writers
- memPart *memPart
+ tsTable *tsTable
+ fileSystem fs.FileSystem
+ writers *writers
+ partPath string
+ partMeta partMetadata
+ partID uint64
}
func (s *syncPartContext) NewPartType(_ *queue.ChunkedSyncPartContext) error {
@@ -43,23 +48,36 @@ func (s *syncPartContext) NewPartType(_ *queue.ChunkedSyncPartContext) error {
}
func (s *syncPartContext) FinishSync() error {
- mp := s.memPart
- s.memPart = nil
- s.tsTable.mustAddMemPart(mp)
+ // Close writers first so file data is flushed before we write metadata.
+ s.releaseCoreWriters()
+
+ s.partMeta.mustWriteMetadata(s.fileSystem, s.partPath)
+ s.fileSystem.SyncPath(s.partPath)
+
+ s.tsTable.mustAddFilePart(s.partID)
+ s.partPath = ""
return s.Close()
}
-func (s *syncPartContext) Close() error {
+// releaseCoreWriters closes and releases the core writers, if any. Safe to call multiple times.
+func (s *syncPartContext) releaseCoreWriters() {
+ if s.writers == nil {
+ return
+ }
s.writers.MustClose()
releaseWriters(s.writers)
s.writers = nil
- if s.memPart != nil {
- // syncPartContext owns the memPart directly without partWrapper refcounting.
- // It must release via releaseMemPart, not decRef, which is used in mustAddMemPart.
- releaseMemPart(s.memPart)
- s.memPart = nil
+}
+
+func (s *syncPartContext) Close() error {
+ s.releaseCoreWriters()
+ // Clean up incomplete filePart on error (partPath is cleared on success in FinishSync)
+ if s.partPath != "" && s.fileSystem != nil {
+ s.fileSystem.MustRMAll(s.partPath)
+ s.partPath = ""
}
s.tsTable = nil
+ s.fileSystem = nil
return nil
}
@@ -103,20 +121,23 @@ func (s *syncCallback) CreatePartHandler(ctx *queue.ChunkedSyncPartContext) (que
}
tsdb.Tick(ctx.MaxTimestamp)
- memPart := generateMemPart()
- memPart.partMetadata.CompressedSizeBytes = ctx.CompressedSizeBytes
- memPart.partMetadata.UncompressedSizeBytes = ctx.UncompressedSizeBytes
- memPart.partMetadata.TotalCount = ctx.TotalCount
- memPart.partMetadata.BlocksCount = ctx.BlocksCount
- memPart.partMetadata.MinTimestamp = ctx.MinTimestamp
- memPart.partMetadata.MaxTimestamp = ctx.MaxTimestamp
- writers := generateWriters()
- writers.mustInitForMemPart(memPart)
- return &syncPartContext{
- tsTable: tsTable,
- writers: writers,
- memPart: memPart,
- }, nil
+
+ partID := atomic.AddUint64(&tsTable.curPartID, 1)
+ pp := partPath(tsTable.root, partID)
+ fileSystem := tsTable.fileSystem
+
+ w := generateWriters()
+ w.mustInitForFilePart(fileSystem, pp, tsTable.pm.ShouldCache(int64(ctx.CompressedSizeBytes)))
+
+ partCtx := &syncPartContext{
+ tsTable: tsTable,
+ fileSystem: fileSystem,
+ writers: w,
+ partPath: pp,
+ partID: partID,
+ }
+ partCtx.partMeta.fillFromSyncContext(ctx)
+ return partCtx, nil
}
// HandleFileChunk implements queue.ChunkedSyncHandler for streaming file chunks.
@@ -180,15 +201,6 @@ func releaseWriters(sw *writers) {
var writersPool = pool.Register[*writers]("measure-writers")
-func (sw *writers) mustInitForMemPart(mp *memPart) {
- sw.reset()
- sw.mustCreateTagFamilyWriters = mp.mustCreateMemTagFamilyWriters
- sw.metaWriter.init(&mp.meta)
- sw.primaryWriter.init(&mp.primary)
- sw.timestampsWriter.init(&mp.timestamps)
- sw.fieldValuesWriter.init(&mp.fieldValues)
-}
-
type syncSeriesContext struct {
streamer index.ExternalSegmentStreamer
segment storage.Segment[*tsTable, *commonv1.ResourceOpts]
diff --git a/banyand/stream/block_writer.go b/banyand/stream/block_writer.go
index 48e352c83..fdc77df25 100644
--- a/banyand/stream/block_writer.go
+++ b/banyand/stream/block_writer.go
@@ -89,12 +89,17 @@ func (sw *writers) reset() {
}
}
-func (sw *writers) mustInitForMemPart(mp *memPart) {
+func (sw *writers) mustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool) {
sw.reset()
- sw.mustCreateTagFamilyWriters = mp.mustCreateMemTagFamilyWriters
- sw.metaWriter.init(&mp.meta)
- sw.primaryWriter.init(&mp.primary)
- sw.timestampsWriter.init(&mp.timestamps)
+ fileSystem.MkdirPanicIfExist(path, storage.DirPerm)
+ sw.mustCreateTagFamilyWriters = func(name string) (fs.Writer, fs.Writer, fs.Writer) {
+ return fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagFamiliesMetadataFilenameExt), storage.FilePerm, shouldCache),
+ fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagFamiliesFilenameExt), storage.FilePerm, shouldCache),
+ fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagFamiliesFilterFilenameExt), storage.FilePerm, shouldCache)
+ }
+ sw.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), storage.FilePerm, shouldCache))
+ sw.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), storage.FilePerm, shouldCache))
+ sw.timestampsWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, timestampsFilename), storage.FilePerm, shouldCache))
}
func (sw *writers) totalBytesWritten() uint64 {
@@ -194,19 +199,7 @@ func (bw *blockWriter) MustInitForMemPart(mp *memPart) {
func (bw *blockWriter) mustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool) {
bw.reset()
- fileSystem.MkdirPanicIfExist(path, storage.DirPerm)
- bw.writers.mustCreateTagFamilyWriters = func(name string) (fs.Writer, fs.Writer, fs.Writer) {
- metaPath := filepath.Join(path, name+tagFamiliesMetadataFilenameExt)
- dataPath := filepath.Join(path, name+tagFamiliesFilenameExt)
- fitlerPath := filepath.Join(path, name+tagFamiliesFilterFilenameExt)
- return fs.MustCreateFile(fileSystem, metaPath, storage.FilePerm, shouldCache),
- fs.MustCreateFile(fileSystem, dataPath, storage.FilePerm, shouldCache),
- fs.MustCreateFile(fileSystem, fitlerPath, storage.FilePerm, shouldCache)
- }
-
- bw.writers.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), storage.FilePerm, shouldCache))
- bw.writers.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), storage.FilePerm, shouldCache))
- bw.writers.timestampsWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, timestampsFilename), storage.FilePerm, shouldCache))
+ bw.writers.mustInitForFilePart(fileSystem, path, shouldCache)
}
func (bw *blockWriter) MustWriteElements(sid common.SeriesID, timestamps []int64, elementIDs []uint64, tagFamilies [][]tagValues) {
diff --git a/banyand/stream/file_part_sync_test.go b/banyand/stream/file_part_sync_test.go
new file mode 100644
index 000000000..bc1c47076
--- /dev/null
+++ b/banyand/stream/file_part_sync_test.go
@@ -0,0 +1,195 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package stream
+
+import (
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+)
+
+func buildAndFlushStreamMemPart(t *testing.T, fileSystem fs.FileSystem, es *elements, destPath string) {
+ t.Helper()
+ mp := generateMemPart()
+ mp.mustInitFromElements(es)
+ mp.mustFlush(fileSystem, destPath)
+ releaseMemPart(mp)
+}
+
+func makeTestElements(now int64) *elements {
+ return &elements{
+ seriesIDs: []common.SeriesID{1, 2},
+ timestamps: []int64{now, now + 1000},
+ elementIDs: []uint64{11, 21},
+ tagFamilies: [][]tagValues{
+ {
+ {
+ tag: "singleTag",
+ values: []*tagValue{
+ {tag: "strTag", valueType: pbv1.ValueTypeStr, value: []byte("val1"), valueArr: nil},
+ },
+ },
+ },
+ {
+ {
+ tag: "singleTag",
+ values: []*tagValue{
+ {tag: "strTag", valueType: pbv1.ValueTypeStr, value: convert.Int64ToBytes(200), valueArr: nil},
+ },
+ },
+ },
+ },
+ }
+}
+
+// TestMustAddFilePart_FilesOnDisk verifies that mustAddFilePart introduces a file-backed part
+// into the snapshot with mp == nil.
+func TestMustAddFilePart_FilesOnDisk(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _, initErr := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), streamSnapshotOption(), nil, false)
+ require.NoError(t, initErr)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ now := int64(1_000_000_000)
+ es := makeTestElements(now)
+
+ partID := uint64(1)
+ destPath := partPath(tabDir, partID)
+ buildAndFlushStreamMemPart(t, fileSystem, es, destPath)
+
+ tst.mustAddFilePart(partID)
+
+ snp := tst.currentSnapshot()
+ require.NotNil(t, snp, "snapshot must not be nil after mustAddFilePart")
+ defer snp.decRef()
+
+ require.Len(t, snp.parts, 1, "snapshot must contain exactly one part")
+ require.Equal(t, partID, snp.parts[0].p.partMetadata.ID, "part ID must match")
+ require.Nil(t, snp.parts[0].mp, "file-backed part must have nil memPart")
+}
+
+// TestIntroduceMemPart_NilMpGuard verifies that mustAddFilePart does not panic when
+// introducing a file-backed part (mp == nil) through the introducer.
+func TestIntroduceMemPart_NilMpGuard(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _, initErr := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), streamSnapshotOption(), nil, false)
+ require.NoError(t, initErr)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ now := int64(2_000_000_000)
+ es := makeTestElements(now)
+
+ partID := uint64(7)
+ destPath := partPath(tabDir, partID)
+ buildAndFlushStreamMemPart(t, fileSystem, es, destPath)
+
+ require.NotPanics(t, func() {
+ tst.mustAddFilePart(partID)
+ })
+
+ snp := tst.currentSnapshot()
+ require.NotNil(t, snp)
+ defer snp.decRef()
+ require.Len(t, snp.parts, 1)
+ require.Nil(t, snp.parts[0].mp, "file-backed part must have nil memPart")
+}
+
+// TestFilePart_ErrorCleanup verifies that when a syncPartContext is closed without calling
+// FinishSync, the incomplete part directory is removed from disk.
+func TestFilePart_ErrorCleanup(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _, initErr := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), streamSnapshotOption(), nil, false)
+ require.NoError(t, initErr)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ partID := uint64(300)
+ destPath := partPath(tabDir, partID)
+ fileSystem.MkdirPanicIfExist(destPath, storage.DirPerm)
+
+ spc := &syncPartContext{
+ tsTable: tst,
+ fileSystem: fileSystem,
+ partPath: destPath,
+ partID: partID,
+ }
+ w := generateWriters()
+ w.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, metaFilename), storage.FilePerm, false))
+ w.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, primaryFilename), storage.FilePerm, false))
+ w.timestampsWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, timestampsFilename), storage.FilePerm, false))
+ spc.writers = w
+
+ spc.writers.metaWriter.MustWrite([]byte("partial"))
+
+ _, statErr := os.Stat(destPath)
+ require.NoError(t, statErr, "part directory must exist before Close")
+
+ require.NoError(t, spc.Close())
+
+ _, statErr = os.Stat(destPath)
+ require.True(t, os.IsNotExist(statErr), "incomplete part directory must be removed after Close without FinishSync")
+
+ snp := tst.currentSnapshot()
+ if snp != nil {
+ defer snp.decRef()
+ require.Empty(t, snp.parts, "no parts must be in snapshot after error cleanup")
+ }
+}
diff --git a/banyand/stream/introducer.go b/banyand/stream/introducer.go
index a9dda7460..e2255578d 100644
--- a/banyand/stream/introducer.go
+++ b/banyand/stream/introducer.go
@@ -23,12 +23,12 @@ import (
)
type introduction struct {
- memPart *partWrapper
+ part *partWrapper
applied chan struct{}
}
func (i *introduction) reset() {
- i.memPart = nil
+ i.part = nil
i.applied = nil
}
@@ -149,7 +149,7 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
return
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
- tst.introduceMemPart(next, epoch)
+ tst.introducePart(next, epoch)
tst.incTotalIntroduceLoopFinished("mem")
epoch++
case next := <-flushCh:
@@ -183,7 +183,7 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, me
return
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
- tst.introduceMemPart(next, epoch)
+ tst.introducePart(next, epoch)
tst.incTotalIntroduceLoopFinished("mem")
epoch++
case next := <-flushCh:
@@ -212,7 +212,7 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, me
}
}
-func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint64) {
+func (tst *tsTable) introducePart(nextIntroduction *introduction, epoch uint64) {
cur := tst.currentSnapshot()
if cur != nil {
defer cur.decRef()
@@ -220,12 +220,19 @@ func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint6
cur = new(snapshot)
}
- next := nextIntroduction.memPart
- tst.addPendingDataCount(-int64(next.mp.partMetadata.TotalCount))
+ next := nextIntroduction.part
+ if next.mp != nil {
+ tst.addPendingDataCount(-int64(next.mp.partMetadata.TotalCount))
+ }
nextSnp := cur.copyAllTo(epoch)
nextSnp.parts = append(nextSnp.parts, next)
nextSnp.creator = snapshotCreatorMemPart
tst.replaceSnapshot(&nextSnp)
+ // Persist snapshot for file-backed introductions so the part survives restart.
+ // memPart introductions skip persist because the flusher persists after mem→file conversion.
+ if next.mp == nil {
+ tst.persistSnapshot(&nextSnp)
+ }
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
}
diff --git a/banyand/stream/part_metadata.go b/banyand/stream/part_metadata.go
index 983861d5f..25c0eaa2f 100644
--- a/banyand/stream/part_metadata.go
+++ b/banyand/stream/part_metadata.go
@@ -49,6 +49,15 @@ func (pm *partMetadata) reset() {
pm.ID = 0
}
+func (pm *partMetadata) fillFromSyncContext(ctx *queue.ChunkedSyncPartContext) {
+ pm.CompressedSizeBytes = ctx.CompressedSizeBytes
+ pm.UncompressedSizeBytes = ctx.UncompressedSizeBytes
+ pm.TotalCount = ctx.TotalCount
+ pm.BlocksCount = ctx.BlocksCount
+ pm.MinTimestamp = ctx.MinTimestamp
+ pm.MaxTimestamp = ctx.MaxTimestamp
+}
+
func validatePartMetadata(fileSystem fs.FileSystem, partPath string) error {
metadataPath := filepath.Join(partPath, metadataFilename)
metadata, err := fileSystem.Read(metadataPath)
diff --git a/banyand/stream/tstable.go b/banyand/stream/tstable.go
index 60ef8dcee..9f3f5e7a0 100644
--- a/banyand/stream/tstable.go
+++ b/banyand/stream/tstable.go
@@ -360,14 +360,35 @@ func (tst *tsTable) Close() error {
return nil
}
+func (tst *tsTable) mustAddFilePart(partID uint64) {
+ p := mustOpenFilePart(partID, tst.root, tst.fileSystem)
+ p.partMetadata.ID = partID
+
+ ind := generateIntroduction()
+ defer releaseIntroduction(ind)
+ ind.applied = make(chan struct{})
+ ind.part = newPartWrapper(nil, p)
+
+ select {
+ case tst.introductions <- ind:
+ case <-tst.loopCloser.CloseNotify():
+ ind.part.decRef()
+ return
+ }
+ select {
+ case <-ind.applied:
+ case <-tst.loopCloser.CloseNotify():
+ }
+}
+
func (tst *tsTable) mustAddMemPart(mp *memPart) {
p := openMemPart(mp)
ind := generateIntroduction()
defer releaseIntroduction(ind)
ind.applied = make(chan struct{})
- ind.memPart = newPartWrapper(mp, p)
- ind.memPart.p.partMetadata.ID = atomic.AddUint64(&tst.curPartID, 1)
+ ind.part = newPartWrapper(mp, p)
+ ind.part.p.partMetadata.ID = atomic.AddUint64(&tst.curPartID, 1)
startTime := time.Now()
totalCount := mp.partMetadata.TotalCount
tst.addPendingDataCount(int64(totalCount))
@@ -375,7 +396,7 @@ func (tst *tsTable) mustAddMemPart(mp *memPart) {
case tst.introductions <- ind:
case <-tst.loopCloser.CloseNotify():
tst.addPendingDataCount(-int64(totalCount))
- ind.memPart.decRef()
+ ind.part.decRef()
return
}
select {
diff --git a/banyand/stream/write_data.go b/banyand/stream/write_data.go
index 1eeaadded..77a0b07c5 100644
--- a/banyand/stream/write_data.go
+++ b/banyand/stream/write_data.go
@@ -20,20 +20,25 @@ package stream
import (
"fmt"
"strings"
+ "sync/atomic"
"time"
"github.com/apache/skywalking-banyandb/api/common"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
type syncPartContext struct {
- tsTable *tsTable
- writers *writers
- memPart *memPart
+ tsTable *tsTable
+ fileSystem fs.FileSystem
+ writers *writers
+ partPath string
+ partMeta partMetadata
+ partID uint64
}
func (s *syncPartContext) NewPartType(_ *queue.ChunkedSyncPartContext) error {
@@ -42,23 +47,36 @@ func (s *syncPartContext) NewPartType(_ *queue.ChunkedSyncPartContext) error {
}
func (s *syncPartContext) FinishSync() error {
- mp := s.memPart
- s.memPart = nil
- s.tsTable.mustAddMemPart(mp)
+ // Close writers first so file data is flushed before we write metadata.
+ s.releaseCoreWriters()
+
+ s.partMeta.mustWriteMetadata(s.fileSystem, s.partPath)
+ s.fileSystem.SyncPath(s.partPath)
+
+ s.tsTable.mustAddFilePart(s.partID)
+ s.partPath = ""
return s.Close()
}
-func (s *syncPartContext) Close() error {
+// releaseCoreWriters closes and releases the core writers, if any. Safe to call multiple times.
+func (s *syncPartContext) releaseCoreWriters() {
+ if s.writers == nil {
+ return
+ }
s.writers.MustClose()
releaseWriters(s.writers)
s.writers = nil
- if s.memPart != nil {
- // syncPartContext owns the memPart directly without partWrapper refcounting.
- // It must release via releaseMemPart, not decRef, which is used in mustAddMemPart.
- releaseMemPart(s.memPart)
- s.memPart = nil
+}
+
+func (s *syncPartContext) Close() error {
+ s.releaseCoreWriters()
+ // Clean up incomplete filePart on error (partPath is cleared on success in FinishSync)
+ if s.partPath != "" && s.fileSystem != nil {
+ s.fileSystem.MustRMAll(s.partPath)
+ s.partPath = ""
}
s.tsTable = nil
+ s.fileSystem = nil
return nil
}
@@ -102,20 +120,23 @@ func (s *syncCallback) CreatePartHandler(ctx *queue.ChunkedSyncPartContext) (que
}
tsdb.Tick(ctx.MaxTimestamp)
- memPart := generateMemPart()
- memPart.partMetadata.CompressedSizeBytes = ctx.CompressedSizeBytes
- memPart.partMetadata.UncompressedSizeBytes = ctx.UncompressedSizeBytes
- memPart.partMetadata.TotalCount = ctx.TotalCount
- memPart.partMetadata.BlocksCount = ctx.BlocksCount
- memPart.partMetadata.MinTimestamp = ctx.MinTimestamp
- memPart.partMetadata.MaxTimestamp = ctx.MaxTimestamp
- writers := generateWriters()
- writers.mustInitForMemPart(memPart)
- return &syncPartContext{
- tsTable: tsTable,
- writers: writers,
- memPart: memPart,
- }, nil
+
+ partID := atomic.AddUint64(&tsTable.curPartID, 1)
+ pp := partPath(tsTable.root, partID)
+ fileSystem := tsTable.fileSystem
+
+ w := generateWriters()
+ w.mustInitForFilePart(fileSystem, pp, tsTable.pm.ShouldCache(int64(ctx.CompressedSizeBytes)))
+
+ partCtx := &syncPartContext{
+ tsTable: tsTable,
+ fileSystem: fileSystem,
+ writers: w,
+ partPath: pp,
+ partID: partID,
+ }
+ partCtx.partMeta.fillFromSyncContext(ctx)
+ return partCtx, nil
}
// HandleFileChunk implements queue.ChunkedSyncHandler for streaming file chunks.
diff --git a/banyand/trace/block_writer.go b/banyand/trace/block_writer.go
index 46aef8c49..c4267dd58 100644
--- a/banyand/trace/block_writer.go
+++ b/banyand/trace/block_writer.go
@@ -85,12 +85,16 @@ func (sw *writers) reset() {
}
}
-func (sw *writers) mustInitForMemPart(mp *memPart) {
+func (sw *writers) mustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool) {
sw.reset()
- sw.mustCreateTagWriters = mp.mustCreateMemTagWriters
- sw.metaWriter.init(&mp.meta)
- sw.primaryWriter.init(&mp.primary)
- sw.spanWriter.init(&mp.spans)
+ fileSystem.MkdirPanicIfExist(path, storage.DirPerm)
+ sw.mustCreateTagWriters = func(name string) (fs.Writer, fs.Writer) {
+ return fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagsMetadataFilenameExt), storage.FilePerm, shouldCache),
+ fs.MustCreateFile(fileSystem, filepath.Join(path, name+tagsFilenameExt), storage.FilePerm, shouldCache)
+ }
+ sw.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), storage.FilePerm, shouldCache))
+ sw.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), storage.FilePerm, shouldCache))
+ sw.spanWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, spansFilename), storage.FilePerm, shouldCache))
}
func (sw *writers) totalBytesWritten() uint64 {
@@ -198,17 +202,7 @@ func (bw *blockWriter) MustInitForMemPart(mp *memPart, traceSize int) {
func (bw *blockWriter) mustInitForFilePart(fileSystem fs.FileSystem, path string, shouldCache bool, traceSize int) {
bw.reset()
- fileSystem.MkdirPanicIfExist(path, storage.DirPerm)
- bw.writers.mustCreateTagWriters = func(name string) (fs.Writer, fs.Writer) {
- metaPath := filepath.Join(path, name+tagsMetadataFilenameExt)
- dataPath := filepath.Join(path, name+tagsFilenameExt)
- return fs.MustCreateFile(fileSystem, metaPath, storage.FilePerm, shouldCache),
- fs.MustCreateFile(fileSystem, dataPath, storage.FilePerm, shouldCache)
- }
-
- bw.writers.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, metaFilename), storage.FilePerm, shouldCache))
- bw.writers.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, primaryFilename), storage.FilePerm, shouldCache))
- bw.writers.spanWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(path, spansFilename), storage.FilePerm, shouldCache))
+ bw.writers.mustInitForFilePart(fileSystem, path, shouldCache)
if traceSize > 0 {
bw.traceIDFilter = &traceIDFilter{
filter: generateTraceIDBloomFilter(),
diff --git a/banyand/trace/file_part_sync_test.go b/banyand/trace/file_part_sync_test.go
new file mode 100644
index 000000000..3d010b4b0
--- /dev/null
+++ b/banyand/trace/file_part_sync_test.go
@@ -0,0 +1,549 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "os"
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/onsi/gomega"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+func TestMustAddFilePart_FilesOnDisk(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _ := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), traceSnapshotOption(), nil)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ // Build a memPart with sample traces, flush it to disk to get valid binary files.
+ now := int64(1_000_000_000)
+ ts := &traces{
+ traceIDs: []string{"tid-1", "tid-2"},
+ timestamps: []int64{now, now + 1000},
+ spanIDs: []string{"s1", "s2"},
+ spans: [][]byte{[]byte("span-a"), []byte("span-b")},
+ tags: [][]*tagValue{{}, {}},
+ }
+ mp := generateMemPart()
+ mp.mustInitFromTraces(ts)
+
+ // Flush the memPart to a temporary directory to get valid binary files.
+ srcDir := filepath.Join(tmpPath, "src")
+ mp.mustFlush(fileSystem, srcDir)
+ releaseMemPart(mp)
+
+ // Verify that the source files exist on disk.
+ for _, name := range []string{metaFilename, primaryFilename, spansFilename} {
+ _, err := os.Stat(filepath.Join(srcDir, name))
+ require.NoError(t, err, "source file %s should exist", name)
+ }
+
+ // Allocate a part ID and copy files into the tsTable root the same way syncPartContext does.
+ partID := uint64(1)
+ destPath := partPath(tabDir, partID)
+ fileSystem.MkdirPanicIfExist(destPath, 0o755)
+
+ // Copy required files manually (simulating what syncPartContext does via writers).
+ copyFile := func(name string) {
+ data, readErr := fileSystem.Read(filepath.Join(srcDir, name))
+ require.NoError(t, readErr)
+ fs.MustFlush(fileSystem, data, filepath.Join(destPath, name), 0o600)
+ }
+ copyFile(metaFilename)
+ copyFile(primaryFilename)
+ copyFile(spansFilename)
+
+ // Write metadata.json.
+ pm := partMetadata{
+ CompressedSizeBytes: 10,
+ UncompressedSpanSizeBytes: 12,
+ TotalCount: 2,
+ BlocksCount: 1,
+ MinTimestamp: now,
+ MaxTimestamp: now + 1000,
+ }
+ pm.mustWriteMetadata(fileSystem, destPath)
+ fileSystem.SyncPath(destPath)
+
+ // Introduce the file part via mustAddFilePart.
+ tst.mustAddFilePart(partID, nil)
+
+ // Verify the part is in the snapshot.
+ snp := tst.currentSnapshot()
+ require.NotNil(t, snp, "snapshot must not be nil after mustAddFilePart")
+ defer snp.decRef()
+
+ require.Len(t, snp.parts, 1, "snapshot must contain exactly one part")
+ require.Equal(t, partID, snp.parts[0].p.partMetadata.ID, "part ID must match")
+ // File-backed partWrapper has nil mp field.
+ require.Nil(t, snp.parts[0].mp, "file-backed part must have nil memPart")
+}
+
+func TestMustAddFilePart_SurvivesReopenViaMustOpenFilePart(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ // Build and flush a memPart.
+ now := int64(2_000_000_000)
+ ts := &traces{
+ traceIDs: []string{"trace-x"},
+ timestamps: []int64{now},
+ spanIDs: []string{"sx"},
+ spans: [][]byte{[]byte("payload")},
+ tags: [][]*tagValue{{}},
+ }
+ mp := generateMemPart()
+ mp.mustInitFromTraces(ts)
+
+ partID := uint64(42)
+ destPath := partPath(tabDir, partID)
+ mp.mustFlush(fileSystem, destPath)
+ totalCount := mp.partMetadata.TotalCount
+ releaseMemPart(mp)
+
+ // Re-open via mustOpenFilePart.
+ p := mustOpenFilePart(partID, tabDir, fileSystem)
+ defer p.close()
+
+ require.Equal(t, partID, p.partMetadata.ID)
+ require.Equal(t, totalCount, p.partMetadata.TotalCount)
+ require.NotEmpty(t, p.primaryBlockMetadata, "primary block metadata must be loaded")
+}
+
+func TestIntroduceMemPart_NilMpGuard(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _ := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), traceSnapshotOption(), nil)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ now := int64(3_000_000_000)
+ ts := &traces{
+ traceIDs: []string{"trace-nil-guard"},
+ timestamps: []int64{now},
+ spanIDs: []string{"s0"},
+ spans: [][]byte{[]byte("data")},
+ tags: [][]*tagValue{{}},
+ }
+ mp := generateMemPart()
+ mp.mustInitFromTraces(ts)
+ partID := uint64(7)
+ destPath := partPath(tabDir, partID)
+ mp.mustFlush(fileSystem, destPath)
+ releaseMemPart(mp)
+
+ // mustAddFilePart calls introducePart with a nil-mp partWrapper.
+ // It must not panic at addPendingDataCount.
+ require.NotPanics(t, func() {
+ tst.mustAddFilePart(partID, nil)
+ })
+
+ snp := tst.currentSnapshot()
+ require.NotNil(t, snp)
+ defer snp.decRef()
+ require.Len(t, snp.parts, 1)
+ require.Nil(t, snp.parts[0].mp)
+}
+
+// buildAndFlushMemPart creates a memPart from the given traces, flushes it to destPath,
+// and returns the original TotalCount. The memPart is released before returning.
+func buildAndFlushMemPart(t *testing.T, fileSystem fs.FileSystem, ts *traces, destPath string) uint64 {
+ t.Helper()
+ mp := generateMemPart()
+ mp.mustInitFromTraces(ts)
+ totalCount := mp.partMetadata.TotalCount
+ mp.mustFlush(fileSystem, destPath)
+ releaseMemPart(mp)
+ return totalCount
+}
+
+// TestFilePart_DirectWrite_ProducesCorrectFiles verifies that streaming file chunks through
+// syncPartContext.handleTraceFileChunk and calling FinishSync produces a valid file-backed
+// part (pw.mp == nil) that survives a close/reopen cycle.
+func TestFilePart_DirectWrite_ProducesCorrectFiles(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, err := newTSTable(
+ fileSystem, tabDir, common.Position{}, logger.GetLogger("test"),
+ timestamp.TimeRange{}, traceSnapshotOption(), nil,
+ )
+ require.NoError(t, err)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ // Build a valid on-disk part in a temporary source directory (simulates the sender).
+ srcDir := filepath.Join(tmpPath, "src")
+ now := int64(5_000_000_000)
+ ts := &traces{
+ traceIDs: []string{"tid-direct-1", "tid-direct-2"},
+ timestamps: []int64{now, now + 500},
+ spanIDs: []string{"sd1", "sd2"},
+ spans: [][]byte{[]byte("span-x"), []byte("span-y")},
+ tags: [][]*tagValue{{}, {}},
+ }
+ totalCount := buildAndFlushMemPart(t, fileSystem, ts, srcDir)
+ require.Greater(t, totalCount, uint64(0))
+
+ // Open the on-disk files as a reader (simulates sender reading them for transfer).
+ fileInfos, cleanup := CreatePartFileReaderFromPath(srcDir, fileSystem)
+ defer cleanup()
+ require.NotEmpty(t, fileInfos)
+
+ // Allocate a partID and destination path (simulates what syncPartContext does).
+ partID := uint64(100)
+ destPath := partPath(tabDir, partID)
+ fileSystem.MkdirPanicIfExist(destPath, 0o755)
+
+ // Create file-backed writers the same way syncPartContext.NewPartType does.
+ w := generateWriters()
+ w.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, metaFilename), storage.FilePerm, false))
+ w.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, primaryFilename), storage.FilePerm, false))
+ w.spanWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, spansFilename), storage.FilePerm, false))
+
+ // Stream each file's content through the writers (simulates handleTraceFileChunk).
+ const chunkSize = 512
+ for i := range fileInfos {
+ fi := &fileInfos[i]
+ var wr *writer
+ switch fi.Name {
+ case traceMetaName:
+ wr = &w.metaWriter
+ case tracePrimaryName:
+ wr = &w.primaryWriter
+ case traceSpansName:
+ wr = &w.spanWriter
+ default:
+ continue
+ }
+ buf := make([]byte, chunkSize)
+ for {
+ n, readErr := fi.Reader.Read(buf)
+ if n > 0 {
+ wr.MustWrite(buf[:n])
+ }
+ if readErr != nil {
+ break
+ }
+ }
+ }
+ w.MustClose()
+ releaseWriters(w)
+
+ // Write metadata (mirrors what syncPartContext.FinishSync does).
+ pm := partMetadata{
+ TotalCount: totalCount,
+ BlocksCount: 1,
+ MinTimestamp: now,
+ MaxTimestamp: now + 500,
+ }
+ pm.mustWriteMetadata(fileSystem, destPath)
+ fileSystem.SyncPath(destPath)
+
+ // Introduce the file-backed part.
+ tst.mustAddFilePart(partID, nil)
+
+ // Verify: snapshot contains the part with nil mp (file-backed).
+ snp := tst.currentSnapshot()
+ require.NotNil(t, snp, "snapshot must exist after mustAddFilePart")
+ defer snp.decRef()
+ require.Len(t, snp.parts, 1)
+ require.Equal(t, partID, snp.parts[0].p.partMetadata.ID)
+ require.Nil(t, snp.parts[0].mp, "file-backed part must have nil memPart")
+
+ // Verify: core files exist on disk.
+ for _, name := range []string{metaFilename, primaryFilename} {
+ _, statErr := os.Stat(filepath.Join(destPath, name))
+ require.NoError(t, statErr, "file %s must exist on disk", name)
+ }
+
+ // Verify: the on-disk part files are readable via mustOpenFilePart (simulates reopen).
+ // Note: snapshot persistence to disk requires a flush cycle; here we verify
+ // that the binary files written by the sync path are well-formed and loadable.
+ p := mustOpenFilePart(partID, tabDir, fileSystem)
+ defer p.close()
+ require.Equal(t, partID, p.partMetadata.ID, "reopened part ID must match")
+ require.Equal(t, totalCount, p.partMetadata.TotalCount, "reopened part TotalCount must match")
+ require.NotEmpty(t, p.primaryBlockMetadata, "primary block metadata must be loadable from disk")
+}
+
+// TestFilePart_ConcurrentSenders verifies that two concurrent goroutines each adding a
+// filePart via mustAddFilePart produce a snapshot containing both parts with no races.
+func TestFilePart_ConcurrentSenders(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _ := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), traceSnapshotOption(), nil)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ // Prepare two parts on disk before starting the goroutines.
+ type partSpec struct {
+ path string
+ id uint64
+ }
+ specs := []partSpec{
+ {id: 201},
+ {id: 202},
+ }
+ traceSets := []*traces{
+ {
+ traceIDs: []string{"concurrent-tid-1"},
+ timestamps: []int64{6_000_000_000},
+ spanIDs: []string{"cs1"},
+ spans: [][]byte{[]byte("span-c1")},
+ tags: [][]*tagValue{{}},
+ },
+ {
+ traceIDs: []string{"concurrent-tid-2"},
+ timestamps: []int64{7_000_000_000},
+ spanIDs: []string{"cs2"},
+ spans: [][]byte{[]byte("span-c2")},
+ tags: [][]*tagValue{{}},
+ },
+ }
+ for i := range specs {
+ specs[i].path = partPath(tabDir, specs[i].id)
+ buildAndFlushMemPart(t, fileSystem, traceSets[i], specs[i].path)
+ }
+
+ // Both goroutines call mustAddFilePart concurrently.
+ var wg sync.WaitGroup
+ for i := range specs {
+ wg.Add(1)
+ spec := specs[i]
+ go func() {
+ defer wg.Done()
+ tst.mustAddFilePart(spec.id, nil)
+ }()
+ }
+ wg.Wait()
+
+ // All parts must appear in the snapshot.
+ var snp *snapshot
+ require.Eventually(t, func() bool {
+ snp = tst.currentSnapshot()
+ if snp == nil {
+ return false
+ }
+ if len(snp.parts) == 2 {
+ return true
+ }
+ snp.decRef()
+ snp = nil
+ return false
+ }, flags.EventuallyTimeout, 10*time.Millisecond, "both parts must be in snapshot")
+ defer snp.decRef()
+
+ ids := make(map[uint64]struct{})
+ for _, pw := range snp.parts {
+ ids[pw.p.partMetadata.ID] = struct{}{}
+ assert.Nil(t, pw.mp, "part %d must be file-backed", pw.p.partMetadata.ID)
+ }
+ assert.Contains(t, ids, specs[0].id, "first sender's part must be present")
+ assert.Contains(t, ids, specs[1].id, "second sender's part must be present")
+}
+
+// TestFilePart_ErrorCleanup verifies that when a syncPartContext is closed without calling
+// FinishSync (e.g. gRPC stream cancellation), the incomplete part directory is removed from disk.
+func TestFilePart_ErrorCleanup(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ tst, _ := initTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), traceSnapshotOption(), nil)
+ tst.startLoop(1)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ // Simulate what CreatePartHandler does: allocate a partID and create file-backed writers.
+ partID := uint64(300)
+ destPath := partPath(tabDir, partID)
+ fileSystem.MkdirPanicIfExist(destPath, storage.DirPerm)
+
+ spc := &syncPartContext{
+ tsTable: tst,
+ fileSystem: fileSystem,
+ partPath: destPath,
+ partID: partID,
+ }
+ w := generateWriters()
+ w.metaWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, metaFilename), storage.FilePerm, false))
+ w.primaryWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, primaryFilename), storage.FilePerm, false))
+ w.spanWriter.init(fs.MustCreateFile(fileSystem, filepath.Join(destPath, spansFilename), storage.FilePerm, false))
+ spc.writers = w
+
+ // Write partial data (simulates a partial transfer before cancellation).
+ spc.writers.metaWriter.MustWrite([]byte("partial"))
+
+ // The directory must exist before Close.
+ _, statErr := os.Stat(destPath)
+ require.NoError(t, statErr, "part directory must exist before Close")
+
+ // Close without FinishSync (simulates gRPC stream cancellation).
+ require.NoError(t, spc.Close())
+
+ // The incomplete part directory must be cleaned up.
+ _, statErr = os.Stat(destPath)
+ require.True(t, os.IsNotExist(statErr), "incomplete part directory must be removed after Close without FinishSync")
+
+ // No parts must appear in the snapshot.
+ snp := tst.currentSnapshot()
+ if snp != nil {
+ defer snp.decRef()
+ assert.Empty(t, snp.parts, "no parts must be in snapshot after error cleanup")
+ }
+}
+
+// TestFilePart_FlusherSkipsFileParts verifies that the flusher does not attempt to re-flush
+// a file-backed part (pw.mp == nil), leaving it unchanged in the snapshot.
+func TestFilePart_FlusherSkipsFileParts(t *testing.T) {
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level: flags.LogLevel}))
+
+ fileSystem := fs.NewLocalFileSystem()
+ tmpPath, deferFn := test.Space(require.New(t))
+ defer deferFn()
+
+ tabDir := filepath.Join(tmpPath, "tab")
+ fileSystem.MkdirPanicIfExist(tabDir, 0o755)
+
+ // Use a non-zero flushTimeout so the flusher loop actually runs.
+ opt := option{
+ flushTimeout: 10 * time.Millisecond,
+ mergePolicy: newDefaultMergePolicyForTesting(),
+ protector: traceSnapshotOption().protector,
+ }
+ tst, err := newTSTable(fileSystem, tabDir, common.Position{}, logger.GetLogger("test"), timestamp.TimeRange{}, opt, nil)
+ require.NoError(t, err)
+ defer func() {
+ require.NoError(t, tst.Close())
+ }()
+
+ // Build and flush a memPart to disk, then introduce it as a filePart.
+ partID := uint64(400)
+ destPath := partPath(tabDir, partID)
+ now := int64(8_000_000_000)
+ ts := &traces{
+ traceIDs: []string{"flusher-skip-tid"},
+ timestamps: []int64{now},
+ spanIDs: []string{"fs1"},
+ spans: [][]byte{[]byte("span-fs")},
+ tags: [][]*tagValue{{}},
+ }
+ buildAndFlushMemPart(t, fileSystem, ts, destPath)
+ tst.mustAddFilePart(partID, nil)
+
+ // Confirm the part is in the snapshot with nil mp before the flusher runs.
+ snpBefore := tst.currentSnapshot()
+ require.NotNil(t, snpBefore)
+ require.Len(t, snpBefore.parts, 1)
+ require.Nil(t, snpBefore.parts[0].mp, "part must be file-backed before flusher cycle")
+ snpBefore.decRef()
+
+ // Assert that across many flusher cycles (flushTimeout=10ms over 500ms window), the
+ // snapshot continuously shows exactly one file-backed part — i.e., the flusher never
+ // mutates it into a memPart nor creates a duplicate.
+ g := gomega.NewWithT(t)
+ g.Consistently(func() bool {
+ snp := tst.currentSnapshot()
+ if snp == nil {
+ return false
+ }
+ defer snp.decRef()
+ if len(snp.parts) != 1 {
+ return false
+ }
+ if snp.parts[0].mp != nil {
+ return false
+ }
+ return snp.parts[0].p.partMetadata.ID == partID
+ }).WithTimeout(500 * time.Millisecond).WithPolling(20 * time.Millisecond).Should(gomega.BeTrue())
+
+ snpAfter := tst.currentSnapshot()
+ require.NotNil(t, snpAfter)
+ defer snpAfter.decRef()
+
+ // Exactly one on-disk part directory must exist (no duplicates).
+ entries := fileSystem.ReadDir(tabDir)
+ partDirs := 0
+ for _, e := range entries {
+ if e.IsDir() && e.Name() != sidxDirName && e.Name() != storage.FailedPartsDirName {
+ partDirs++
+ }
+ }
+ assert.Equal(t, 1, partDirs, "flusher must not create a second part directory for a file-backed part")
+}
diff --git a/banyand/trace/introducer.go b/banyand/trace/introducer.go
index e33bfb52e..93e23da08 100644
--- a/banyand/trace/introducer.go
+++ b/banyand/trace/introducer.go
@@ -25,15 +25,17 @@ import (
)
type introduction struct {
- memPart *partWrapper
- applied chan struct{}
- sidxReqsMap map[string]*sidx.MemPart
+ part *partWrapper
+ applied chan struct{}
+ sidxReqsMap map[string]*sidx.MemPart
+ sidxFilePartsMap map[string]string
}
func (i *introduction) reset() {
- i.memPart = nil
+ i.part = nil
i.applied = nil
i.sidxReqsMap = nil
+ i.sidxFilePartsMap = nil
}
var introductionPool = pool.Register[*introduction]("trace-introduction")
@@ -160,7 +162,7 @@ func (tst *tsTable) introducerLoop(flushCh chan *flusherIntroduction, mergeCh ch
return
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
- tst.introduceMemPart(next, epoch)
+ tst.introducePart(next, epoch)
tst.incTotalIntroduceLoopFinished("mem")
epoch++
case next := <-flushCh:
@@ -194,7 +196,7 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, me
return
case next := <-tst.introductions:
tst.incTotalIntroduceLoopStarted("mem")
- tst.introduceMemPart(next, epoch)
+ tst.introducePart(next, epoch)
tst.incTotalIntroduceLoopFinished("mem")
epoch++
case next := <-flushCh:
@@ -223,16 +225,21 @@ func (tst *tsTable) introducerLoopWithSync(flushCh chan *flusherIntroduction, me
}
}
-func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint64) {
+func (tst *tsTable) introducePart(nextIntroduction *introduction, epoch uint64) {
// Create generic transaction
txn := snapshotpkg.NewTransaction()
defer txn.Release()
// Prepare trace snapshot transition
- next := nextIntroduction.memPart
- tst.addPendingDataCount(-int64(next.mp.partMetadata.TotalCount))
+ next := nextIntroduction.part
+ if next.mp != nil {
+ tst.addPendingDataCount(-int64(next.mp.partMetadata.TotalCount))
+ }
partID := next.p.partMetadata.ID
+ // Capture the snapshot built by the transaction so we can persist it directly
+ // after commit, without re-acquiring a reference via currentSnapshot().
+ var traceNextSnp *snapshot
traceTransition := snapshotpkg.NewTransition(tst, func(cur *snapshot) *snapshot {
if cur == nil {
cur = new(snapshot)
@@ -240,7 +247,8 @@ func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint6
nextSnp := cur.copyAllTo(epoch)
nextSnp.parts = append(nextSnp.parts, next)
nextSnp.creator = snapshotCreatorMemPart
- return &nextSnp
+ traceNextSnp = &nextSnp
+ return traceNextSnp
})
defer traceTransition.Release()
snapshotpkg.AddTransition(txn, traceTransition)
@@ -254,6 +262,13 @@ func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint6
sidxTransitions = append(sidxTransitions, sidxTransition)
snapshotpkg.AddTransition(txn, sidxTransition)
}
+ for name, sidxPartPath := range nextIntroduction.sidxFilePartsMap {
+ sidxInstance := tst.mustGetOrCreateSidx(name)
+ prepareFunc := sidxInstance.PrepareFilePart(partID, sidxPartPath)
+ sidxTransition := snapshotpkg.NewTransition(sidxInstance, prepareFunc)
+ sidxTransitions = append(sidxTransitions, sidxTransition)
+ snapshotpkg.AddTransition(txn, sidxTransition)
+ }
defer func() {
for _, t := range sidxTransitions {
t.Release()
@@ -263,6 +278,12 @@ func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint6
// Commit all atomically under single transaction lock
txn.Commit()
+ // Persist snapshot for file-backed introductions so the part survives restart.
+ // memPart introductions skip persist because the flusher persists after mem→file conversion.
+ if next.mp == nil && traceNextSnp != nil {
+ tst.persistSnapshot(traceNextSnp)
+ }
+
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
}
diff --git a/banyand/trace/migration_test.go b/banyand/trace/migration_test.go
new file mode 100644
index 000000000..2ea48107e
--- /dev/null
+++ b/banyand/trace/migration_test.go
@@ -0,0 +1,1106 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "io"
+ "path/filepath"
+ "sort"
+ "testing"
+ "time"
+
+ "github.com/onsi/gomega"
+ "github.com/onsi/gomega/gleak"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/banyand/internal/sidx"
+ "github.com/apache/skywalking-banyandb/banyand/protector"
+ "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"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/query/model"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/timestamp"
+)
+
+// mustNewSpace creates a temp directory for tests, failing on error.
+func mustNewSpace(g *gomega.WithT) (string, func()) {
+ dir, cleanup, err := test.NewSpace()
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+ return dir, cleanup
+}
+
+// transferParts performs the file-based migration of all parts from sender snapshot to receiver tsTable.
+// For each part it reads the on-disk files, streams chunks through the real syncChunkCallback handler,
+// and calls FinishSync so that the receiver gains a file-backed part.
+// sidxAllPaths is a pre-captured map[sidxName]map[partID]path that was obtained atomically with senderSnp
+// (via waitForConsistentSidxSnapshot). Pass nil to skip SIDX transfer (for tests without SIDX).
+func transferParts(g *gomega.WithT, senderSnp *snapshot, receiverTst *tsTable, fileSystem fs.FileSystem, sidxAllPaths map[string]map[uint64]string) {
+ handler := &syncChunkCallback{l: logger.GetLogger("transfer-handler")}
+
+ buf := make([]byte, 64*1024)
+ streamFiles := func(files []queue.FileInfo, ctx *queue.ChunkedSyncPartContext) {
+ for i := range files {
+ fi := &files[i]
+ ctx.FileName = fi.Name
+ for {
+ n, readErr := fi.Reader.Read(buf)
+ if n > 0 {
+ g.Expect(handler.HandleFileChunk(ctx, buf[:n])).To(gomega.Succeed())
+ }
+ if errors.Is(readErr, io.EOF) {
+ break
+ }
+ g.Expect(readErr).NotTo(gomega.HaveOccurred())
+ }
+ }
+ }
+
+ for _, pw := range senderSnp.parts {
+ if pw.mp != nil {
+ // Only transfer file-backed parts.
+ continue
+ }
+
+ partID := pw.p.partMetadata.ID
+ coreFiles, cleanup := CreatePartFileReaderFromPath(pw.p.path, fileSystem)
+
+ // Load core part metadata from disk via the same helper used by lifecycle migration.
+ coreData, parseErr := ParsePartMetadata(fileSystem, pw.p.path)
+ g.Expect(parseErr).NotTo(gomega.HaveOccurred())
+ ctx := &queue.ChunkedSyncPartContext{
+ ID: coreData.ID,
+ CompressedSizeBytes: coreData.CompressedSizeBytes,
+ UncompressedSizeBytes: coreData.UncompressedSizeBytes,
+ TotalCount: coreData.TotalCount,
+ BlocksCount: coreData.BlocksCount,
+ MinTimestamp: coreData.MinTimestamp,
+ MaxTimestamp: coreData.MaxTimestamp,
+ PartType: PartTypeCore,
+ }
+
+ partCtx := &syncPartContext{tsTable: receiverTst, l: logger.GetLogger("transfer-part")}
+
+ // Transfer SIDX parts first so they share the same partID as the core part
+ // (production order: SIDX parts arrive before core part in VisitShard).
+ for sidxName, pathsByID := range sidxAllPaths {
+ sidxPartPath, ok := pathsByID[partID]
+ if !ok {
+ continue
+ }
+ sidxData, sidxParseErr := sidx.ParsePartMetadata(fileSystem, sidxPartPath)
+ g.Expect(sidxParseErr).NotTo(gomega.HaveOccurred())
+ sidxCtx := &queue.ChunkedSyncPartContext{
+ ID: sidxData.ID,
+ CompressedSizeBytes: sidxData.CompressedSizeBytes,
+ UncompressedSizeBytes: sidxData.UncompressedSizeBytes,
+ TotalCount: sidxData.TotalCount,
+ BlocksCount: sidxData.BlocksCount,
+ MinKey: sidxData.MinKey,
+ MaxKey: sidxData.MaxKey,
+ PartType: sidxName,
+ }
+ g.Expect(partCtx.NewPartType(sidxCtx)).To(gomega.Succeed())
+ sidxCtx.Handler = partCtx
+
+ sidxFiles, sidxCleanup := sidx.CreatePartFileReaderFromPath(sidxPartPath, fileSystem)
+ streamFiles(sidxFiles, sidxCtx)
+ sidxCleanup()
+ }
+
+ // Transfer core part.
+ g.Expect(partCtx.NewPartType(ctx)).To(gomega.Succeed())
+ ctx.Handler = partCtx
+ streamFiles(coreFiles, ctx)
+ cleanup()
+
+ g.Expect(partCtx.FinishSync()).To(gomega.Succeed())
+ }
+}
+
+// readPartFiles reads every regular file in partPath into a name→bytes map.
+// Directories are skipped. Intended for byte-level comparison of two migrated parts.
+func readPartFiles(g *gomega.WithT, fileSystem fs.FileSystem, partPath string) map[string][]byte {
+ entries := fileSystem.ReadDir(partPath)
+ files := make(map[string][]byte, len(entries))
+ for _, e := range entries {
+ if e.IsDir() {
+ continue
+ }
+ data, err := fileSystem.Read(filepath.Join(partPath, e.Name()))
+ g.Expect(err).NotTo(gomega.HaveOccurred(), "read %s/%s", partPath, e.Name())
+ files[e.Name()] = data
+ }
+ return files
+}
+
+// assertFileSetsEqual verifies that sender and receiver produced byte-identical files,
+// excluding the manifest file whose ID/local-state may legitimately differ.
+func assertFileSetsEqual(g *gomega.WithT, senderFiles, receiverFiles map[string][]byte, manifestName, context string) {
+ g.Expect(receiverFiles).To(gomega.HaveLen(len(senderFiles)), "[%s] file count", context)
+ for name, senderBytes := range senderFiles {
+ receiverBytes, ok := receiverFiles[name]
+ g.Expect(ok).To(gomega.BeTrue(), "[%s] receiver missing file %q", context, name)
+ if name == manifestName {
+ continue
+ }
+ g.Expect(receiverBytes).To(gomega.Equal(senderBytes), "[%s] file %q bytes mismatch", context, name)
+ }
+ for name := range receiverFiles {
+ _, ok := senderFiles[name]
+ g.Expect(ok).To(gomega.BeTrue(), "[%s] receiver has extra file %q", context, name)
+ }
+}
+
+// assertCorePartsEqual verifies a receiver core trace part byte-matches the sender's,
+// and that every non-ID field in metadata.json round-trips faithfully.
+func assertCorePartsEqual(g *gomega.WithT, fileSystem fs.FileSystem, senderPath, receiverPath string) {
+ senderMeta, err := ParsePartMetadata(fileSystem, senderPath)
+ g.Expect(err).NotTo(gomega.HaveOccurred(), "parse sender core metadata.json at %s", senderPath)
+ receiverMeta, err := ParsePartMetadata(fileSystem, receiverPath)
+ g.Expect(err).NotTo(gomega.HaveOccurred(), "parse receiver core metadata.json at %s", receiverPath)
+
+ g.Expect(receiverMeta.CompressedSizeBytes).To(gomega.Equal(senderMeta.CompressedSizeBytes), "core CompressedSizeBytes")
+ g.Expect(receiverMeta.UncompressedSizeBytes).To(gomega.Equal(senderMeta.UncompressedSizeBytes), "core UncompressedSizeBytes")
+ g.Expect(receiverMeta.TotalCount).To(gomega.Equal(senderMeta.TotalCount), "core TotalCount")
+ g.Expect(receiverMeta.BlocksCount).To(gomega.Equal(senderMeta.BlocksCount), "core BlocksCount")
+ g.Expect(receiverMeta.MinTimestamp).To(gomega.Equal(senderMeta.MinTimestamp), "core MinTimestamp")
+ g.Expect(receiverMeta.MaxTimestamp).To(gomega.Equal(senderMeta.MaxTimestamp), "core MaxTimestamp")
+
+ senderFiles := readPartFiles(g, fileSystem, senderPath)
+ receiverFiles := readPartFiles(g, fileSystem, receiverPath)
+ assertFileSetsEqual(g, senderFiles, receiverFiles, metadataFilename, "core "+senderPath)
+}
+
+// assertSidxPartsEqual verifies a receiver SIDX part byte-matches the sender's,
+// and that every non-ID field in manifest.json round-trips faithfully — including
+// SegmentID, which is smuggled through the MinTimestamp field in the sync protocol.
+func assertSidxPartsEqual(g *gomega.WithT, fileSystem fs.FileSystem, senderPath, receiverPath string) {
+ senderMeta, err := sidx.ParsePartMetadata(fileSystem, senderPath)
+ g.Expect(err).NotTo(gomega.HaveOccurred(), "parse sender sidx manifest.json at %s", senderPath)
+ receiverMeta, err := sidx.ParsePartMetadata(fileSystem, receiverPath)
+ g.Expect(err).NotTo(gomega.HaveOccurred(), "parse receiver sidx manifest.json at %s", receiverPath)
+
+ g.Expect(receiverMeta.CompressedSizeBytes).To(gomega.Equal(senderMeta.CompressedSizeBytes), "sidx CompressedSizeBytes")
+ g.Expect(receiverMeta.UncompressedSizeBytes).To(gomega.Equal(senderMeta.UncompressedSizeBytes), "sidx UncompressedSizeBytes")
+ g.Expect(receiverMeta.TotalCount).To(gomega.Equal(senderMeta.TotalCount), "sidx TotalCount")
+ g.Expect(receiverMeta.BlocksCount).To(gomega.Equal(senderMeta.BlocksCount), "sidx BlocksCount")
+ g.Expect(receiverMeta.MinKey).To(gomega.Equal(senderMeta.MinKey), "sidx MinKey")
+ g.Expect(receiverMeta.MaxKey).To(gomega.Equal(senderMeta.MaxKey), "sidx MaxKey")
+ // The sender hijacks StreamingPartData.MinTimestamp to carry SegmentID (see sidx/sync.go),
+ // so this assertion is what proves SegmentID round-trips through the sync protocol.
+ g.Expect(receiverMeta.MinTimestamp).To(gomega.Equal(senderMeta.MinTimestamp), "sidx MinTimestamp")
+ g.Expect(receiverMeta.MaxTimestamp).To(gomega.Equal(senderMeta.MaxTimestamp), "sidx MaxTimestamp")
+
+ senderFiles := readPartFiles(g, fileSystem, senderPath)
+ receiverFiles := readPartFiles(g, fileSystem, receiverPath)
+ assertFileSetsEqual(g, senderFiles, receiverFiles, "manifest.json", "sidx "+senderPath)
+}
+
+// pairSnapshotParts returns sender→receiver part path pairs. The sender sorts file-backed
+// parts by partID ASC before streaming (syncer.executeSyncOperation), and the receiver
+// allocates local partIDs monotonically as each part arrives (CreatePartHandler), so sorting
+// both sides by partID ASC yields 1-1 pairing in transfer order — robust against any future
+// change in snapshot.parts iteration order.
+func pairSnapshotParts(g *gomega.WithT, senderSnp *snapshot, receiverTst *tsTable) []struct{ senderPath, receiverPath string } {
+ var senderParts []*partWrapper
+ for _, pw := range senderSnp.parts {
+ if pw.mp == nil {
+ senderParts = append(senderParts, pw)
+ }
+ }
+ sort.Slice(senderParts, func(i, j int) bool {
+ return senderParts[i].p.partMetadata.ID < senderParts[j].p.partMetadata.ID
+ })
+ receiverSnp := receiverTst.currentSnapshot()
+ g.Expect(receiverSnp).NotTo(gomega.BeNil(), "receiver snapshot must exist after migration")
+ defer receiverSnp.decRef()
+ g.Expect(len(receiverSnp.parts)).To(gomega.Equal(len(senderParts)), "receiver part count must match sender")
+
+ receiverParts := make([]*partWrapper, len(receiverSnp.parts))
+ copy(receiverParts, receiverSnp.parts)
+ sort.Slice(receiverParts, func(i, j int) bool {
+ return receiverParts[i].p.partMetadata.ID < receiverParts[j].p.partMetadata.ID
+ })
+
+ pairs := make([]struct{ senderPath, receiverPath string }, 0, len(senderParts))
+ for i := range senderParts {
+ pairs = append(pairs, struct{ senderPath, receiverPath string }{
+ senderPath: senderParts[i].p.path,
+ receiverPath: receiverParts[i].p.path,
+ })
+ }
+ return pairs
+}
+
+// assertMigrationPreservesAllFiles walks every file-backed part in the sender snapshot and
+// asserts byte-identical migration on the receiver side: every file, every metadata field.
+// senderSidxPaths is the sender's SIDX part paths captured atomically via
+// waitForConsistentSidxSnapshot; pass nil for tests without SIDX. The function derives the
+// matching receiver SIDX paths from the receiver's tsTable so callers don't need to plumb them.
+func assertMigrationPreservesAllFiles(g *gomega.WithT, fileSystem fs.FileSystem, senderSnp *snapshot,
+ receiverTst *tsTable, senderSidxPaths map[string]map[uint64]string,
+) {
+ pairs := pairSnapshotParts(g, senderSnp, receiverTst)
+ for _, p := range pairs {
+ assertCorePartsEqual(g, fileSystem, p.senderPath, p.receiverPath)
+ }
+
+ if len(senderSidxPaths) == 0 {
+ return
+ }
+
+ // SIDX parts: pair by sender partID→receiver partID via iteration order.
+ senderCoreIDs := make([]uint64, 0, len(pairs))
+ for _, pw := range senderSnp.parts {
+ if pw.mp == nil {
+ senderCoreIDs = append(senderCoreIDs, pw.p.partMetadata.ID)
+ }
+ }
+ receiverSnp := receiverTst.currentSnapshot()
+ g.Expect(receiverSnp).NotTo(gomega.BeNil())
+ defer receiverSnp.decRef()
+ receiverCoreIDs := make([]uint64, 0, len(pairs))
+ receiverPartIDSet := make(map[uint64]struct{}, len(pairs))
+ for _, pw := range receiverSnp.parts {
+ receiverCoreIDs = append(receiverCoreIDs, pw.p.partMetadata.ID)
+ receiverPartIDSet[pw.p.partMetadata.ID] = struct{}{}
+ }
+ receiverSidxMap := receiverTst.getAllSidx()
+
+ for name, senderPathsByID := range senderSidxPaths {
+ receiverInst, ok := receiverSidxMap[name]
+ g.Expect(ok).To(gomega.BeTrue(), "receiver missing sidx instance %q", name)
+ receiverPathsByID := receiverInst.PartPaths(receiverPartIDSet)
+ for i, senderID := range senderCoreIDs {
+ senderSidxPath, hasSender := senderPathsByID[senderID]
+ if !hasSender {
+ continue
+ }
+ receiverID := receiverCoreIDs[i]
+ receiverSidxPath, hasReceiver := receiverPathsByID[receiverID]
+ g.Expect(hasReceiver).To(gomega.BeTrue(), "receiver sidx %q missing part for receiver partID %d", name, receiverID)
+ assertSidxPartsEqual(g, fileSystem, senderSidxPath, receiverSidxPath)
+ }
+ }
+}
+
+// reopenTSTable closes the existing tsTable and opens a new one at the same root.
+// This simulates a process restart and exercises the snapshot persistence path.
+func reopenTSTable(g *gomega.WithT, tst *tsTable, fileSystem fs.FileSystem, path string, loggerName string) *tsTable {
+ g.Expect(tst.Close()).To(gomega.Succeed())
+ reopened, err := newTSTable(
+ fileSystem, path, common.Position{},
+ logger.GetLogger(loggerName), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred(), "reopen tsTable at %s", path)
+ return reopened
+}
+
+// mustBuildSidxMemPart creates a sidx.MemPart for the given index name and write requests.
+// This simulates what write_standalone.go does when processing index rules.
+func mustBuildSidxMemPart(g *gomega.WithT, tst *tsTable, sidxName string, reqs []sidx.WriteRequest, minTS, maxTS int64) *sidx.MemPart {
+ sidxInstance, err := tst.getOrCreateSidx(sidxName)
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+ mp, err := sidxInstance.ConvertToMemPart(reqs, 0, &minTS, &maxTS)
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+ return mp
+}
+
+// snapshotPartIDSet returns a set of all file-backed part IDs in the snapshot.
+func snapshotPartIDSet(snp *snapshot) map[uint64]struct{} {
+ ids := make(map[uint64]struct{}, len(snp.parts))
+ for _, pw := range snp.parts {
+ if pw.mp == nil {
+ ids[pw.p.partMetadata.ID] = struct{}{}
+ }
+ }
+ return ids
+}
+
+// waitForFileParts blocks until the tsTable has at least one file-backed part (no memParts)
+// or the timeout elapses.
+func waitForFileParts(g *gomega.WithT, tst *tsTable) {
+ g.Eventually(func() bool {
+ snp := tst.currentSnapshot()
+ if snp == nil {
+ return false
+ }
+ defer snp.decRef()
+ if len(snp.parts) == 0 {
+ return false
+ }
+ for _, pw := range snp.parts {
+ if pw.mp != nil {
+ return false
+ }
+ }
+ return true
+ }).WithTimeout(30 * time.Second).WithPolling(100 * time.Millisecond).Should(gomega.BeTrue())
+}
+
+// waitForConsistentSidxSnapshot polls until the tsTable's core snapshot and all named SIDX instances
+// are consistent: every file-backed core part ID has a corresponding file-backed SIDX part.
+// Returns the snapshot (caller must call decRef) and a map[sidxName]map[partID]path captured at the
+// same moment the consistency check passed, so they are guaranteed to be in sync.
+func waitForConsistentSidxSnapshot(g *gomega.WithT, tst *tsTable, sidxNames []string) (*snapshot, map[string]map[uint64]string) {
+ var resultSnp *snapshot
+ var resultPaths map[string]map[uint64]string
+
+ g.Eventually(func() bool {
+ snp := tst.currentSnapshot()
+ if snp == nil {
+ return false
+ }
+ coreIDs := snapshotPartIDSet(snp)
+ if len(coreIDs) == 0 {
+ snp.decRef()
+ return false
+ }
+ sidxMap := tst.getAllSidx()
+ captured := make(map[string]map[uint64]string, len(sidxNames))
+ for _, name := range sidxNames {
+ inst, ok := sidxMap[name]
+ if !ok {
+ snp.decRef()
+ return false
+ }
+ paths := inst.PartPaths(coreIDs)
+ if len(paths) == 0 {
+ snp.decRef()
+ return false
+ }
+ captured[name] = paths
+ }
+ // All SIDX instances are consistent with the core snapshot. Keep the ref.
+ if resultSnp != nil {
+ resultSnp.decRef()
+ }
+ resultSnp = snp
+ resultPaths = captured
+ return true
+ }).WithTimeout(30 * time.Second).WithPolling(50 * time.Millisecond).Should(gomega.BeTrue())
+
+ return resultSnp, resultPaths
+}
+
+// queryTraceID executes the standard query pattern used in query_test.go and returns all TraceResults.
+// The timestamp range is fixed at [1, 2] since every migration test writes traces at timestamp 1 or 2.
+func queryTraceID(g *gomega.WithT, snp *snapshot, traceID string) []model.TraceResult {
+ pp, _ := snp.getParts(nil, 1, 2, []string{traceID})
+
+ bma := generateBlockMetadataArray()
+ defer releaseBlockMetadataArray(bma)
+
+ ti := &tstIter{}
+ groupedTids := make([][]string, len(pp))
+ for i := range groupedTids {
+ groupedTids[i] = []string{traceID}
+ }
+ ti.init(bma, pp, groupedTids)
+
+ queryOpts := queryOptions{
+ schemaTagTypes: testSchemaTagTypes,
+ }
+
+ var cursors []*blockCursor
+ for ti.nextBlock() {
+ bc := generateBlockCursor()
+ p := ti.piPool[ti.idx]
+ opts := queryOpts
+ opts.TagProjection = allTagProjections
+ bc.init(p.p, p.curBlock, opts)
+ cursors = append(cursors, bc)
+ }
+ g.Expect(ti.Error()).NotTo(gomega.HaveOccurred())
+
+ cursorCh := make(chan scanCursorResult, len(cursors))
+ for _, bc := range cursors {
+ cursorCh <- scanCursorResult{cursor: bc}
+ }
+ close(cursorCh)
+
+ cursorBatch := make(chan *scanBatch, 1)
+ traceIDMap := make(map[uint64][]string)
+ traceIDMap[0] = []string{traceID}
+ cursorBatch <- &scanBatch{
+ traceBatch: traceBatch{
+ traceIDs: traceIDMap,
+ traceIDsOrder: []string{traceID},
+ keys: map[string]int64{traceID: 0},
+ },
+ cursorCh: cursorCh,
+ }
+ close(cursorBatch)
+
+ result := queryResult{
+ ctx: context.Background(),
+ tagProjection: allTagProjections,
+ cursorBatchCh: cursorBatch,
+ keys: map[string]int64{traceID: 0},
+ }
+ defer result.Release()
+
+ var got []model.TraceResult
+ for {
+ r := result.Pull()
+ if r == nil {
+ break
+ }
+ got = append(got, *r)
+ }
+ return got
+}
+
+// checkGoroutineLeak captures the set of live goroutines and registers a t.Cleanup
+// that fails the test if any new goroutine remains after the test returns.
+func checkGoroutineLeak(t *testing.T, g *gomega.WithT) {
+ t.Helper()
+ goods := gleak.Goroutines()
+ t.Cleanup(func() {
+ g.Eventually(gleak.Goroutines, 10*time.Second).ShouldNot(gleak.HaveLeaked(goods))
+ })
+}
+
+// TestLifecycleMigration_PreserveTraceData verifies that file-based migration preserves trace data correctly.
+func TestLifecycleMigration_PreserveTraceData(t *testing.T) {
+ g := gomega.NewWithT(t)
+ checkGoroutineLeak(t, g)
+ fileSystem := fs.NewLocalFileSystem()
+
+ // Phase 1: Sender — write known data and flush to disk.
+ senderPath, senderCleanup := mustNewSpace(g)
+ defer senderCleanup()
+
+ senderTst, err := newTSTable(
+ fileSystem, senderPath, common.Position{},
+ logger.GetLogger("sender"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+ defer func() {
+ g.Expect(senderTst.Close()).To(gomega.Succeed())
+ }()
+
+ senderTst.mustAddTraces(tsTS1, nil)
+ time.Sleep(100 * time.Millisecond)
+ senderTst.mustAddTraces(tsTS2, nil)
+ time.Sleep(100 * time.Millisecond)
+
+ waitForFileParts(g, senderTst)
+
+ senderSnp := senderTst.currentSnapshot()
+ g.Expect(senderSnp).NotTo(gomega.BeNil())
+ defer senderSnp.decRef()
+ g.Expect(len(senderSnp.parts)).To(gomega.BeNumerically(">", 0))
+
+ // Phase 2: Receiver — create tsTable and transfer all parts via the real handler.
+ receiverPath, receiverCleanup := mustNewSpace(g)
+ defer receiverCleanup()
+
+ receiverTst, err := newTSTable(
+ fileSystem, receiverPath, common.Position{},
+ logger.GetLogger("receiver"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+
+ transferParts(g, senderSnp, receiverTst, fileSystem, nil)
+
+ // Verify every core file/metadata field round-tripped byte-for-byte.
+ assertMigrationPreservesAllFiles(g, fileSystem, senderSnp, receiverTst, nil)
+
+ // Verify the receiver snapshot has only file-backed parts.
+ recvSnp := receiverTst.currentSnapshot()
+ g.Expect(recvSnp).NotTo(gomega.BeNil())
+ for _, pw := range recvSnp.parts {
+ g.Expect(pw.mp).To(gomega.BeNil(), "receiver should have file-backed parts only")
+ }
+ recvSnp.decRef()
+
+ defer func() {
+ g.Expect(receiverTst.Close()).To(gomega.Succeed())
+ }()
+
+ // Phase 3: Verify all parts on disk are readable via mustOpenFilePart and
+ // query returns correct data from the in-memory snapshot.
+ recvSnp = receiverTst.currentSnapshot()
+ g.Expect(recvSnp).NotTo(gomega.BeNil())
+ defer recvSnp.decRef()
+
+ // Verify each receiver part is well-formed on disk.
+ for _, pw := range recvSnp.parts {
+ p := mustOpenFilePart(pw.p.partMetadata.ID, receiverTst.root, fileSystem)
+ g.Expect(p.partMetadata.TotalCount).To(gomega.BeNumerically(">", uint64(0)), "reopened part must have data")
+ g.Expect(p.primaryBlockMetadata).NotTo(gomega.BeEmpty(), "primary block metadata must be loadable")
+ p.close()
+
+ // Verify metadata.json was written with correct values.
+ pp := partPath(receiverTst.root, pw.p.partMetadata.ID)
+ spd, metaErr := ParsePartMetadata(fileSystem, pp)
+ g.Expect(metaErr).NotTo(gomega.HaveOccurred(), "metadata.json must be readable")
+ g.Expect(spd.TotalCount).To(gomega.BeNumerically(">", uint64(0)), "metadata.json TotalCount must be > 0")
+ g.Expect(spd.MinTimestamp).To(gomega.BeNumerically("<=", spd.MaxTimestamp), "metadata.json timestamps must be valid")
+ }
+
+ // trace1 should return spans from both tsTS1 (span1) and tsTS2 (span4).
+ got := queryTraceID(g, recvSnp, "trace1")
+ g.Expect(got).NotTo(gomega.BeEmpty(), "should have query results for trace1")
+
+ totalSpans := 0
+ for _, tr := range got {
+ totalSpans += len(tr.Spans)
+ }
+ g.Expect(totalSpans).To(gomega.Equal(2), "trace1 should have 2 spans (one from each write batch)")
+
+ // Verify span content: collect all span bytes from all TraceResult entries.
+ var allSpans [][]byte
+ for _, tr := range got {
+ allSpans = append(allSpans, tr.Spans...)
+ }
+ containsSpan := func(spanData [][]byte, target []byte) bool {
+ for _, s := range spanData {
+ if bytes.Equal(s, target) {
+ return true
+ }
+ }
+ return false
+ }
+ g.Expect(containsSpan(allSpans, []byte("span1"))).To(gomega.BeTrue(), "span1 from tsTS1 must be present after migration")
+ g.Expect(containsSpan(allSpans, []byte("span4"))).To(gomega.BeTrue(), "span4 from tsTS2 must be present after migration")
+
+ // Verify tag values: find the Tags slice from any result that contains trace1 data.
+ // The query merges spans from tsTS1 (timestamp=1) and tsTS2 (timestamp=2) into one TraceResult.
+ // Tags are parallel to Spans: Tags[i].Values[j] corresponds to span j.
+ // strTag for trace1: tsTS1="value1", tsTS2="value4".
+ // intTag for trace1: tsTS1=10, tsTS2=40.
+ // strArrTag for trace1: tsTS1=["value1","value2"], tsTS2=["value5","value6"].
+ for _, tr := range got {
+ if len(tr.Tags) == 0 {
+ continue
+ }
+ for _, tag := range tr.Tags {
+ switch tag.Name {
+ case "strTag":
+ for _, tv := range tag.Values {
+ sv := tv.GetStr()
+ g.Expect(sv).NotTo(gomega.BeNil(), "strTag must be a string value")
+ g.Expect(sv.Value == "value1" || sv.Value == "value4").To(gomega.BeTrue(),
+ "strTag value must be value1 (tsTS1) or value4 (tsTS2), got: %s", sv.Value)
+ }
+ case "intTag":
+ for _, tv := range tag.Values {
+ iv := tv.GetInt()
+ g.Expect(iv).NotTo(gomega.BeNil(), "intTag must be an int64 value")
+ g.Expect(iv.Value == int64(10) || iv.Value == int64(40)).To(gomega.BeTrue(),
+ "intTag value must be 10 (tsTS1) or 40 (tsTS2), got: %d", iv.Value)
+ }
+ case "strArrTag":
+ for _, tv := range tag.Values {
+ av := tv.GetStrArray()
+ g.Expect(av).NotTo(gomega.BeNil(), "strArrTag must be a string array value")
+ g.Expect(len(av.Value)).To(gomega.Equal(2), "strArrTag must have 2 elements")
+ }
+ }
+ }
+ }
+
+ // trace2 and trace3 should also be queryable.
+ got2 := queryTraceID(g, recvSnp, "trace2")
+ g.Expect(got2).NotTo(gomega.BeEmpty(), "should have query results for trace2")
+
+ got3 := queryTraceID(g, recvSnp, "trace3")
+ g.Expect(got3).NotTo(gomega.BeEmpty(), "should have query results for trace3")
+}
+
+// TestLifecycleMigration_TwoConcurrentSenders verifies that data from two concurrent senders is preserved correctly.
+func TestLifecycleMigration_TwoConcurrentSenders(t *testing.T) {
+ g := gomega.NewWithT(t)
+ checkGoroutineLeak(t, g)
+ fileSystem := fs.NewLocalFileSystem()
+
+ // Sender 1 writes tsTS1 (timestamps=1), Sender 2 writes tsTS2 (timestamps=2).
+ // Both transfer to the same receiver.
+
+ sender1Path, s1Cleanup := mustNewSpace(g)
+ defer s1Cleanup()
+ sender2Path, s2Cleanup := mustNewSpace(g)
+ defer s2Cleanup()
+
+ mkSender := func(path string, name string, data *traces) (*tsTable, *snapshot) {
+ tst, mkErr := newTSTable(
+ fileSystem, path, common.Position{},
+ logger.GetLogger(name), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(mkErr).NotTo(gomega.HaveOccurred())
+ tst.mustAddTraces(data, nil)
+ time.Sleep(100 * time.Millisecond)
+ waitForFileParts(g, tst)
+ snp := tst.currentSnapshot()
+ g.Expect(snp).NotTo(gomega.BeNil())
+ return tst, snp
+ }
+
+ sender1Tst, sender1Snp := mkSender(sender1Path, "sender1", tsTS1)
+ defer func() {
+ sender1Snp.decRef()
+ g.Expect(sender1Tst.Close()).To(gomega.Succeed())
+ }()
+ sender2Tst, sender2Snp := mkSender(sender2Path, "sender2", tsTS2)
+ defer func() {
+ sender2Snp.decRef()
+ g.Expect(sender2Tst.Close()).To(gomega.Succeed())
+ }()
+
+ receiverPath, receiverCleanup := mustNewSpace(g)
+ defer receiverCleanup()
+
+ receiverTst, err := newTSTable(
+ fileSystem, receiverPath, common.Position{},
+ logger.GetLogger("receiver-two"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+ defer func() {
+ g.Expect(receiverTst.Close()).To(gomega.Succeed())
+ }()
+
+ // Transfer from both senders into the single receiver.
+ transferParts(g, sender1Snp, receiverTst, fileSystem, nil)
+ transferParts(g, sender2Snp, receiverTst, fileSystem, nil)
+
+ // Verify the receiver carries one file-backed part per sender file-backed part across both transfers.
+ expectedParts := 0
+ for _, pw := range sender1Snp.parts {
+ if pw.mp == nil {
+ expectedParts++
+ }
+ }
+ for _, pw := range sender2Snp.parts {
+ if pw.mp == nil {
+ expectedParts++
+ }
+ }
+ recvSnp := receiverTst.currentSnapshot()
+ g.Expect(recvSnp).NotTo(gomega.BeNil())
+ defer recvSnp.decRef()
+ g.Expect(len(recvSnp.parts)).To(gomega.Equal(expectedParts), "receiver part count must equal sum of senders' file-backed parts")
+ for _, pw := range recvSnp.parts {
+ g.Expect(pw.mp).To(gomega.BeNil(), "receiver parts must be file-backed")
+ }
+
+ // trace1 should surface spans from both senders.
+ got := queryTraceID(g, recvSnp, "trace1")
+ g.Expect(got).NotTo(gomega.BeEmpty())
+ totalSpans := 0
+ for _, tr := range got {
+ totalSpans += len(tr.Spans)
+ }
+ g.Expect(totalSpans).To(gomega.Equal(2), "trace1 should have 2 spans from two senders")
+
+ // trace2 should also be present.
+ got2 := queryTraceID(g, recvSnp, "trace2")
+ g.Expect(got2).NotTo(gomega.BeEmpty())
+}
+
+// TestLifecycleMigration_BloomFilterAfterMigration verifies that BloomFilter correctly filters after migration.
+func TestLifecycleMigration_BloomFilterAfterMigration(t *testing.T) {
+ g := gomega.NewWithT(t)
+ checkGoroutineLeak(t, g)
+ fileSystem := fs.NewLocalFileSystem()
+
+ senderPath, senderCleanup := mustNewSpace(g)
+ defer senderCleanup()
+
+ senderTst, err := newTSTable(
+ fileSystem, senderPath, common.Position{},
+ logger.GetLogger("sender-bf"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+ defer func() {
+ g.Expect(senderTst.Close()).To(gomega.Succeed())
+ }()
+
+ senderTst.mustAddTraces(tsTS1, nil)
+ time.Sleep(100 * time.Millisecond)
+
+ waitForFileParts(g, senderTst)
+
+ senderSnp := senderTst.currentSnapshot()
+ g.Expect(senderSnp).NotTo(gomega.BeNil())
+ defer senderSnp.decRef()
+
+ receiverPath, receiverCleanup := mustNewSpace(g)
+ defer receiverCleanup()
+
+ receiverTst, err := newTSTable(
+ fileSystem, receiverPath, common.Position{},
+ logger.GetLogger("receiver-bf"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+ defer func() {
+ g.Expect(receiverTst.Close()).To(gomega.Succeed())
+ }()
+
+ transferParts(g, senderSnp, receiverTst, fileSystem, nil)
+
+ // Verify every core file/metadata field round-tripped byte-for-byte.
+ assertMigrationPreservesAllFiles(g, fileSystem, senderSnp, receiverTst, nil)
+
+ recvSnp := receiverTst.currentSnapshot()
+ g.Expect(recvSnp).NotTo(gomega.BeNil())
+ defer recvSnp.decRef()
+
+ // Existing trace IDs should pass the bloom filter.
+ ppHit, nHit := recvSnp.getParts(nil, 1, 1, []string{"trace1"})
+ g.Expect(nHit).To(gomega.BeNumerically(">", 0), "trace1 should pass BloomFilter")
+ g.Expect(ppHit).NotTo(gomega.BeEmpty())
+
+ // Non-existing trace ID should be rejected by the bloom filter.
+ ppMiss, nMiss := recvSnp.getParts(nil, 1, 1, []string{"nonexistent-trace-id"})
+ g.Expect(nMiss).To(gomega.Equal(0), "nonexistent-trace-id should be filtered by BloomFilter")
+ g.Expect(ppMiss).To(gomega.BeEmpty())
+}
+
+// TestLifecycleMigration_WithSIDX verifies that SIDX parts are correctly migrated alongside core parts.
+// It creates trace data with SIDX (secondary index) entries, transfers both core and SIDX parts to a
+// receiver, and verifies the receiver's SIDX contains the expected file-backed parts.
+func TestLifecycleMigration_WithSIDX(t *testing.T) {
+ g := gomega.NewWithT(t)
+ checkGoroutineLeak(t, g)
+ fileSystem := fs.NewLocalFileSystem()
+
+ const sidxIndexName = "timestamp_millis"
+
+ // Phase 1: Sender — write trace data with SIDX.
+ senderPath, senderCleanup := mustNewSpace(g)
+ defer senderCleanup()
+
+ senderTst, err := newTSTable(
+ fileSystem, senderPath, common.Position{},
+ logger.GetLogger("sender-sidx"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+
+ // Build SIDX write requests that index the traces by timestamp (simulating a timestamp_millis index).
+ sidxReqs1 := []sidx.WriteRequest{
+ {SeriesID: 1, Key: 1, Data: []byte("trace1-ts1")},
+ {SeriesID: 2, Key: 1, Data: []byte("trace2-ts1")},
+ }
+ sidxReqs2 := []sidx.WriteRequest{
+ {SeriesID: 1, Key: 2, Data: []byte("trace1-ts2")},
+ {SeriesID: 3, Key: 2, Data: []byte("trace3-ts2")},
+ }
+
+ // Write first batch with SIDX data.
+ sidxMp1 := mustBuildSidxMemPart(g, senderTst, sidxIndexName, sidxReqs1, 1, 1)
+ senderTst.mustAddTraces(tsTS1, map[string]*sidx.MemPart{sidxIndexName: sidxMp1})
+ time.Sleep(100 * time.Millisecond)
+
+ // Write second batch with SIDX data.
+ sidxMp2 := mustBuildSidxMemPart(g, senderTst, sidxIndexName, sidxReqs2, 2, 2)
+ senderTst.mustAddTraces(tsTS2, map[string]*sidx.MemPart{sidxIndexName: sidxMp2})
+ time.Sleep(100 * time.Millisecond)
+
+ // Wait until core snapshot and SIDX are consistent (both merged to the same part IDs).
+ // waitForConsistentSidxSnapshot returns a snapshot and SIDX paths captured atomically.
+ senderSnp, senderSidxAllPaths := waitForConsistentSidxSnapshot(g, senderTst, []string{sidxIndexName})
+ g.Expect(senderSnp).NotTo(gomega.BeNil())
+ defer senderSnp.decRef()
+ g.Expect(len(senderSnp.parts)).To(gomega.BeNumerically(">", 0))
+ g.Expect(senderSidxAllPaths).To(gomega.HaveKey(sidxIndexName), "sender must have SIDX index %q", sidxIndexName)
+ senderSidxPaths := senderSidxAllPaths[sidxIndexName]
+ g.Expect(senderSidxPaths).NotTo(gomega.BeEmpty(), "sender SIDX must have file-backed parts after flush")
+
+ // Stop the sender's background goroutines (merger, flusher) before transferring.
+ // This prevents the merger from deleting on-disk part directories that transferParts reads.
+ g.Expect(senderTst.Close()).To(gomega.Succeed())
+
+ // Phase 2: Receiver — transfer both core and SIDX parts.
+ receiverPath, receiverCleanup := mustNewSpace(g)
+ defer receiverCleanup()
+
+ receiverTst, err := newTSTable(
+ fileSystem, receiverPath, common.Position{},
+ logger.GetLogger("receiver-sidx"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+ defer func() {
+ g.Expect(receiverTst.Close()).To(gomega.Succeed())
+ }()
+
+ transferParts(g, senderSnp, receiverTst, fileSystem, senderSidxAllPaths)
+
+ // Verify every core and SIDX file/metadata field (including SegmentID) round-tripped byte-for-byte.
+ assertMigrationPreservesAllFiles(g, fileSystem, senderSnp, receiverTst, senderSidxAllPaths)
+
+ // Phase 3: Verify core parts migrated correctly.
+ recvSnp := receiverTst.currentSnapshot()
+ g.Expect(recvSnp).NotTo(gomega.BeNil())
+ defer recvSnp.decRef()
+
+ g.Expect(len(recvSnp.parts)).To(gomega.BeNumerically(">", 0), "receiver must have core parts after migration")
+ for _, pw := range recvSnp.parts {
+ g.Expect(pw.mp).To(gomega.BeNil(), "receiver core parts must be file-backed")
+ }
+
+ // Verify trace data is queryable after migration.
+ got := queryTraceID(g, recvSnp, "trace1")
+ g.Expect(got).NotTo(gomega.BeEmpty(), "trace1 must be queryable after SIDX migration")
+
+ // Phase 4: Verify SIDX parts migrated to receiver.
+ receiverSidxMap := receiverTst.getAllSidx()
+ g.Expect(receiverSidxMap).To(gomega.HaveKey(sidxIndexName), "receiver must have SIDX index %q after migration", sidxIndexName)
+
+ recvPartIDSet := snapshotPartIDSet(recvSnp)
+ receiverSidxPaths := receiverSidxMap[sidxIndexName].PartPaths(recvPartIDSet)
+ g.Expect(receiverSidxPaths).NotTo(gomega.BeEmpty(), "receiver SIDX must have file-backed parts after migration")
+ g.Expect(len(receiverSidxPaths)).To(gomega.Equal(len(senderSidxPaths)),
+ "receiver SIDX part count must match sender SIDX part count")
+
+ // Verify manifest.json in each receiver SIDX part directory.
+ for _, sidxPartPath := range receiverSidxPaths {
+ sidxMetaErr := sidx.ValidatePartMetadata(fileSystem, sidxPartPath)
+ g.Expect(sidxMetaErr).NotTo(gomega.HaveOccurred(), "manifest.json must be valid in SIDX part %s", sidxPartPath)
+ }
+}
+
+// buildSidxTagReqs is a helper for constructing sidx.WriteRequest slices with a string tag value
+// under the fixed tag name "service", suitable for use with mustBuildSidxMemPart when testing
+// tagged SIDX data.
+func buildSidxTagReqs(seriesID uint64, key int64, data []byte, tagValue string) sidx.WriteRequest {
+ return sidx.WriteRequest{
+ SeriesID: common.SeriesID(seriesID),
+ Key: key,
+ Data: data,
+ Tags: []sidx.Tag{
+ {
+ Name: "service",
+ Value: []byte(tagValue),
+ ValueType: pbv1.ValueTypeStr,
+ },
+ },
+ }
+}
+
+// TestLifecycleMigration_WithSIDXTags verifies that SIDX tag data is correctly migrated alongside core parts.
+func TestLifecycleMigration_WithSIDXTags(t *testing.T) {
+ g := gomega.NewWithT(t)
+ checkGoroutineLeak(t, g)
+ fileSystem := fs.NewLocalFileSystem()
+
+ const sidxIndexName = "service_latency"
+
+ // Phase 1: Sender — write trace data with SIDX including tag data.
+ senderPath, senderCleanup := mustNewSpace(g)
+ defer senderCleanup()
+
+ senderTst, err := newTSTable(
+ fileSystem, senderPath, common.Position{},
+ logger.GetLogger("sender-sidx-tags"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+
+ // Build SIDX write requests with tags (simulating indexed span attributes like latency).
+ sidxReqs := []sidx.WriteRequest{
+ buildSidxTagReqs(1, 1, convert.Int64ToBytes(100), "frontend"),
+ buildSidxTagReqs(2, 1, convert.Int64ToBytes(200), "backend"),
+ buildSidxTagReqs(3, 2, convert.Int64ToBytes(150), "cache"),
+ }
+
+ sidxMp := mustBuildSidxMemPart(g, senderTst, sidxIndexName, sidxReqs, 1, 2)
+ senderTst.mustAddTraces(tsTS1, map[string]*sidx.MemPart{sidxIndexName: sidxMp})
+ time.Sleep(100 * time.Millisecond)
+
+ // Wait until core snapshot and SIDX are consistent.
+ senderSnp, senderSidxAllPaths := waitForConsistentSidxSnapshot(g, senderTst, []string{sidxIndexName})
+ g.Expect(senderSnp).NotTo(gomega.BeNil())
+ defer senderSnp.decRef()
+ g.Expect(senderSidxAllPaths).To(gomega.HaveKey(sidxIndexName))
+ senderSidxPaths := senderSidxAllPaths[sidxIndexName]
+ g.Expect(senderSidxPaths).NotTo(gomega.BeEmpty())
+
+ // Stop the sender's background goroutines before transferring to prevent merger from
+ // deleting on-disk part directories that transferParts reads.
+ g.Expect(senderTst.Close()).To(gomega.Succeed())
+
+ // Phase 2: Transfer to receiver.
+ receiverPath, receiverCleanup := mustNewSpace(g)
+ defer receiverCleanup()
+
+ receiverTst, err := newTSTable(
+ fileSystem, receiverPath, common.Position{},
+ logger.GetLogger("receiver-sidx-tags"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+ defer func() {
+ g.Expect(receiverTst.Close()).To(gomega.Succeed())
+ }()
+
+ transferParts(g, senderSnp, receiverTst, fileSystem, senderSidxAllPaths)
+
+ // Verify every core and SIDX file/metadata field (including tag files and SegmentID) round-tripped byte-for-byte.
+ assertMigrationPreservesAllFiles(g, fileSystem, senderSnp, receiverTst, senderSidxAllPaths)
+
+ // Phase 3: Verify receiver has SIDX with tag data.
+ receiverSidxMap := receiverTst.getAllSidx()
+ g.Expect(receiverSidxMap).To(gomega.HaveKey(sidxIndexName), "receiver must have SIDX index %q after migration", sidxIndexName)
+
+ recvSnp := receiverTst.currentSnapshot()
+ g.Expect(recvSnp).NotTo(gomega.BeNil())
+ defer recvSnp.decRef()
+ recvPartIDSet := snapshotPartIDSet(recvSnp)
+ receiverSidxPaths := receiverSidxMap[sidxIndexName].PartPaths(recvPartIDSet)
+ g.Expect(receiverSidxPaths).NotTo(gomega.BeEmpty(), "receiver SIDX must have file-backed parts with tag data")
+ g.Expect(len(receiverSidxPaths)).To(gomega.Equal(len(senderSidxPaths)),
+ "receiver SIDX part count must match sender")
+}
+
+// TestLifecycleMigration_ReceiverSurvivesRestart verifies that file-backed parts and SIDX parts
+// introduced via migration survive a receiver restart. This guards the snapshot-persist invariant:
+// introducePart must persist the snapshot for file-backed introductions, otherwise loadSnapshot
+// would treat the on-disk part as orphan and gc.removePart would delete it.
+func TestLifecycleMigration_ReceiverSurvivesRestart(t *testing.T) {
+ g := gomega.NewWithT(t)
+ checkGoroutineLeak(t, g)
+ fileSystem := fs.NewLocalFileSystem()
+
+ const sidxIndexName = "service_latency_restart"
+
+ // Phase 1: Sender writes core trace data + SIDX with non-zero SegmentID and tag data.
+ senderPath, senderCleanup := mustNewSpace(g)
+ defer senderCleanup()
+
+ senderTst, err := newTSTable(
+ fileSystem, senderPath, common.Position{},
+ logger.GetLogger("sender-restart"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+
+ sidxReqs := []sidx.WriteRequest{
+ buildSidxTagReqs(1, 1, convert.Int64ToBytes(100), "frontend"),
+ buildSidxTagReqs(2, 2, convert.Int64ToBytes(200), "backend"),
+ }
+ sidxMp := mustBuildSidxMemPart(g, senderTst, sidxIndexName, sidxReqs, 1, 2)
+ senderTst.mustAddTraces(tsTS1, map[string]*sidx.MemPart{sidxIndexName: sidxMp})
+ time.Sleep(100 * time.Millisecond)
+
+ senderSnp, senderSidxAllPaths := waitForConsistentSidxSnapshot(g, senderTst, []string{sidxIndexName})
+ g.Expect(senderSnp).NotTo(gomega.BeNil())
+ defer senderSnp.decRef()
+ g.Expect(senderSidxAllPaths).To(gomega.HaveKey(sidxIndexName))
+
+ g.Expect(senderTst.Close()).To(gomega.Succeed())
+
+ // Phase 2: Receiver receives the migration.
+ receiverPath, receiverCleanup := mustNewSpace(g)
+ defer receiverCleanup()
+
+ receiverTst, err := newTSTable(
+ fileSystem, receiverPath, common.Position{},
+ logger.GetLogger("receiver-restart-1"), timestamp.TimeRange{},
+ option{flushTimeout: 0, mergePolicy: newDefaultMergePolicyForTesting(), protector: protector.Nop{}},
+ nil,
+ )
+ g.Expect(err).NotTo(gomega.HaveOccurred())
+
+ transferParts(g, senderSnp, receiverTst, fileSystem, senderSidxAllPaths)
+ assertMigrationPreservesAllFiles(g, fileSystem, senderSnp, receiverTst, senderSidxAllPaths)
+
+ // Capture pre-restart state for comparison.
+ preRestartSnp := receiverTst.currentSnapshot()
+ g.Expect(preRestartSnp).NotTo(gomega.BeNil())
+ preRestartCorePartIDs := snapshotPartIDSet(preRestartSnp)
+ g.Expect(preRestartCorePartIDs).NotTo(gomega.BeEmpty(), "pre-restart receiver must have file-backed parts")
+
+ preRestartSidxMap := receiverTst.getAllSidx()
+ preRestartSidxPaths := preRestartSidxMap[sidxIndexName].PartPaths(preRestartCorePartIDs)
+ g.Expect(preRestartSidxPaths).NotTo(gomega.BeEmpty(), "pre-restart receiver SIDX must have file-backed parts")
+ preRestartSnp.decRef()
+
+ preRestartGot := func() []model.TraceResult {
+ s := receiverTst.currentSnapshot()
+ defer s.decRef()
+ return queryTraceID(g, s, "trace1")
+ }()
+ g.Expect(preRestartGot).NotTo(gomega.BeEmpty(), "trace1 must be queryable before restart")
+
+ // Phase 3: Restart — close and reopen the receiver tsTable at the same root.
+ reopenedTst := reopenTSTable(g, receiverTst, fileSystem, receiverPath, "receiver-restart-2")
+ defer func() {
+ g.Expect(reopenedTst.Close()).To(gomega.Succeed())
+ }()
+
+ // Verify core parts survived: same partIDs, all file-backed.
+ postRestartSnp := reopenedTst.currentSnapshot()
+ g.Expect(postRestartSnp).NotTo(gomega.BeNil(), "reopened receiver must have a snapshot")
+ defer postRestartSnp.decRef()
+ postRestartCorePartIDs := snapshotPartIDSet(postRestartSnp)
+ g.Expect(postRestartCorePartIDs).To(gomega.Equal(preRestartCorePartIDs),
+ "reopened core part IDs must match pre-restart (snapshot must have been persisted)")
+ for _, pw := range postRestartSnp.parts {
+ g.Expect(pw.mp).To(gomega.BeNil(), "reopened parts must be file-backed")
+ }
+
+ // Verify SIDX survived: same partIDs and on-disk paths still exist.
+ postRestartSidxMap := reopenedTst.getAllSidx()
+ g.Expect(postRestartSidxMap).To(gomega.HaveKey(sidxIndexName), "SIDX index must reopen")
+ postRestartSidxPaths := postRestartSidxMap[sidxIndexName].PartPaths(postRestartCorePartIDs)
+ g.Expect(postRestartSidxPaths).To(gomega.Equal(preRestartSidxPaths),
+ "reopened SIDX part paths must match pre-restart (SIDX snapshot must persist as well)")
+
+ // Verify queries still return data on the reopened tsTable.
+ postRestartGot := queryTraceID(g, postRestartSnp, "trace1")
+ g.Expect(postRestartGot).NotTo(gomega.BeEmpty(), "trace1 must remain queryable after restart")
+
+ preSpans := 0
+ for _, tr := range preRestartGot {
+ preSpans += len(tr.Spans)
+ }
+ postSpans := 0
+ for _, tr := range postRestartGot {
+ postSpans += len(tr.Spans)
+ }
+ g.Expect(postSpans).To(gomega.Equal(preSpans), "span count must be identical after restart")
+}
diff --git a/banyand/trace/part_metadata.go b/banyand/trace/part_metadata.go
index bd3aec7a5..ae1a029a2 100644
--- a/banyand/trace/part_metadata.go
+++ b/banyand/trace/part_metadata.go
@@ -54,6 +54,15 @@ func (pm *partMetadata) reset() {
pm.ID = 0
}
+func (pm *partMetadata) fillFromSyncContext(ctx *queue.ChunkedSyncPartContext) {
+ pm.CompressedSizeBytes = ctx.CompressedSizeBytes
+ pm.UncompressedSpanSizeBytes = ctx.UncompressedSizeBytes
+ pm.TotalCount = ctx.TotalCount
+ pm.BlocksCount = ctx.BlocksCount
+ pm.MinTimestamp = ctx.MinTimestamp
+ pm.MaxTimestamp = ctx.MaxTimestamp
+}
+
func validatePartMetadata(fileSystem fs.FileSystem, partPath string) error {
metadataPath := filepath.Join(partPath, metadataFilename)
metadata, err := fileSystem.Read(metadataPath)
diff --git a/banyand/trace/streaming_pipeline_test.go b/banyand/trace/streaming_pipeline_test.go
index f1435c246..0d7dab1ae 100644
--- a/banyand/trace/streaming_pipeline_test.go
+++ b/banyand/trace/streaming_pipeline_test.go
@@ -90,6 +90,10 @@ func (f *fakeSIDX) PrepareMemPart(uint64, *sidx.MemPart) func(cur *sidx.Snapshot
return func(cur *sidx.Snapshot) *sidx.Snapshot { return cur }
}
+func (f *fakeSIDX) PrepareFilePart(uint64, string) func(cur *sidx.Snapshot) *sidx.Snapshot {
+ return func(cur *sidx.Snapshot) *sidx.Snapshot { return cur }
+}
+
func (f *fakeSIDX) PrepareFlushed(*sidx.FlusherIntroduction) func(cur *sidx.Snapshot) *sidx.Snapshot {
return func(cur *sidx.Snapshot) *sidx.Snapshot { return cur }
}
@@ -684,6 +688,10 @@ func (f *fakeSIDXInfinite) PrepareMemPart(uint64, *sidx.MemPart) func(cur *sidx.
return func(cur *sidx.Snapshot) *sidx.Snapshot { return cur }
}
+func (f *fakeSIDXInfinite) PrepareFilePart(uint64, string) func(cur *sidx.Snapshot) *sidx.Snapshot {
+ return func(cur *sidx.Snapshot) *sidx.Snapshot { return cur }
+}
+
func (f *fakeSIDXInfinite) PrepareFlushed(*sidx.FlusherIntroduction) func(cur *sidx.Snapshot) *sidx.Snapshot {
return func(cur *sidx.Snapshot) *sidx.Snapshot { return cur }
}
diff --git a/banyand/trace/tstable.go b/banyand/trace/tstable.go
index 741f1221f..05d5e11d1 100644
--- a/banyand/trace/tstable.go
+++ b/banyand/trace/tstable.go
@@ -502,14 +502,36 @@ func (tst *tsTable) Close() error {
return tst.closeSidxMap()
}
+func (tst *tsTable) mustAddFilePart(partID uint64, sidxFilePartsMap map[string]string) {
+ p := mustOpenFilePart(partID, tst.root, tst.fileSystem)
+ p.partMetadata.ID = partID
+
+ ind := generateIntroduction()
+ defer releaseIntroduction(ind)
+ ind.applied = make(chan struct{})
+ ind.part = newPartWrapper(nil, p)
+ ind.sidxFilePartsMap = sidxFilePartsMap
+
+ select {
+ case tst.introductions <- ind:
+ case <-tst.loopCloser.CloseNotify():
+ ind.part.decRef()
+ return
+ }
+ select {
+ case <-ind.applied:
+ case <-tst.loopCloser.CloseNotify():
+ }
+}
+
func (tst *tsTable) mustAddMemPart(mp *memPart, sidxReqsMap map[string]*sidx.MemPart) {
p := openMemPart(mp)
ind := generateIntroduction()
defer releaseIntroduction(ind)
ind.applied = make(chan struct{})
- ind.memPart = newPartWrapper(mp, p)
- ind.memPart.p.partMetadata.ID = atomic.AddUint64(&tst.curPartID, 1)
+ ind.part = newPartWrapper(mp, p)
+ ind.part.p.partMetadata.ID = atomic.AddUint64(&tst.curPartID, 1)
ind.sidxReqsMap = sidxReqsMap
startTime := time.Now()
totalCount := mp.partMetadata.TotalCount
@@ -518,7 +540,7 @@ func (tst *tsTable) mustAddMemPart(mp *memPart, sidxReqsMap map[string]*sidx.Mem
case tst.introductions <- ind:
case <-tst.loopCloser.CloseNotify():
tst.addPendingDataCount(-int64(totalCount))
- ind.memPart.decRef()
+ ind.part.decRef()
return
}
select {
diff --git a/banyand/trace/write_data.go b/banyand/trace/write_data.go
index c7290c349..5c6e6d302 100644
--- a/banyand/trace/write_data.go
+++ b/banyand/trace/write_data.go
@@ -19,7 +19,9 @@ package trace
import (
"fmt"
+ "path/filepath"
"strings"
+ "sync/atomic"
"time"
"github.com/apache/skywalking-banyandb/api/common"
@@ -27,28 +29,41 @@ import (
"github.com/apache/skywalking-banyandb/banyand/internal/sidx"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
type syncPartContext struct {
tsTable *tsTable
+ fileSystem fs.FileSystem
l *logger.Logger
writers *writers
- memPart *memPart
+ partPath string
sidxPartContexts map[string]*sidx.SyncPartContext
traceIDFilterBuffer []byte
tagTypeBuffer []byte
+ partMeta partMetadata
+ partID uint64
}
func (s *syncPartContext) NewPartType(ctx *queue.ChunkedSyncPartContext) error {
if ctx.PartType != PartTypeCore {
+ // Allocate partID on first call so SIDX parts use the same ID as the core part.
+ if s.partID == 0 {
+ s.partID = atomic.AddUint64(&s.tsTable.curPartID, 1)
+ s.fileSystem = s.tsTable.fileSystem
+ }
+ // Pre-register the SIDX instance before writing files so its init/loadSnapshot
+ // runs against an empty directory. Otherwise mustGetOrCreateSidx in introducePart
+ // would lazily create the SIDX, scan the directory, and delete the freshly-written
+ // part because it isn't in the (nil) availablePartIDs list.
+ if _, err := s.tsTable.getOrCreateSidx(ctx.PartType); err != nil {
+ return fmt.Errorf("cannot pre-create sidx %q: %w", ctx.PartType, err)
+ }
sidxPartContext := sidx.NewSyncPartContext()
- memPart := sidx.GenerateMemPart()
- memPart.SetPartMetadata(ctx.CompressedSizeBytes, ctx.UncompressedSizeBytes, ctx.TotalCount, ctx.BlocksCount, ctx.MinKey, ctx.MaxKey, ctx.ID)
- writers := sidx.GenerateWriters()
- writers.MustInitForMemPart(memPart)
- sidxPartContext.Set(ctx.PartType, memPart, writers)
+ sidxPartPath := filepath.Join(s.tsTable.root, sidxDirName, ctx.PartType, fmt.Sprintf("%016x", s.partID))
+ sidxPartContext.SetForFile(ctx.PartType, s.fileSystem, sidxPartPath, ctx, s.tsTable.pm.ShouldCache(int64(ctx.CompressedSizeBytes)))
if s.sidxPartContexts == nil {
s.sidxPartContexts = make(map[string]*sidx.SyncPartContext)
}
@@ -56,59 +71,68 @@ func (s *syncPartContext) NewPartType(ctx *queue.ChunkedSyncPartContext) error {
return nil
}
- memPart := generateMemPart()
- memPart.partMetadata.CompressedSizeBytes = ctx.CompressedSizeBytes
- memPart.partMetadata.UncompressedSpanSizeBytes = ctx.UncompressedSizeBytes
- memPart.partMetadata.TotalCount = ctx.TotalCount
- memPart.partMetadata.BlocksCount = ctx.BlocksCount
- memPart.partMetadata.MinTimestamp = ctx.MinTimestamp
- memPart.partMetadata.MaxTimestamp = ctx.MaxTimestamp
- memPart.partMetadata.ID = ctx.ID
- writers := generateWriters()
- writers.mustInitForMemPart(memPart)
- s.writers = writers
- s.memPart = memPart
+ // Allocate partID if not already allocated by an earlier SIDX NewPartType call.
+ if s.partID == 0 {
+ s.partID = atomic.AddUint64(&s.tsTable.curPartID, 1)
+ }
+ s.partPath = partPath(s.tsTable.root, s.partID)
+ s.fileSystem = s.tsTable.fileSystem
+
+ w := generateWriters()
+ w.mustInitForFilePart(s.fileSystem, s.partPath, s.tsTable.pm.ShouldCache(int64(ctx.CompressedSizeBytes)))
+ s.writers = w
+
+ s.partMeta.fillFromSyncContext(ctx)
return nil
}
func (s *syncPartContext) FinishSync() error {
- if len(s.traceIDFilterBuffer) > 0 && s.memPart != nil {
- bf := generateTraceIDBloomFilter()
- s.memPart.traceIDFilter.filter = decodeBloomFilter(s.traceIDFilterBuffer, bf)
+ if s.partPath == "" {
+ // No core part was written (only SIDX parts arrived). Nothing to finalize;
+ // Close will discard any partially-written SIDX parts too.
+ return s.Close()
}
- if len(s.tagTypeBuffer) > 0 && s.memPart != nil {
- if s.memPart.tagType == nil {
- s.memPart.tagType = make(tagType)
- }
- if err := s.memPart.tagType.unmarshal(s.tagTypeBuffer); err != nil {
- s.l.Error().Err(err).Msg("failed to unmarshal tag type data")
- return err
- }
+
+ // Close writers first so file data is flushed before we write sidecar metadata.
+ s.releaseCoreWriters()
+
+ if len(s.traceIDFilterBuffer) > 0 {
+ fs.MustFlush(s.fileSystem, s.traceIDFilterBuffer, filepath.Join(s.partPath, traceIDFilterFilename), storage.FilePerm)
+ }
+ if len(s.tagTypeBuffer) > 0 {
+ fs.MustFlush(s.fileSystem, s.tagTypeBuffer, filepath.Join(s.partPath, tagTypeFilename), storage.FilePerm)
}
+ s.partMeta.mustWriteMetadata(s.fileSystem, s.partPath)
+ s.fileSystem.SyncPath(s.partPath)
- if s.memPart != nil {
- mp := s.memPart
- s.memPart = nil
- sidxPartContexts := make(map[string]*sidx.MemPart, len(s.sidxPartContexts))
- for _, sidxPartContext := range s.sidxPartContexts {
- sidxPartContexts[sidxPartContext.Name()] = sidxPartContext.GetMemPart()
- }
- s.tsTable.mustAddMemPart(mp, sidxPartContexts)
+ // Finish SIDX writers and collect file paths for file-backed parts.
+ sidxFilePartsMap := make(map[string]string, len(s.sidxPartContexts))
+ for _, sidxPartContext := range s.sidxPartContexts {
+ sidxFilePartsMap[sidxPartContext.Name()] = sidxPartContext.Finish()
}
+ s.tsTable.mustAddFilePart(s.partID, sidxFilePartsMap)
+ s.partPath = ""
+ s.traceIDFilterBuffer = nil
+ s.tagTypeBuffer = nil
return s.Close()
}
-func (s *syncPartContext) Close() error {
- if s.writers != nil {
- s.writers.MustClose()
- releaseWriters(s.writers)
- s.writers = nil
+// releaseCoreWriters closes and releases the core writers, if any. Safe to call multiple times.
+func (s *syncPartContext) releaseCoreWriters() {
+ if s.writers == nil {
+ return
}
- if s.memPart != nil {
- // syncPartContext owns the memPart directly without partWrapper refcounting.
- // It must release via releaseMemPart, not decRef, which is used in mustAddMemPart.
- releaseMemPart(s.memPart)
- s.memPart = nil
+ s.writers.MustClose()
+ releaseWriters(s.writers)
+ s.writers = nil
+}
+
+func (s *syncPartContext) Close() error {
+ s.releaseCoreWriters()
+ // Clean up incomplete filePart on error (partPath is cleared on success in FinishSync)
+ if s.partPath != "" && s.fileSystem != nil {
+ s.fileSystem.MustRMAll(s.partPath)
+ s.partPath = ""
}
if s.sidxPartContexts != nil {
for _, sidxPartContext := range s.sidxPartContexts {
@@ -117,6 +141,7 @@ func (s *syncPartContext) Close() error {
s.sidxPartContexts = nil
}
s.tsTable = nil
+ s.fileSystem = nil
s.traceIDFilterBuffer = nil
s.tagTypeBuffer = nil
return nil
@@ -329,13 +354,9 @@ func (s *syncChunkCallback) handleTraceFileChunk(ctx *queue.ChunkedSyncPartConte
case fileName == traceSpansName:
partCtx.writers.spanWriter.MustWrite(chunk)
case fileName == traceIDFilterFilename:
- if partCtx.memPart != nil {
- s.handleTraceIDFilterChunk(partCtx, chunk)
- }
+ s.handleTraceIDFilterChunk(partCtx, chunk)
case fileName == tagTypeFilename:
- if partCtx.memPart != nil {
- s.handleTagTypeChunk(partCtx, chunk)
- }
+ s.handleTagTypeChunk(partCtx, chunk)
case strings.HasPrefix(fileName, traceTagsPrefix):
tagName := fileName[len(traceTagsPrefix):]
_, tagWriter := partCtx.writers.getWriters(tagName)