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 2025/07/28 01:41:46 UTC
(skywalking-banyandb) 01/01: Add chunked data sync
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch chunked-sync
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit ce69da41e8b767f01757f7ee78d18b94829dc7a9
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sun Jul 27 22:36:17 2025 +0800
Add chunked data sync
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 1 +
api/proto/banyandb/cluster/v1/rpc.proto | 91 ++++
banyand/measure/flusher.go | 5 +-
banyand/measure/part.go | 17 +-
banyand/measure/svc_data.go | 7 +-
banyand/measure/svc_standalone.go | 2 +
banyand/measure/syncer.go | 141 ++++--
banyand/measure/wqueue_test.go | 28 ++
banyand/measure/write_data.go | 250 ++++-----
banyand/measure/{write_data.go => write_index.go} | 85 ----
banyand/queue/local.go | 27 +
banyand/queue/pub/chunked_sync.go | 416 +++++++++++++++
banyand/queue/pub/chunked_sync_test.go | 242 +++++++++
banyand/queue/pub/pub.go | 41 ++
banyand/queue/queue.go | 99 ++++
banyand/queue/sub/chunked_sync.go | 458 +++++++++++++++++
banyand/queue/sub/chunked_sync_test.go | 556 +++++++++++++++++++++
banyand/queue/sub/server.go | 111 +++-
banyand/queue/test/chunked_sync_basic_test.go | 115 +++++
banyand/queue/test/chunked_sync_common.go | 387 ++++++++++++++
banyand/queue/test/chunked_sync_metadata_test.go | 280 +++++++++++
.../queue/test/chunked_sync_performance_test.go | 464 +++++++++++++++++
banyand/stream/block_writer.go | 27 +
banyand/stream/part.go | 14 +-
banyand/stream/svc_liaison.go | 4 +
banyand/stream/svc_standalone.go | 6 +-
banyand/stream/syncer.go | 141 ++++--
banyand/stream/wqueue_test.go | 28 ++
banyand/stream/write_data.go | 304 +++--------
banyand/stream/{write_data.go => write_index.go} | 85 ----
docs/api-reference.md | 189 +++++++
31 files changed, 3974 insertions(+), 647 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 9c7a5a65..0f28dffd 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -24,6 +24,7 @@ Release Notes.
- Push down aggregation for topN query.
- Introduce write queue mechanism in liaison nodes to efficiently synchronize stream and measure partition folders, improving write throughput and consistency
- Add trace module metadata management.
+- Add chunked data sync to improve memory efficiency and performance during data transfer operations, supporting configurable chunk sizes, retry mechanisms, and out-of-order handling for both measure and stream services.
### Bug Fixes
diff --git a/api/proto/banyandb/cluster/v1/rpc.proto b/api/proto/banyandb/cluster/v1/rpc.proto
index 1d922f8c..8f49294f 100644
--- a/api/proto/banyandb/cluster/v1/rpc.proto
+++ b/api/proto/banyandb/cluster/v1/rpc.proto
@@ -47,7 +47,98 @@ message HealthCheckResponse {
string error = 3;
}
+// Chunked Sync Service Messages.
+message SyncPartRequest {
+ string session_id = 1; // Unique session identifier for this sync operation.
+ uint32 chunk_index = 2; // Current chunk index (0-based).
+ bytes chunk_data = 4; // Actual chunk data.
+ string chunk_checksum = 5; // CRC32 checksum for this chunk.
+ repeated PartInfo parts_info = 6; // Information about parts contained in this chunk.
+
+ oneof content {
+ SyncMetadata metadata = 7; // Sent with first chunk (chunk_index = 0).
+ SyncCompletion completion = 8; // Sent with last chunk to finalize.
+ }
+}
+
+// Information about a part contained within a chunk.
+message PartInfo {
+ uint64 id = 1; // Unique identifier for this part.
+ repeated FileInfo files = 2; // Information about individual files within this part.
+ uint64 compressed_size_bytes = 3; // Compressed size in bytes from partMetadata.
+ uint64 uncompressed_size_bytes = 4; // Uncompressed size in bytes from partMetadata.
+ uint64 total_count = 5; // Total count from partMetadata.
+ uint64 blocks_count = 6; // Blocks count from partMetadata.
+ int64 min_timestamp = 7; // Minimum timestamp from partMetadata.
+ int64 max_timestamp = 8; // Maximum timestamp from partMetadata.
+}
+
+// Information about an individual file within a part.
+message FileInfo {
+ string name = 1; // File identifier (e.g., "primary", "timestamps", "tagFamilies:seriesId").
+ uint32 offset = 2; // Byte offset within the part where this file starts.
+ uint32 size = 3; // Size of this file in bytes.
+}
+
+// SyncMetadata contains metadata for the sync operation.
+message SyncMetadata {
+ string group = 1; // Group name (stream/measure).
+ uint32 shard_id = 2; // Shard identifier.
+ string topic = 4; // Sync topic (stream-part-sync or measure-part-sync).
+ int64 timestamp = 6; // Timestamp when sync started.
+ uint32 total_parts = 7; // Total number of parts being synced.
+}
+
+// SyncCompletion contains completion information for the sync operation.
+message SyncCompletion {
+ uint64 total_bytes_sent = 1; // Total bytes sent for validation.
+ uint32 total_parts_sent = 2; // Total number of parts sent.
+ uint32 total_chunks = 3; // Total number of chunks in this sync.
+}
+
+// SyncPartResponse contains the response for a sync part request.
+message SyncPartResponse {
+ string session_id = 1;
+ uint32 chunk_index = 2;
+ SyncStatus status = 3;
+ string error = 4;
+ SyncResult sync_result = 5; // Final result when sync completes.
+}
+
+// SyncResult contains the result of a sync operation.
+message SyncResult {
+ bool success = 1; // Whether entire sync was successful.
+ uint64 total_bytes_received = 2; // Total bytes received.
+ int64 duration_ms = 3; // Time taken for sync in milliseconds.
+ uint32 chunks_received = 4; // Number of chunks successfully received.
+ uint32 parts_received = 5; // Number of parts successfully received.
+ repeated PartResult parts_results = 6; // Results for each part.
+}
+
+// PartResult contains the result for individual parts.
+message PartResult {
+ bool success = 1; // Whether this part was processed successfully.
+ string error = 2; // Error message if processing failed.
+ uint32 bytes_processed = 3; // Number of bytes processed for this part.
+}
+
+// SyncStatus represents the status of a sync operation.
+enum SyncStatus {
+ SYNC_STATUS_UNSPECIFIED = 0; // Unspecified status.
+ SYNC_STATUS_CHUNK_RECEIVED = 1; // Chunk received and validated successfully.
+ SYNC_STATUS_CHUNK_CHECKSUM_MISMATCH = 2; // Chunk checksum validation failed.
+ SYNC_STATUS_CHUNK_OUT_OF_ORDER = 3; // Chunk received out of expected order.
+ SYNC_STATUS_SESSION_NOT_FOUND = 4; // Session ID not recognized.
+ SYNC_STATUS_SYNC_COMPLETE = 5; // Entire sync operation completed successfully.
+}
+
service Service {
rpc Send(stream SendRequest) returns (stream SendResponse);
rpc HealthCheck(HealthCheckRequest) returns (HealthCheckResponse);
}
+
+// ChunkedSyncService provides streaming sync capabilities for chunked data transfer.
+service ChunkedSyncService {
+ // SyncPart synchronizes part data using chunked transfer.
+ rpc SyncPart(stream SyncPartRequest) returns (stream SyncPartResponse);
+}
diff --git a/banyand/measure/flusher.go b/banyand/measure/flusher.go
index 4bc14826..01f4443f 100644
--- a/banyand/measure/flusher.go
+++ b/banyand/measure/flusher.go
@@ -211,7 +211,10 @@ func (tst *tsTable) flusherLoopNoMerger(flushCh chan *flusherIntroduction, intro
}()
curSnapshot := tst.currentSnapshot()
if curSnapshot != nil {
- defer curSnapshot.decRef()
+ defer func() {
+ curSnapshot.decRef()
+ curSnapshot = nil
+ }()
tst.flush(curSnapshot, flushCh)
epoch = curSnapshot.epoch
if tst.currentEpoch() != epoch {
diff --git a/banyand/measure/part.go b/banyand/measure/part.go
index c94f92fa..0d885620 100644
--- a/banyand/measure/part.go
+++ b/banyand/measure/part.go
@@ -37,11 +37,19 @@ import (
)
const (
+ // Streaming file names for measure data parts (without extensions).
+ measurePrimaryName = "primary"
+ measureMetaName = "meta"
+ measureTimestampsName = "timestamps"
+ measureFieldValuesName = "fv"
+ measureTagFamiliesPrefix = "tf:"
+ measureTagMetadataPrefix = "tfm:"
+
metadataFilename = "metadata.json"
- primaryFilename = "primary.bin"
- metaFilename = "meta.bin"
- timestampsFilename = "timestamps.bin"
- fieldValuesFilename = "fields.bin"
+ primaryFilename = measurePrimaryName + ".bin"
+ metaFilename = measureMetaName + ".bin"
+ timestampsFilename = measureTimestampsName + ".bin"
+ fieldValuesFilename = measureFieldValuesName + ".bin"
tagFamiliesMetadataFilenameExt = ".tfm"
tagFamiliesFilenameExt = ".tf"
)
@@ -322,6 +330,7 @@ func (mp *memPart) Unmarshal(data []byte) error {
return nil
}
+//nolint:unused // This function is used in test files
func (mp *memPart) mustInitFromPart(p *part) {
mp.reset()
diff --git a/banyand/measure/svc_data.go b/banyand/measure/svc_data.go
index 00d1ba4f..ef8e7262 100644
--- a/banyand/measure/svc_data.go
+++ b/banyand/measure/svc_data.go
@@ -179,11 +179,8 @@ func (s *dataSVC) PreRun(ctx context.Context) error {
return err
}
- err := s.pipeline.Subscribe(data.TopicMeasurePartSync, setUpSyncCallback(s.l, s.schemaRepo))
- if err != nil {
- return err
- }
- err = s.pipeline.Subscribe(data.TopicMeasureSeriesIndexInsert, setUpIndexCallback(s.l, s.schemaRepo, data.TopicMeasureSeriesIndexInsert))
+ s.pipeline.RegisterChunkedSyncHandler(data.TopicMeasurePartSync, setUpChunkedSyncCallback(s.l, s.schemaRepo))
+ err := s.pipeline.Subscribe(data.TopicMeasureSeriesIndexInsert, setUpIndexCallback(s.l, s.schemaRepo, data.TopicMeasureSeriesIndexInsert))
if err != nil {
return err
}
diff --git a/banyand/measure/svc_standalone.go b/banyand/measure/svc_standalone.go
index e9048550..8f8a1780 100644
--- a/banyand/measure/svc_standalone.go
+++ b/banyand/measure/svc_standalone.go
@@ -192,6 +192,8 @@ func (s *standalone) PreRun(ctx context.Context) error {
if err != nil {
return err
}
+ // Register chunked sync handler for measure data.
+ s.pipeline.RegisterChunkedSyncHandler(data.TopicMeasurePartSync, setUpChunkedSyncCallback(s.l, s.schemaRepo))
return s.localPipeline.Subscribe(data.TopicMeasureWrite, writeListener)
}
diff --git a/banyand/measure/syncer.go b/banyand/measure/syncer.go
index f3e3fbe6..9774e159 100644
--- a/banyand/measure/syncer.go
+++ b/banyand/measure/syncer.go
@@ -23,9 +23,8 @@ import (
"time"
"github.com/apache/skywalking-banyandb/api/data"
- "github.com/apache/skywalking-banyandb/pkg/bus"
- "github.com/apache/skywalking-banyandb/pkg/convert"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
"github.com/apache/skywalking-banyandb/pkg/pool"
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
@@ -92,6 +91,7 @@ func (tst *tsTable) syncLoop(syncCh chan *syncIntroduction, flusherNotifier watc
}
tst.l.Error().Err(err).Msgf("cannot sync snapshot: %d", curSnapshot.epoch)
tst.incTotalFlushLoopErr(1)
+ time.Sleep(2 * time.Second)
return false
}
epoch = curSnapshot.epoch
@@ -108,6 +108,58 @@ func (tst *tsTable) syncLoop(syncCh chan *syncIntroduction, flusherNotifier watc
}
}
+func createPartFileReaders(part *part) ([]queue.FileInfo, func()) {
+ var files []queue.FileInfo
+
+ buf := bigValuePool.Generate()
+ // Stream metadata.
+ for i := range part.primaryBlockMetadata {
+ buf.Buf = part.primaryBlockMetadata[i].marshal(buf.Buf)
+ }
+ bb := bigValuePool.Generate()
+ bb.Buf = zstd.Compress(bb.Buf[:0], buf.Buf, 1)
+ bigValuePool.Release(buf)
+ files = append(files,
+ queue.FileInfo{
+ Name: measureMetaName,
+ Reader: bb.SequentialRead(),
+ },
+ queue.FileInfo{
+ Name: measurePrimaryName,
+ Reader: part.primary.SequentialRead(),
+ },
+ queue.FileInfo{
+ Name: measureTimestampsName,
+ Reader: part.timestamps.SequentialRead(),
+ },
+ queue.FileInfo{
+ Name: measureFieldValuesName,
+ Reader: part.fieldValues.SequentialRead(),
+ },
+ )
+
+ // Stream tag families data.
+ if part.tagFamilies != nil {
+ for name, reader := range part.tagFamilies {
+ files = append(files, queue.FileInfo{
+ Name: fmt.Sprintf("%s%s", measureTagFamiliesPrefix, name),
+ Reader: reader.SequentialRead(),
+ })
+ }
+
+ for name, reader := range part.tagFamilyMetadata {
+ files = append(files, queue.FileInfo{
+ Name: fmt.Sprintf("%s%s", measureTagMetadataPrefix, name),
+ Reader: reader.SequentialRead(),
+ })
+ }
+ }
+
+ return files, func() {
+ bigValuePool.Release(bb)
+ }
+}
+
func (tst *tsTable) syncSnapshot(curSnapshot *snapshot, syncCh chan *syncIntroduction) error {
var partsToSync []*part
for _, pw := range curSnapshot.parts {
@@ -120,7 +172,7 @@ func (tst *tsTable) syncSnapshot(curSnapshot *snapshot, syncCh chan *syncIntrodu
return nil
}
- // Sort parts from old to new (by part ID)
+ // Sort parts from old to new (by part ID).
for i := 0; i < len(partsToSync); i++ {
for j := i + 1; j < len(partsToSync); j++ {
if partsToSync[i].partMetadata.ID > partsToSync[j].partMetadata.ID {
@@ -129,45 +181,68 @@ func (tst *tsTable) syncSnapshot(curSnapshot *snapshot, syncCh chan *syncIntrodu
}
}
- nodes := tst.getNodes
- if nodes == nil {
- nodes = func() []string { return nil }
- }
- allNodes := nodes()
- if len(allNodes) == 0 {
+ nodes := tst.getNodes()
+ if len(nodes) == 0 {
return fmt.Errorf("no nodes to sync parts")
}
+ // Use chunked sync with streaming for better memory efficiency.
ctx := context.Background()
+ releaseFuncs := make([]func(), 0, len(partsToSync))
+ defer func() {
+ for _, release := range releaseFuncs {
+ release()
+ }
+ }()
- for _, node := range allNodes {
+ for _, node := range nodes {
+ // Get chunked sync client for this node.
+ chunkedClient, err := tst.option.tire2Client.NewChunkedSyncClient(node, 512*1024)
+ if err != nil {
+ return fmt.Errorf("failed to create chunked sync client for node %s: %w", node, err)
+ }
+ defer chunkedClient.Close()
+
+ // Prepare streaming parts data for chunked sync.
+ var streamingParts []queue.StreamingPartData
for _, part := range partsToSync {
- memPart := generateMemPart()
- memPart.mustInitFromPart(part)
+ // Create streaming reader for the part.
+ files, release := createPartFileReaders(part)
+ releaseFuncs = append(releaseFuncs, release)
- partData, err := memPart.Marshal()
- releaseMemPart(memPart)
- if err != nil {
- return fmt.Errorf("failed to marshal part %d: %w", part.partMetadata.ID, err)
- }
+ // Create streaming part sync data.
+ streamingParts = append(streamingParts, queue.StreamingPartData{
+ ID: part.partMetadata.ID,
+ Group: tst.group,
+ ShardID: uint32(tst.shardID),
+ Topic: data.TopicMeasurePartSync.String(),
+ Files: files,
+ CompressedSizeBytes: part.partMetadata.CompressedSizeBytes,
+ UncompressedSizeBytes: part.partMetadata.UncompressedSizeBytes,
+ TotalCount: part.partMetadata.TotalCount,
+ BlocksCount: part.partMetadata.BlocksCount,
+ MinTimestamp: part.partMetadata.MinTimestamp,
+ MaxTimestamp: part.partMetadata.MaxTimestamp,
+ })
+ }
- combinedData := make([]byte, 0, len(partData)+len(tst.group)+4)
- combinedData = encoding.EncodeBytes(combinedData, convert.StringToBytes(tst.group))
- combinedData = encoding.Uint32ToBytes(combinedData, uint32(tst.shardID))
- combinedData = append(combinedData, partData...)
+ // Sync parts using chunked transfer with streaming.
+ result, err := chunkedClient.SyncStreamingParts(ctx, streamingParts)
+ if err != nil {
+ return fmt.Errorf("failed to sync streaming parts to node %s: %w", node, err)
+ }
- message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), node, combinedData)
- f, publishErr := tst.option.tire2Client.Publish(ctx, data.TopicMeasurePartSync, message)
- if publishErr != nil {
- return fmt.Errorf("failed to publish part %d: %w", part.partMetadata.ID, publishErr)
- }
- if f != nil {
- _, publishErr = f.GetAll()
- if publishErr != nil {
- tst.l.Warn().Err(publishErr).Msgf("failed to get all sync results: %v", publishErr)
- }
- }
+ if !result.Success {
+ return fmt.Errorf("chunked sync partially failed: %v", result.ErrorMessage)
}
+ tst.l.Info().
+ Str("node", node).
+ Str("session", result.SessionID).
+ Uint64("bytes", result.TotalBytes).
+ Int64("duration_ms", result.DurationMs).
+ Uint32("chunks", result.ChunksCount).
+ Uint32("parts", result.PartsCount).
+ Msg("chunked sync completed successfully")
}
si := generateSyncIntroduction()
diff --git a/banyand/measure/wqueue_test.go b/banyand/measure/wqueue_test.go
index ad8b2160..24e5f99f 100644
--- a/banyand/measure/wqueue_test.go
+++ b/banyand/measure/wqueue_test.go
@@ -18,6 +18,7 @@
package measure
import (
+ "context"
"testing"
"time"
@@ -43,11 +44,20 @@ func Test_newWriteQueue(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockTire2Client := queue.NewMockClient(ctrl)
+
+ // Mock Publish method
mockTire2Client.EXPECT().
Publish(gomock.Any(), gomock.Any(), gomock.Any()).
Return(bus.Future(nil), nil).
AnyTimes()
+ // Mock NewChunkedSyncClient method
+ mockChunkedSyncClient := &mockChunkedSyncClient{}
+ mockTire2Client.EXPECT().
+ NewChunkedSyncClient(gomock.Any(), gomock.Any()).
+ Return(mockChunkedSyncClient, nil).
+ AnyTimes()
+
// Create tsTable using newWriteQueue
opts := option{
tire2Client: mockTire2Client,
@@ -88,3 +98,21 @@ func Test_newWriteQueue(t *testing.T) {
// Verify that the tire2Client was properly set
assert.Equal(t, mockTire2Client, tst.option.tire2Client)
}
+
+// mockChunkedSyncClient implements queue.ChunkedSyncClient for testing
+type mockChunkedSyncClient struct{}
+
+func (m *mockChunkedSyncClient) SyncStreamingParts(ctx context.Context, parts []queue.StreamingPartData) (*queue.SyncResult, error) {
+ return &queue.SyncResult{
+ Success: true,
+ SessionID: "mock-session",
+ TotalBytes: 0,
+ DurationMs: 0,
+ ChunksCount: 0,
+ PartsCount: uint32(len(parts)),
+ }, nil
+}
+
+func (m *mockChunkedSyncClient) Close() error {
+ return nil
+}
diff --git a/banyand/measure/write_data.go b/banyand/measure/write_data.go
index c7aa6215..986bb099 100644
--- a/banyand/measure/write_data.go
+++ b/banyand/measure/write_data.go
@@ -18,23 +18,42 @@
package measure
import (
- "context"
+ "fmt"
+ "strings"
"time"
"github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/api/data"
- "github.com/apache/skywalking-banyandb/pkg/bus"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
- "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pool"
)
+type syncPartContext struct {
+ tsTable *tsTable
+ writers *writers
+ memPart *memPart
+}
+
+func (s *syncPartContext) FinishSync() error {
+ s.tsTable.mustAddMemPart(s.memPart)
+ return s.Close()
+}
+
+func (s *syncPartContext) Close() error {
+ s.writers.MustClose()
+ releaseWriters(s.writers)
+ s.writers = nil
+ s.memPart = nil
+ s.tsTable = nil
+ return nil
+}
+
type syncCallback struct {
l *logger.Logger
schemaRepo *schemaRepo
}
-func setUpSyncCallback(l *logger.Logger, schemaRepo *schemaRepo) bus.MessageListener {
+func setUpChunkedSyncCallback(l *logger.Logger, schemaRepo *schemaRepo) queue.ChunkedSyncHandler {
return &syncCallback{
l: l,
schemaRepo: schemaRepo,
@@ -45,162 +64,109 @@ func (s *syncCallback) CheckHealth() *common.Error {
return nil
}
-func (s *syncCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
- data, ok := message.Data().([]byte)
- if !ok {
- s.l.Warn().Msg("invalid sync message data type")
- return
- }
-
- // Decode group name
- tail, groupBytes, err := encoding.DecodeBytes(data)
- if err != nil {
- s.l.Error().Err(err).Msg("failed to decode group name from sync message")
- return
- }
- group := string(groupBytes)
-
- // Decode shardID (uint32)
- if len(tail) < 4 {
- s.l.Error().Msg("insufficient data for shardID in sync message")
- return
- }
- shardID := encoding.BytesToUint32(tail[:4])
- tail = tail[4:]
-
- // Decode memPart
- memPart := generateMemPart()
-
- if err = memPart.Unmarshal(tail); err != nil {
- s.l.Error().Err(err).Msg("failed to unmarshal memPart from sync message")
- return
- }
-
- // Get the group schema from schemaRepo by the string group
- tsdb, err := s.schemaRepo.loadTSDB(group)
+// CreatePartHandler implements queue.ChunkedSyncHandler.
+func (s *syncCallback) CreatePartHandler(ctx *queue.ChunkedSyncPartContext) (queue.PartHandler, error) {
+ tsdb, err := s.schemaRepo.loadTSDB(ctx.Group)
if err != nil {
- s.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
- return
+ s.l.Error().Err(err).Str("group", ctx.Group).Msg("failed to load TSDB for group")
+ return nil, err
}
-
- // Get the segment using memPart.partMetadata.MinTimestamp
- segmentTime := time.Unix(0, memPart.partMetadata.MinTimestamp)
+ segmentTime := time.Unix(0, ctx.MinTimestamp)
segment, err := tsdb.CreateSegmentIfNotExist(segmentTime)
if err != nil {
- s.l.Error().Err(err).Str("group", group).Time("segmentTime", segmentTime).Msg("failed to create segment")
- return
+ s.l.Error().Err(err).Str("group", ctx.Group).Time("segmentTime", segmentTime).Msg("failed to create segment")
+ return nil, err
}
defer segment.DecRef()
-
- // Get Shard from a segment using shardID
- tsTable, err := segment.CreateTSTableIfNotExist(common.ShardID(shardID))
+ tsTable, err := segment.CreateTSTableIfNotExist(common.ShardID(ctx.ShardID))
if err != nil {
- s.l.Error().Err(err).Str("group", group).Uint32("shardID", shardID).Msg("failed to create ts table")
- return
+ s.l.Error().Err(err).Str("group", ctx.Group).Uint32("shardID", ctx.ShardID).Msg("failed to create ts table")
+ return nil, err
}
- tsdb.Tick(memPart.partMetadata.MaxTimestamp)
- // Use tsTable.mustAddMemPart to introduce memPart to tsTable
- tsTable.mustAddMemPart(memPart)
-
- if e := s.l.Debug(); e.Enabled() {
- e.
- Str("group", group).
- Uint32("shardID", shardID).
- Uint64("partID", memPart.partMetadata.ID).
- Msg("successfully stored memPart to tsTable")
- }
-
- return
-}
-
-type indexCallback struct {
- l *logger.Logger
- schemaRepo *schemaRepo
- topic bus.Topic
+ 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
}
-func setUpIndexCallback(l *logger.Logger, schemaRepo *schemaRepo, topic bus.Topic) bus.MessageListener {
- return &indexCallback{
- l: l,
- schemaRepo: schemaRepo,
- topic: topic,
+// HandleFileChunk implements queue.ChunkedSyncHandler for streaming file chunks.
+func (s *syncCallback) HandleFileChunk(ctx *queue.ChunkedSyncPartContext, chunk []byte) error {
+ if ctx.Handler == nil {
+ return fmt.Errorf("part handler is nil")
+ }
+ partCtx := ctx.Handler.(*syncPartContext)
+
+ // Select the appropriate writer based on the filename and write the chunk.
+ fileName := ctx.FileName
+ switch {
+ case fileName == measureMetaName:
+ partCtx.writers.metaWriter.MustWrite(chunk)
+ case fileName == measurePrimaryName:
+ partCtx.writers.primaryWriter.MustWrite(chunk)
+ case fileName == measureTimestampsName:
+ partCtx.writers.timestampsWriter.MustWrite(chunk)
+ case fileName == measureFieldValuesName:
+ partCtx.writers.fieldValuesWriter.MustWrite(chunk)
+ case strings.HasPrefix(fileName, measureTagFamiliesPrefix):
+ tagName := fileName[len(measureTagFamiliesPrefix):]
+ _, tagWriter := partCtx.writers.getColumnMetadataWriterAndColumnWriter(tagName)
+ tagWriter.MustWrite(chunk)
+ case strings.HasPrefix(fileName, measureTagMetadataPrefix):
+ tagName := fileName[len(measureTagMetadataPrefix):]
+ tagMetadataWriter, _ := partCtx.writers.getColumnMetadataWriterAndColumnWriter(tagName)
+ tagMetadataWriter.MustWrite(chunk)
+ default:
+ s.l.Warn().Str("fileName", fileName).Msg("unknown file type in chunked sync")
+ return fmt.Errorf("unknown file type: %s", fileName)
}
-}
-func (i *indexCallback) CheckHealth() *common.Error {
return nil
}
-func (i *indexCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
- msgData, ok := message.Data().([]byte)
- if !ok {
- i.l.Warn().Msg("invalid index insert message data type")
- return
- }
-
- // Decode group name (first part of the message)
- tail, groupBytes, err := encoding.DecodeBytes(msgData)
- if err != nil {
- i.l.Error().Err(err).Msg("failed to decode group name from index insert message")
- return
- }
- group := string(groupBytes)
-
- // Decode timestamp (uint64) - second part of the message
- if len(tail) < 8 {
- i.l.Error().Msg("insufficient data for timestamp in index insert message")
- return
- }
- timestamp := encoding.BytesToInt64(tail[:8])
- tail = tail[8:]
-
- // Unmarshal Documents from the remaining data
- var documents index.Documents
- if err = documents.Unmarshal(tail); err != nil {
- i.l.Error().Err(err).Msg("failed to unmarshal documents from index insert message")
- return
- }
-
- if len(documents) == 0 {
- i.l.Warn().Msg("empty documents in index insert message")
- return
+// HandlePartComplete implements queue.ChunkedSyncHandler for part completion.
+func (s *syncCallback) HandlePartComplete(ctx *queue.ChunkedSyncPartContext) error {
+ if ctx.Handler == nil {
+ return fmt.Errorf("part handler is nil")
}
+ partCtx := ctx.Handler.(*syncPartContext)
+ return partCtx.FinishSync()
+}
- // Get TSDB by group name using schemaRepo
- tsdb, err := i.schemaRepo.loadTSDB(group)
- if err != nil {
- i.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
- return
+func generateWriters() *writers {
+ v := writersPool.Get()
+ if v == nil {
+ return &writers{
+ tagFamilyMetadataWriters: make(map[string]*writer),
+ tagFamilyWriters: make(map[string]*writer),
+ }
}
+ return v
+}
- // Use the timestamp to get the segment directly
- segmentTime := time.Unix(0, timestamp)
- segment, err := tsdb.CreateSegmentIfNotExist(segmentTime)
- if err != nil {
- i.l.Error().Err(err).Str("group", group).Time("segmentTime", segmentTime).Msg("failed to create segment")
- return
- }
- defer segment.DecRef()
-
- switch i.topic {
- case data.TopicMeasureSeriesIndexInsert:
- err = segment.IndexDB().Insert(documents)
- case data.TopicMeasureSeriesIndexUpdate:
- err = segment.IndexDB().Update(documents)
- }
- if err != nil {
- i.l.Error().Err(err).Str("group", group).Int("documentCount", len(documents)).Msg("failed to insert documents to index")
- return
- }
+func releaseWriters(sw *writers) {
+ sw.reset()
+ writersPool.Put(sw)
+}
- if e := i.l.Debug(); e.Enabled() {
- e.
- Str("group", group).
- Int("documentCount", len(documents)).
- Time("segmentTime", segmentTime).
- Msg("successfully inserted documents to index")
- }
+var writersPool = pool.Register[*writers]("measure-writers")
- return
+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)
}
diff --git a/banyand/measure/write_data.go b/banyand/measure/write_index.go
similarity index 61%
copy from banyand/measure/write_data.go
copy to banyand/measure/write_index.go
index c7aa6215..b7146074 100644
--- a/banyand/measure/write_data.go
+++ b/banyand/measure/write_index.go
@@ -29,91 +29,6 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
)
-type syncCallback struct {
- l *logger.Logger
- schemaRepo *schemaRepo
-}
-
-func setUpSyncCallback(l *logger.Logger, schemaRepo *schemaRepo) bus.MessageListener {
- return &syncCallback{
- l: l,
- schemaRepo: schemaRepo,
- }
-}
-
-func (s *syncCallback) CheckHealth() *common.Error {
- return nil
-}
-
-func (s *syncCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
- data, ok := message.Data().([]byte)
- if !ok {
- s.l.Warn().Msg("invalid sync message data type")
- return
- }
-
- // Decode group name
- tail, groupBytes, err := encoding.DecodeBytes(data)
- if err != nil {
- s.l.Error().Err(err).Msg("failed to decode group name from sync message")
- return
- }
- group := string(groupBytes)
-
- // Decode shardID (uint32)
- if len(tail) < 4 {
- s.l.Error().Msg("insufficient data for shardID in sync message")
- return
- }
- shardID := encoding.BytesToUint32(tail[:4])
- tail = tail[4:]
-
- // Decode memPart
- memPart := generateMemPart()
-
- if err = memPart.Unmarshal(tail); err != nil {
- s.l.Error().Err(err).Msg("failed to unmarshal memPart from sync message")
- return
- }
-
- // Get the group schema from schemaRepo by the string group
- tsdb, err := s.schemaRepo.loadTSDB(group)
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
- return
- }
-
- // Get the segment using memPart.partMetadata.MinTimestamp
- segmentTime := time.Unix(0, memPart.partMetadata.MinTimestamp)
- segment, err := tsdb.CreateSegmentIfNotExist(segmentTime)
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Time("segmentTime", segmentTime).Msg("failed to create segment")
- return
- }
- defer segment.DecRef()
-
- // Get Shard from a segment using shardID
- tsTable, err := segment.CreateTSTableIfNotExist(common.ShardID(shardID))
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Uint32("shardID", shardID).Msg("failed to create ts table")
- return
- }
-
- tsdb.Tick(memPart.partMetadata.MaxTimestamp)
- // Use tsTable.mustAddMemPart to introduce memPart to tsTable
- tsTable.mustAddMemPart(memPart)
-
- if e := s.l.Debug(); e.Enabled() {
- e.
- Str("group", group).
- Uint32("shardID", shardID).
- Uint64("partID", memPart.partMetadata.ID).
- Msg("successfully stored memPart to tsTable")
- }
-
- return
-}
-
type indexCallback struct {
l *logger.Logger
schemaRepo *schemaRepo
diff --git a/banyand/queue/local.go b/banyand/queue/local.go
index b91a8eb0..1bb44a0c 100644
--- a/banyand/queue/local.go
+++ b/banyand/queue/local.go
@@ -148,3 +148,30 @@ func (l *localBatchPublisher) Close() (map[string]*common.Error, error) {
}
return nil, nil
}
+
+func (l *local) RegisterChunkedSyncHandler(_ bus.Topic, _ ChunkedSyncHandler) {
+}
+
+func (l *local) NewChunkedSyncClient(node string, _ uint32) (ChunkedSyncClient, error) {
+ return &localChunkedSyncClient{local: l.local, node: node}, nil
+}
+
+type localChunkedSyncClient struct {
+ local *bus.Bus
+ node string
+}
+
+func (l *localChunkedSyncClient) Close() error {
+ return nil
+}
+
+func (l *localChunkedSyncClient) SyncStreamingParts(_ context.Context, parts []StreamingPartData) (*SyncResult, error) {
+ return &SyncResult{
+ Success: true,
+ SessionID: "local-session",
+ TotalBytes: 0,
+ DurationMs: 0,
+ ChunksCount: 0,
+ PartsCount: uint32(len(parts)),
+ }, nil
+}
diff --git a/banyand/queue/pub/chunked_sync.go b/banyand/queue/pub/chunked_sync.go
new file mode 100644
index 00000000..1a6f6efe
--- /dev/null
+++ b/banyand/queue/pub/chunked_sync.go
@@ -0,0 +1,416 @@
+// 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 pub
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "hash/crc32"
+ "io"
+ "strings"
+ "time"
+
+ "google.golang.org/grpc"
+
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+const (
+ defaultChunkSize = 1024 * 1024
+ maxRetries = 3
+ retryInterval = 100 * time.Millisecond
+)
+
+type chunkedSyncClient struct {
+ client clusterv1.ServiceClient
+ conn *grpc.ClientConn
+ log *logger.Logger
+ node string
+ chunkSize uint32
+ config *ChunkedSyncClientConfig
+}
+
+// SyncStreamingParts implements queue.ChunkedSyncClient with streaming support.
+func (c *chunkedSyncClient) SyncStreamingParts(ctx context.Context, parts []queue.StreamingPartData) (*queue.SyncResult, error) {
+ if len(parts) == 0 {
+ return &queue.SyncResult{
+ Success: true,
+ SessionID: "",
+ PartsCount: 0,
+ ChunksCount: 0,
+ }, nil
+ }
+ defer func() {
+ for _, part := range parts {
+ for _, file := range part.Files {
+ fs.MustClose(file.Reader)
+ }
+ }
+ }()
+
+ sessionID := generateSessionID()
+
+ chunkedClient := clusterv1.NewChunkedSyncServiceClient(c.conn)
+
+ stream, err := chunkedClient.SyncPart(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("failed to create sync stream: %w", err)
+ }
+ defer func() {
+ if closeErr := stream.CloseSend(); closeErr != nil {
+ c.log.Error().Err(closeErr).Msg("failed to close send stream")
+ }
+ }()
+
+ startTime := time.Now()
+
+ metadata := &clusterv1.SyncMetadata{
+ Group: parts[0].Group,
+ ShardId: parts[0].ShardID,
+ Topic: parts[0].Topic,
+ Timestamp: startTime.UnixMilli(),
+ TotalParts: uint32(len(parts)),
+ }
+
+ var totalBytesSent uint64
+
+ totalChunks, err := c.streamPartsAsChunks(stream, sessionID, metadata, parts, &totalBytesSent)
+ if err != nil {
+ return nil, fmt.Errorf("failed to stream parts: %w", err)
+ }
+
+ var finalResp *clusterv1.SyncPartResponse
+ for {
+ resp, err := stream.Recv()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ return nil, fmt.Errorf("failed to receive final response: %w", err)
+ }
+ finalResp = resp
+ if resp.GetSyncResult() != nil {
+ break
+ }
+ }
+
+ duration := time.Since(startTime)
+ success := false
+
+ if finalResp != nil && finalResp.GetSyncResult() != nil {
+ result := finalResp.GetSyncResult()
+ success = result.Success
+ }
+
+ return &queue.SyncResult{
+ Success: success,
+ SessionID: sessionID,
+ TotalBytes: totalBytesSent,
+ DurationMs: duration.Milliseconds(),
+ ChunksCount: totalChunks,
+ PartsCount: uint32(len(parts)),
+ }, nil
+}
+
+// Close implements queue.ChunkedSyncClient.
+func (c *chunkedSyncClient) Close() error {
+ return nil
+}
+
+func (c *chunkedSyncClient) streamPartsAsChunks(
+ stream clusterv1.ChunkedSyncService_SyncPartClient,
+ sessionID string,
+ metadata *clusterv1.SyncMetadata,
+ parts []queue.StreamingPartData,
+ totalBytesSent *uint64,
+) (uint32, error) {
+ var totalChunks uint32
+ var chunkIndex uint32
+ isFirstChunk := true
+
+ buffer := make([]byte, 0, c.chunkSize)
+
+ type fileState struct {
+ info queue.FileInfo
+ partIndex int
+ fileIndex int
+ filesCount int
+ bytesRead uint64
+ finished bool
+ }
+
+ type chunkFileInfo struct {
+ fileInfo *clusterv1.FileInfo
+ filePartIdx int
+ }
+
+ var fileStates []*fileState
+
+ for partIdx, part := range parts {
+ for fileIdx, file := range part.Files {
+ fileStates = append(fileStates, &fileState{
+ partIndex: partIdx,
+ fileIndex: fileIdx,
+ filesCount: len(part.Files),
+ info: file,
+ bytesRead: 0,
+ finished: false,
+ })
+ }
+ }
+
+ currentFileIdx := 0
+
+ for currentFileIdx < len(fileStates) {
+ var chunkFileInfos []*chunkFileInfo
+
+ for len(buffer) < cap(buffer) && currentFileIdx < len(fileStates) {
+ fileState := fileStates[currentFileIdx]
+ if fileState.finished {
+ currentFileIdx++
+ continue
+ }
+
+ availableSpace := cap(buffer) - len(buffer)
+ if availableSpace == 0 {
+ break
+ }
+
+ fileStartInChunk := len(buffer)
+ originalLen := len(buffer)
+ buffer = buffer[:originalLen+availableSpace]
+
+ n, err := fileState.info.Reader.Read(buffer[originalLen:])
+ buffer = buffer[:originalLen+n]
+
+ if errors.Is(err, io.EOF) {
+ fileState.finished = true
+ currentFileIdx++
+ } else if err != nil {
+ return totalChunks, fmt.Errorf("failed to read from file %s: %w", fileState.info.Name, err)
+ }
+
+ if n > 0 {
+ fileState.bytesRead += uint64(n)
+
+ chunkFileInfos = append(chunkFileInfos, &chunkFileInfo{
+ fileInfo: &clusterv1.FileInfo{
+ Name: fileState.info.Name,
+ Offset: uint32(fileStartInChunk),
+ Size: uint32(n),
+ },
+ filePartIdx: fileState.partIndex,
+ })
+ }
+ }
+
+ if len(buffer) > 0 {
+ var chunkPartsInfo []*clusterv1.PartInfo
+ var currentPartInfo *clusterv1.PartInfo
+ currentPartIdx := -1
+
+ for _, chunkFile := range chunkFileInfos {
+ filePartIdx := chunkFile.filePartIdx
+
+ if filePartIdx != currentPartIdx {
+ if currentPartInfo != nil {
+ chunkPartsInfo = append(chunkPartsInfo, currentPartInfo)
+ }
+
+ originalPart := parts[filePartIdx]
+ currentPartInfo = &clusterv1.PartInfo{
+ Id: originalPart.ID,
+ Files: []*clusterv1.FileInfo{},
+ CompressedSizeBytes: originalPart.CompressedSizeBytes,
+ UncompressedSizeBytes: originalPart.UncompressedSizeBytes,
+ TotalCount: originalPart.TotalCount,
+ BlocksCount: originalPart.BlocksCount,
+ MinTimestamp: originalPart.MinTimestamp,
+ MaxTimestamp: originalPart.MaxTimestamp,
+ }
+ currentPartIdx = filePartIdx
+ }
+
+ currentPartInfo.Files = append(currentPartInfo.Files, chunkFile.fileInfo)
+ }
+
+ if currentPartInfo != nil {
+ chunkPartsInfo = append(chunkPartsInfo, currentPartInfo)
+ }
+
+ if err := c.sendChunk(stream, sessionID, buffer, chunkPartsInfo, &chunkIndex, &totalChunks, totalBytesSent, isFirstChunk, metadata); err != nil {
+ return totalChunks, err
+ }
+ isFirstChunk = false
+ buffer = buffer[:0]
+ }
+
+ if len(buffer) == 0 && currentFileIdx >= len(fileStates) {
+ break
+ }
+ }
+
+ if totalChunks > 0 {
+ completionReq := &clusterv1.SyncPartRequest{
+ SessionId: sessionID,
+ ChunkIndex: chunkIndex + 1,
+ Content: &clusterv1.SyncPartRequest_Completion{
+ Completion: &clusterv1.SyncCompletion{
+ TotalBytesSent: *totalBytesSent,
+ TotalPartsSent: uint32(len(parts)),
+ TotalChunks: totalChunks,
+ },
+ },
+ }
+
+ if err := stream.Send(completionReq); err != nil {
+ return totalChunks, fmt.Errorf("failed to send completion: %w", err)
+ }
+ totalChunks++
+ }
+
+ return totalChunks, nil
+}
+
+func (c *chunkedSyncClient) sendChunk(
+ stream clusterv1.ChunkedSyncService_SyncPartClient,
+ sessionID string,
+ chunkData []byte,
+ partsInfo []*clusterv1.PartInfo,
+ chunkIndex *uint32,
+ totalChunks *uint32,
+ totalBytesSent *uint64,
+ isFirstChunk bool,
+ metadata *clusterv1.SyncMetadata,
+) error {
+ chunkChecksum := fmt.Sprintf("%x", crc32.ChecksumIEEE(chunkData))
+
+ retryCount := 0
+ for {
+ req := &clusterv1.SyncPartRequest{
+ SessionId: sessionID,
+ ChunkIndex: *chunkIndex,
+ ChunkData: chunkData,
+ ChunkChecksum: chunkChecksum,
+ PartsInfo: partsInfo,
+ }
+
+ if isFirstChunk {
+ req.Content = &clusterv1.SyncPartRequest_Metadata{
+ Metadata: metadata,
+ }
+ }
+
+ if err := stream.Send(req); err != nil {
+ return fmt.Errorf("failed to send chunk %d: %w", *chunkIndex, err)
+ }
+
+ resp, err := stream.Recv()
+ if err != nil {
+ return fmt.Errorf("failed to receive response for chunk %d: %w", *chunkIndex, err)
+ }
+
+ switch resp.Status {
+ case clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED:
+ *totalBytesSent += uint64(len(chunkData))
+ *chunkIndex++
+ *totalChunks++
+ return nil
+
+ case clusterv1.SyncStatus_SYNC_STATUS_CHUNK_CHECKSUM_MISMATCH:
+ retryCount++
+ if retryCount > maxRetries {
+ return fmt.Errorf("chunk %d checksum mismatch after %d retries", *chunkIndex, maxRetries)
+ }
+ c.log.Warn().Int("retry_count", retryCount).Uint32("chunk_index", *chunkIndex).Msg("chunk checksum mismatch, retrying")
+ time.Sleep(retryInterval * time.Duration(retryCount))
+ continue
+
+ case clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER:
+ if err := c.handleOutOfOrderResponse(resp, *chunkIndex, &retryCount); err != nil {
+ return err
+ }
+ if strings.Contains(resp.Error, "buffered") {
+ *totalBytesSent += uint64(len(chunkData))
+ *chunkIndex++
+ *totalChunks++
+ return nil
+ }
+ continue
+
+ case clusterv1.SyncStatus_SYNC_STATUS_SESSION_NOT_FOUND:
+ return fmt.Errorf("session %s not found on server for chunk %d: %s", sessionID, *chunkIndex, resp.Error)
+
+ default:
+ if resp.Error != "" {
+ return fmt.Errorf("chunk %d sync failed: %s", *chunkIndex, resp.Error)
+ }
+ return fmt.Errorf("chunk %d sync failed with status: %v", *chunkIndex, resp.Status)
+ }
+ }
+}
+
+func (c *chunkedSyncClient) handleOutOfOrderResponse(resp *clusterv1.SyncPartResponse, chunkIndex uint32, retryCount *int) error {
+ config := c.config
+ if config == nil {
+ config = &ChunkedSyncClientConfig{
+ EnableRetryOnOOO: true,
+ MaxOOORetries: 3,
+ OOORetryDelay: 100 * time.Millisecond,
+ }
+ }
+
+ if !config.EnableRetryOnOOO {
+ return fmt.Errorf("chunk %d out of order: %s", chunkIndex, resp.Error)
+ }
+
+ if strings.Contains(resp.Error, "buffered") {
+ c.log.Info().
+ Uint32("chunk_index", chunkIndex).
+ Str("server_message", resp.Error).
+ Msg("chunk was buffered by server due to reordering")
+ return nil
+ }
+
+ if strings.Contains(resp.Error, "gap too large") || strings.Contains(resp.Error, "buffer full") {
+ return fmt.Errorf("unrecoverable out-of-order error for chunk %d: %s", chunkIndex, resp.Error)
+ }
+
+ *retryCount++
+ if *retryCount > config.MaxOOORetries {
+ return fmt.Errorf("chunk %d out of order after %d retries: %s",
+ chunkIndex, config.MaxOOORetries, resp.Error)
+ }
+
+ c.log.Warn().
+ Int("retry_count", *retryCount).
+ Uint32("chunk_index", chunkIndex).
+ Str("error", resp.Error).
+ Msg("retrying out-of-order chunk")
+
+ time.Sleep(config.OOORetryDelay * time.Duration(*retryCount))
+ return nil
+}
+
+func generateSessionID() string {
+ return fmt.Sprintf("sync-%d", time.Now().UnixNano())
+}
diff --git a/banyand/queue/pub/chunked_sync_test.go b/banyand/queue/pub/chunked_sync_test.go
new file mode 100644
index 00000000..693c83ff
--- /dev/null
+++ b/banyand/queue/pub/chunked_sync_test.go
@@ -0,0 +1,242 @@
+// 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 pub
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "sync"
+ "time"
+
+ "github.com/onsi/ginkgo/v2"
+ "github.com/onsi/gomega"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
+
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+type mockChunkedSyncServer struct {
+ clusterv1.UnimplementedChunkedSyncServiceServer
+ checksumMismatchCount int
+ maxMismatchResponses int
+ mu sync.Mutex
+}
+
+func (m *mockChunkedSyncServer) SyncPart(stream clusterv1.ChunkedSyncService_SyncPartServer) error {
+ for {
+ req, err := stream.Recv()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ return err
+ }
+
+ if req.GetCompletion() != nil {
+ return stream.Send(&clusterv1.SyncPartResponse{
+ SessionId: req.SessionId,
+ ChunkIndex: req.ChunkIndex,
+ Status: clusterv1.SyncStatus_SYNC_STATUS_SYNC_COMPLETE,
+ SyncResult: &clusterv1.SyncResult{
+ Success: true,
+ },
+ })
+ }
+
+ m.mu.Lock()
+ shouldMismatch := m.checksumMismatchCount < m.maxMismatchResponses
+ if shouldMismatch {
+ m.checksumMismatchCount++
+ }
+ m.mu.Unlock()
+
+ var status clusterv1.SyncStatus
+ var errorMsg string
+
+ if shouldMismatch {
+ status = clusterv1.SyncStatus_SYNC_STATUS_CHUNK_CHECKSUM_MISMATCH
+ errorMsg = fmt.Sprintf("chunk %d checksum mismatch", req.ChunkIndex)
+ } else {
+ status = clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED
+ }
+
+ resp := &clusterv1.SyncPartResponse{
+ SessionId: req.SessionId,
+ ChunkIndex: req.ChunkIndex,
+ Status: status,
+ Error: errorMsg,
+ }
+
+ if err := stream.Send(resp); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func setupChunkedSyncServer(address string, maxMismatchResponses int) (*mockChunkedSyncServer, func()) {
+ s := grpc.NewServer()
+ mockServer := &mockChunkedSyncServer{
+ maxMismatchResponses: maxMismatchResponses,
+ }
+ clusterv1.RegisterChunkedSyncServiceServer(s, mockServer)
+
+ lis, err := net.Listen("tcp", address)
+ if err != nil {
+ logger.Panicf("failed to listen: %v", err)
+ return nil, nil
+ }
+
+ go func() {
+ if err := s.Serve(lis); err != nil {
+ logger.Panicf("Server exited with error: %v", err)
+ }
+ }()
+
+ return mockServer, s.GracefulStop
+}
+
+var _ = ginkgo.Describe("Chunked Sync Retry Mechanism", func() {
+ ginkgo.Context("when checksum mismatches occur", func() {
+ ginkgo.It("should retry and succeed within max retries", func() {
+ address := getAddress()
+ mockServer, cleanup := setupChunkedSyncServer(address, 2)
+ defer cleanup()
+
+ conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ defer conn.Close()
+
+ client := &chunkedSyncClient{
+ conn: conn,
+ log: logger.GetLogger("test-chunked-sync"),
+ chunkSize: 1024,
+ }
+
+ parts := []queue.StreamingPartData{
+ {
+ ID: 1,
+ Group: "test-group",
+ ShardID: 1,
+ Topic: "stream_write",
+ Files: []queue.FileInfo{
+ createFileInfo("test-file.dat", []byte("test data for chunked sync")),
+ },
+ },
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+
+ result, err := client.SyncStreamingParts(ctx, parts)
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ gomega.Expect(result.Success).To(gomega.BeTrue())
+ gomega.Expect(mockServer.checksumMismatchCount).To(gomega.Equal(2))
+ })
+
+ ginkgo.It("should fail after exceeding max retries", func() {
+ address := getAddress()
+ mockServer, cleanup := setupChunkedSyncServer(address, maxRetries+1)
+ defer cleanup()
+
+ conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ defer conn.Close()
+
+ client := &chunkedSyncClient{
+ conn: conn,
+ log: logger.GetLogger("test-chunked-sync"),
+ chunkSize: 1024,
+ }
+
+ parts := []queue.StreamingPartData{
+ {
+ ID: 1,
+ Group: "test-group",
+ ShardID: 1,
+ Topic: "stream_write",
+ Files: []queue.FileInfo{
+ createFileInfo("test-file.dat", []byte("test data for chunked sync")),
+ },
+ },
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+
+ _, err = client.SyncStreamingParts(ctx, parts)
+ gomega.Expect(err).To(gomega.HaveOccurred())
+ gomega.Expect(err.Error()).To(gomega.ContainSubstring("checksum mismatch after"))
+ gomega.Expect(mockServer.checksumMismatchCount).To(gomega.Equal(maxRetries + 1))
+ })
+
+ ginkgo.It("should handle no retry needed scenario", func() {
+ address := getAddress()
+ mockServer, cleanup := setupChunkedSyncServer(address, 0)
+ defer cleanup()
+
+ conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ defer conn.Close()
+
+ client := &chunkedSyncClient{
+ conn: conn,
+ log: logger.GetLogger("test-chunked-sync"),
+ chunkSize: 1024,
+ }
+
+ parts := []queue.StreamingPartData{
+ {
+ ID: 1,
+ Group: "test-group",
+ ShardID: 1,
+ Topic: "stream_write",
+ Files: []queue.FileInfo{
+ createFileInfo("test-file.dat", []byte("test data for chunked sync")),
+ },
+ },
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ result, err := client.SyncStreamingParts(ctx, parts)
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ gomega.Expect(result.Success).To(gomega.BeTrue())
+ gomega.Expect(mockServer.checksumMismatchCount).To(gomega.Equal(0))
+ })
+ })
+})
+
+func createFileInfo(name string, data []byte) queue.FileInfo {
+ var buf bytes.Buffer
+ if _, err := buf.Write(data); err != nil {
+ panic(fmt.Sprintf("failed to write content to buffer: %v", err))
+ }
+ return queue.FileInfo{
+ Name: name,
+ Reader: buf.SequentialRead(),
+ }
+}
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 40f79043..6f117f2a 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -47,6 +47,14 @@ import (
"github.com/apache/skywalking-banyandb/pkg/run"
)
+// ChunkedSyncClientConfig configures chunked sync client behavior.
+type ChunkedSyncClientConfig struct {
+ ChunkSize uint32 // Size of each chunk in bytes
+ EnableRetryOnOOO bool // Enable retry on out-of-order errors
+ MaxOOORetries int // Maximum retries for out-of-order chunks
+ OOORetryDelay time.Duration // Delay between retries
+}
+
var (
_ run.PreRunner = (*pub)(nil)
_ run.Service = (*pub)(nil)
@@ -403,3 +411,36 @@ func (p *pub) getClientTransportCredentials() ([]grpc.DialOption, error) {
}
return opts, nil
}
+
+// NewChunkedSyncClient implements queue.Client.
+func (p *pub) NewChunkedSyncClient(node string, chunkSize uint32) (queue.ChunkedSyncClient, error) {
+ return p.NewChunkedSyncClientWithConfig(node, &ChunkedSyncClientConfig{
+ ChunkSize: chunkSize,
+ EnableRetryOnOOO: true,
+ MaxOOORetries: 3,
+ OOORetryDelay: 100 * time.Millisecond,
+ })
+}
+
+// NewChunkedSyncClientWithConfig creates a chunked sync client with advanced configuration.
+func (p *pub) NewChunkedSyncClientWithConfig(node string, config *ChunkedSyncClientConfig) (queue.ChunkedSyncClient, error) {
+ p.mu.RLock()
+ client, ok := p.active[node]
+ p.mu.RUnlock()
+ if !ok {
+ return nil, fmt.Errorf("no active client for node %s", node)
+ }
+
+ if config.ChunkSize == 0 {
+ config.ChunkSize = defaultChunkSize
+ }
+
+ return &chunkedSyncClient{
+ client: client.client,
+ conn: client.conn,
+ node: node,
+ log: p.log,
+ chunkSize: config.ChunkSize,
+ config: config,
+ }, nil
+}
diff --git a/banyand/queue/queue.go b/banyand/queue/queue.go
index 73a9e695..4f2c22f0 100644
--- a/banyand/queue/queue.go
+++ b/banyand/queue/queue.go
@@ -18,11 +18,14 @@
package queue
import (
+ "context"
"time"
"github.com/apache/skywalking-banyandb/api/common"
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/run"
)
@@ -43,6 +46,7 @@ type Client interface {
bus.Publisher
bus.Broadcaster
NewBatchPublisher(timeout time.Duration) BatchPublisher
+ NewChunkedSyncClient(node string, chunkSize uint32) (ChunkedSyncClient, error)
Register(bus.Topic, schema.EventHandler)
OnAddOrUpdate(md schema.Metadata)
GracefulStop()
@@ -52,6 +56,7 @@ type Client interface {
type Server interface {
run.Unit
bus.Subscriber
+ RegisterChunkedSyncHandler(topic bus.Topic, handler ChunkedSyncHandler)
GetPort() *uint32
}
@@ -62,3 +67,97 @@ type BatchPublisher interface {
bus.Publisher
Close() (map[string]*common.Error, error)
}
+
+// FileInfo represents information about an individual file within a part.
+type FileInfo struct {
+ Reader fs.SeqReader
+ Name string
+}
+
+// StreamingPartData represents streaming part data for syncing.
+type StreamingPartData struct {
+ Group string
+ Topic string
+ Files []FileInfo
+ ID uint64
+ CompressedSizeBytes uint64
+ UncompressedSizeBytes uint64
+ TotalCount uint64
+ BlocksCount uint64
+ MinTimestamp int64
+ MaxTimestamp int64
+ ShardID uint32
+}
+
+// ChunkedSyncClient provides methods for chunked data synchronization with streaming support.
+type ChunkedSyncClient interface {
+ // SyncStreamingParts sends streaming parts using chunked transfer.
+ SyncStreamingParts(ctx context.Context, parts []StreamingPartData) (*SyncResult, error)
+ // Close releases resources associated with the client.
+ Close() error
+}
+
+// ChunkedSyncPartContext represents the context for a chunked sync operation.
+type ChunkedSyncPartContext struct {
+ Handler PartHandler
+ Group string
+ FileName string
+ ID uint64
+ CompressedSizeBytes uint64
+ UncompressedSizeBytes uint64
+ TotalCount uint64
+ BlocksCount uint64
+ MinTimestamp int64
+ MaxTimestamp int64
+ ShardID uint32
+}
+
+// Close releases resources associated with the context.
+func (c *ChunkedSyncPartContext) Close() error {
+ defer func() {
+ c.Handler = nil
+ c.FileName = ""
+ }()
+ if c.Handler != nil {
+ return c.Handler.Close()
+ }
+ return nil
+}
+
+// PartHandler handles individual parts during sync operations.
+type PartHandler interface {
+ FinishSync() error
+ Close() error
+}
+
+// ChunkedSyncHandler handles incoming chunked sync requests on the server side.
+//
+//go:generate mockgen -destination=./chunked_sync_handler_mock.go -package=queue . ChunkedSyncHandler
+type ChunkedSyncHandler interface {
+ // HandleFileChunk processes incoming file chunks as they arrive.
+ // This method is called for each chunk of file data to enable streaming processing.
+ HandleFileChunk(ctx *ChunkedSyncPartContext, chunk []byte) error
+
+ // CreatePartHandler creates a new part handler for the given context.
+ CreatePartHandler(ctx *ChunkedSyncPartContext) (PartHandler, error)
+}
+
+// FileData represents file data with name and content.
+type FileData struct {
+ FileName string
+ Data []byte
+}
+
+// SyncResult represents the result of a sync operation.
+type SyncResult struct {
+ SessionID string
+ ErrorMessage string
+ TotalBytes uint64
+ DurationMs int64
+ ChunksCount uint32
+ PartsCount uint32
+ Success bool
+}
+
+// SyncMetadata is an alias for clusterv1.SyncMetadata.
+type SyncMetadata = clusterv1.SyncMetadata
diff --git a/banyand/queue/sub/chunked_sync.go b/banyand/queue/sub/chunked_sync.go
new file mode 100644
index 00000000..ce300b97
--- /dev/null
+++ b/banyand/queue/sub/chunked_sync.go
@@ -0,0 +1,458 @@
+// 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 sub
+
+import (
+ "errors"
+ "fmt"
+ "hash/crc32"
+ "io"
+ "time"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+)
+
+type chunkBuffer struct {
+ chunks map[uint32]*clusterv1.SyncPartRequest
+ expectedIndex uint32
+ maxBufferSize uint32
+ bufferTimeout time.Duration
+ lastActivity time.Time
+}
+
+type syncSession struct {
+ startTime time.Time
+ metadata *clusterv1.SyncMetadata
+ partsProgress map[int]*partProgress
+ partCtx *queue.ChunkedSyncPartContext
+ sessionID string
+ errorMsg string
+ totalReceived uint64
+ chunksReceived uint32
+ completed bool
+ chunkBuffer *chunkBuffer
+}
+
+type partProgress struct {
+ totalBytes uint32
+ receivedBytes uint32
+ completed bool
+}
+
+// SyncPart implements clusterv1.ChunkedSyncServiceServer.
+func (s *server) SyncPart(stream clusterv1.ChunkedSyncService_SyncPartServer) error {
+ ctx := stream.Context()
+ var currentSession *syncSession
+ var sessionID string
+ defer func() {
+ if currentSession != nil {
+ if currentSession.partCtx != nil {
+ currentSession.partCtx.Close()
+ }
+ }
+ }()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ default:
+ }
+
+ req, err := stream.Recv()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ s.log.Error().Err(err).Msg("failed to receive chunk")
+ return err
+ }
+
+ sessionID = req.SessionId
+
+ if req.GetMetadata() != nil {
+ currentSession = &syncSession{
+ sessionID: sessionID,
+ metadata: req.GetMetadata(),
+ startTime: time.Now(),
+ chunksReceived: 0,
+ partsProgress: make(map[int]*partProgress),
+ }
+ s.log.Info().Str("session_id", sessionID).
+ Str("topic", req.GetMetadata().Topic).
+ Uint32("total_parts", req.GetMetadata().TotalParts).
+ Msg("started chunked sync session")
+ }
+
+ if currentSession == nil {
+ errMsg := fmt.Sprintf("session %s not found", sessionID)
+ s.log.Error().Str("session_id", sessionID).Msg(errMsg)
+ return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_SESSION_NOT_FOUND, errMsg, nil)
+ }
+
+ if req.GetCompletion() != nil {
+ return s.handleCompletion(stream, currentSession, req)
+ }
+
+ if err := s.processChunk(stream, currentSession, req); err != nil {
+ s.log.Error().Err(err).Str("session_id", sessionID).Msg("failed to process chunk")
+ return err
+ }
+ }
+
+ return nil
+}
+
+func (s *server) processChunk(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
+ if !s.enableChunkReordering {
+ return s.processChunkSequential(stream, session, req)
+ }
+
+ if session.chunkBuffer == nil {
+ session.chunkBuffer = &chunkBuffer{
+ chunks: make(map[uint32]*clusterv1.SyncPartRequest),
+ expectedIndex: 0,
+ maxBufferSize: s.maxChunkBufferSize,
+ bufferTimeout: s.chunkBufferTimeout,
+ lastActivity: time.Now(),
+ }
+ }
+
+ return s.processChunkWithReordering(stream, session, req)
+}
+
+func (s *server) processChunkSequential(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
+ if req.ChunkIndex != session.chunksReceived {
+ errMsg := fmt.Sprintf("out of order chunk received: expected %d, got %d",
+ session.chunksReceived, req.ChunkIndex)
+ s.log.Warn().Str("session_id", req.SessionId).
+ Uint32("expected_chunk", session.chunksReceived).
+ Uint32("received_chunk", req.ChunkIndex).
+ Msg("out of order chunk received")
+ return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, errMsg, nil)
+ }
+
+ return s.processExpectedChunk(stream, session, req)
+}
+
+func (s *server) processChunkWithReordering(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
+ buffer := session.chunkBuffer
+ buffer.lastActivity = time.Now()
+
+ if req.ChunkIndex == buffer.expectedIndex {
+ if err := s.processExpectedChunk(stream, session, req); err != nil {
+ return err
+ }
+ buffer.expectedIndex++
+
+ return s.processBufferedChunks(stream, session)
+ }
+
+ if req.ChunkIndex > buffer.expectedIndex {
+ gap := req.ChunkIndex - buffer.expectedIndex
+ s.updateChunkOrderMetrics("out_of_order_received", req.SessionId)
+
+ if gap > s.maxChunkGapSize {
+ errMsg := fmt.Sprintf("chunk gap too large: expected %d, got %d (gap: %d > max: %d)",
+ buffer.expectedIndex, req.ChunkIndex, gap, s.maxChunkGapSize)
+ s.log.Warn().Str("session_id", req.SessionId).
+ Uint32("expected_chunk", buffer.expectedIndex).
+ Uint32("received_chunk", req.ChunkIndex).
+ Uint32("gap", gap).
+ Uint32("max_gap", s.maxChunkGapSize).
+ Msg("chunk gap too large, rejecting")
+ s.updateChunkOrderMetrics("gap_too_large", req.SessionId)
+ return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, errMsg, nil)
+ }
+
+ if uint32(len(buffer.chunks)) >= buffer.maxBufferSize {
+ errMsg := fmt.Sprintf("chunk buffer full, cannot store out-of-order chunk %d", req.ChunkIndex)
+ s.log.Warn().Str("session_id", req.SessionId).
+ Uint32("chunk_index", req.ChunkIndex).
+ Uint32("buffer_size", uint32(len(buffer.chunks))).
+ Uint32("max_buffer_size", buffer.maxBufferSize).
+ Msg("chunk buffer full, rejecting chunk")
+ s.updateChunkOrderMetrics("buffer_full", req.SessionId)
+ return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, errMsg, nil)
+ }
+
+ buffer.chunks[req.ChunkIndex] = req
+ s.log.Info().Str("session_id", req.SessionId).
+ Uint32("chunk_index", req.ChunkIndex).
+ Uint32("expected_index", buffer.expectedIndex).
+ Uint32("buffered_chunks", uint32(len(buffer.chunks))).
+ Msg("buffered out-of-order chunk")
+ s.updateChunkOrderMetrics("chunk_buffered", req.SessionId)
+
+ return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED,
+ fmt.Sprintf("chunk %d buffered (waiting for %d)", req.ChunkIndex, buffer.expectedIndex), nil)
+ }
+
+ if req.ChunkIndex < buffer.expectedIndex {
+ s.log.Warn().
+ Str("session_id", req.SessionId).
+ Uint32("chunk_index", req.ChunkIndex).
+ Uint32("expected_index", buffer.expectedIndex).
+ Msg("received duplicate or old chunk, ignoring")
+
+ return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED,
+ fmt.Sprintf("duplicate chunk %d ignored", req.ChunkIndex), nil)
+ }
+
+ return nil
+}
+
+func (s *server) processExpectedChunk(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
+ calculatedChecksum := fmt.Sprintf("%x", crc32.ChecksumIEEE(req.ChunkData))
+ if calculatedChecksum != req.ChunkChecksum {
+ errMsg := fmt.Sprintf("chunk %d checksum mismatch: expected %s, got %s",
+ req.ChunkIndex, req.ChunkChecksum, calculatedChecksum)
+ s.log.Warn().Str("session_id", req.SessionId).Msg(errMsg)
+ return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_CHECKSUM_MISMATCH, errMsg, nil)
+ }
+
+ session.totalReceived += uint64(len(req.ChunkData))
+ session.chunksReceived++
+
+ var topic bus.Topic
+ t, ok := data.TopicMap[session.metadata.Topic]
+ if !ok {
+ return fmt.Errorf("unknown sync topic: %s", session.metadata.Topic)
+ }
+ topic = t
+
+ handler, exists := s.chunkedSyncHandlers[topic]
+ if !exists {
+ return fmt.Errorf("no handler registered for topic %s", topic)
+ }
+
+ for partIndex, partInfo := range req.PartsInfo {
+ if session.partCtx == nil {
+ session.partCtx = &queue.ChunkedSyncPartContext{
+ ID: partInfo.Id,
+ Group: session.metadata.Group,
+ ShardID: session.metadata.ShardId,
+ CompressedSizeBytes: partInfo.CompressedSizeBytes,
+ UncompressedSizeBytes: partInfo.UncompressedSizeBytes,
+ TotalCount: partInfo.TotalCount,
+ BlocksCount: partInfo.BlocksCount,
+ MinTimestamp: partInfo.MinTimestamp,
+ MaxTimestamp: partInfo.MaxTimestamp,
+ }
+ } else if session.partCtx.ID != partInfo.Id {
+ if session.partCtx.Handler != nil {
+ if err := session.partCtx.Handler.FinishSync(); err != nil {
+ return fmt.Errorf("failed to complete part %d: %w", session.partCtx.ID, err)
+ }
+ session.partCtx.Handler = nil
+ }
+ session.partCtx.ID = partInfo.Id
+ session.partCtx.CompressedSizeBytes = partInfo.CompressedSizeBytes
+ session.partCtx.UncompressedSizeBytes = partInfo.UncompressedSizeBytes
+ session.partCtx.TotalCount = partInfo.TotalCount
+ session.partCtx.BlocksCount = partInfo.BlocksCount
+ session.partCtx.MinTimestamp = partInfo.MinTimestamp
+ session.partCtx.MaxTimestamp = partInfo.MaxTimestamp
+ }
+
+ if session.partCtx.Handler == nil {
+ partHandler, err := handler.CreatePartHandler(session.partCtx)
+ if err != nil {
+ return fmt.Errorf("failed to create part handler: %w", err)
+ }
+ session.partCtx.Handler = partHandler
+ }
+
+ if err := s.processPart(session, req, partInfo, partIndex, handler); err != nil {
+ s.log.Error().Err(err).
+ Str("session_id", req.SessionId).
+ Int("part_index", partIndex).
+ Msg("failed to process part")
+ return err
+ }
+ }
+
+ return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, "", nil)
+}
+
+func (s *server) processBufferedChunks(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession) error {
+ buffer := session.chunkBuffer
+
+ for {
+ if chunk, exists := buffer.chunks[buffer.expectedIndex]; exists {
+ delete(buffer.chunks, buffer.expectedIndex)
+
+ s.log.Debug().Str("session_id", session.sessionID).
+ Uint32("chunk_index", buffer.expectedIndex).
+ Uint32("remaining_buffered", uint32(len(buffer.chunks))).
+ Msg("processing buffered chunk")
+
+ if err := s.processExpectedChunk(stream, session, chunk); err != nil {
+ return err
+ }
+ buffer.expectedIndex++
+ } else {
+ break
+ }
+ }
+
+ return nil
+}
+
+func (s *server) checkBufferTimeout(session *syncSession) error {
+ if session.chunkBuffer == nil {
+ return nil
+ }
+
+ if time.Since(session.chunkBuffer.lastActivity) > session.chunkBuffer.bufferTimeout {
+ if len(session.chunkBuffer.chunks) > 0 {
+ missing := make([]uint32, 0)
+ for i := session.chunkBuffer.expectedIndex; i < session.chunkBuffer.expectedIndex+10; i++ {
+ if _, exists := session.chunkBuffer.chunks[i]; !exists {
+ missing = append(missing, i)
+ }
+ }
+
+ return fmt.Errorf("buffer timeout: missing chunks %v after %v",
+ missing, session.chunkBuffer.bufferTimeout)
+ }
+ }
+
+ return nil
+}
+
+func (s *server) processPart(session *syncSession, req *clusterv1.SyncPartRequest, partInfo *clusterv1.PartInfo, partIndex int, handler queue.ChunkedSyncHandler) error {
+ var totalPartSizeInChunk uint32
+ for _, fileInfo := range partInfo.Files {
+ totalPartSizeInChunk += fileInfo.Size
+ }
+ progress := session.partsProgress[partIndex]
+ if progress == nil {
+ session.partsProgress[partIndex] = &partProgress{
+ totalBytes: totalPartSizeInChunk,
+ receivedBytes: 0,
+ completed: false,
+ }
+ progress = session.partsProgress[partIndex]
+ } else {
+ progress.completed = false
+ progress.totalBytes += totalPartSizeInChunk
+ }
+
+ var partDataSize uint32
+ for _, fileInfo := range partInfo.Files {
+ if fileInfo.Offset >= uint32(len(req.ChunkData)) {
+ s.log.Warn().Str("session_id", session.sessionID).
+ Str("file_name", fileInfo.Name).
+ Uint32("offset", fileInfo.Offset).
+ Uint32("chunk_size", uint32(len(req.ChunkData))).
+ Msg("file data not in this chunk")
+ continue
+ }
+
+ fileEndOffset := fileInfo.Offset + fileInfo.Size
+ actualFileSize := fileInfo.Size
+ if fileEndOffset > uint32(len(req.ChunkData)) {
+ actualFileSize = uint32(len(req.ChunkData)) - fileInfo.Offset
+ }
+
+ fileChunk := req.ChunkData[fileInfo.Offset : fileInfo.Offset+actualFileSize]
+ partDataSize += actualFileSize
+
+ session.partCtx.FileName = fileInfo.Name
+
+ if err := handler.HandleFileChunk(session.partCtx, fileChunk); err != nil {
+ return fmt.Errorf("failed to stream file chunk for %s: %w", fileInfo.Name, err)
+ }
+ }
+
+ progress.receivedBytes += partDataSize
+ if progress.receivedBytes > progress.totalBytes {
+ return fmt.Errorf("received more bytes (%d) than expected (%d) for part %d",
+ progress.receivedBytes, progress.totalBytes, partIndex)
+ }
+
+ progress.completed = progress.receivedBytes == progress.totalBytes
+ return nil
+}
+
+func (s *server) handleCompletion(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
+ if session.partCtx.Handler != nil {
+ if err := session.partCtx.Handler.FinishSync(); err != nil {
+ return fmt.Errorf("failed to complete part %d: %w", session.partCtx.ID, err)
+ }
+ session.partCtx.Handler = nil
+ }
+
+ session.completed = true
+
+ partsResults := make([]*clusterv1.PartResult, 0, len(session.partsProgress))
+ allPartsSuccessful := true
+
+ for _, progress := range session.partsProgress {
+ success := progress.completed
+ if !success {
+ allPartsSuccessful = false
+ }
+
+ partsResults = append(partsResults, &clusterv1.PartResult{
+ Success: success,
+ Error: session.errorMsg,
+ BytesProcessed: progress.receivedBytes,
+ })
+ }
+
+ syncResult := &clusterv1.SyncResult{
+ Success: allPartsSuccessful,
+ TotalBytesReceived: session.totalReceived,
+ DurationMs: time.Since(session.startTime).Milliseconds(),
+ ChunksReceived: session.chunksReceived,
+ PartsReceived: uint32(len(session.partsProgress)),
+ PartsResults: partsResults,
+ }
+
+ s.log.Info().
+ Str("session_id", session.sessionID).
+ Bool("success", syncResult.Success).
+ Uint64("bytes_received", syncResult.TotalBytesReceived).
+ Int64("duration_ms", syncResult.DurationMs).
+ Msg("completed chunked sync session")
+
+ return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_SYNC_COMPLETE, "", syncResult)
+}
+
+func (s *server) sendResponse(
+ stream clusterv1.ChunkedSyncService_SyncPartServer,
+ req *clusterv1.SyncPartRequest,
+ status clusterv1.SyncStatus,
+ errorMsg string,
+ syncResult *clusterv1.SyncResult,
+) error {
+ resp := &clusterv1.SyncPartResponse{
+ SessionId: req.SessionId,
+ ChunkIndex: req.ChunkIndex,
+ Status: status,
+ Error: errorMsg,
+ SyncResult: syncResult,
+ }
+
+ return stream.Send(resp)
+}
diff --git a/banyand/queue/sub/chunked_sync_test.go b/banyand/queue/sub/chunked_sync_test.go
new file mode 100644
index 00000000..c18906ce
--- /dev/null
+++ b/banyand/queue/sub/chunked_sync_test.go
@@ -0,0 +1,556 @@
+// 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 sub
+
+import (
+ "context"
+ "fmt"
+ "hash/crc32"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "google.golang.org/grpc/metadata"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+func TestChunkedSyncOutOfOrderHandling(t *testing.T) {
+ // Initialize logger for tests
+ err := logger.Init(logger.Logging{
+ Env: "dev",
+ Level: "debug",
+ })
+ require.NoError(t, err)
+
+ tests := []struct {
+ name string
+ enableChunkReordering bool
+ maxChunkBufferSize uint32
+ maxChunkGapSize uint32
+ chunkSequence []uint32
+ expectedStatus []clusterv1.SyncStatus
+ expectedBuffered int
+ }{
+ {
+ name: "strict_sequential_mode_rejects_out_of_order",
+ enableChunkReordering: false,
+ chunkSequence: []uint32{0, 2}, // Missing chunk 1
+ expectedStatus: []clusterv1.SyncStatus{clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER},
+ expectedBuffered: 0,
+ },
+ {
+ name: "reordering_mode_buffers_out_of_order_chunks",
+ enableChunkReordering: true,
+ maxChunkBufferSize: 10,
+ maxChunkGapSize: 5,
+ chunkSequence: []uint32{0, 2, 1}, // Chunk 2 arrives before 1
+ expectedStatus: []clusterv1.SyncStatus{clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED},
+ expectedBuffered: 1, // Chunk 2 gets buffered initially
+ },
+ {
+ name: "large_gap_rejected",
+ enableChunkReordering: true,
+ maxChunkBufferSize: 10,
+ maxChunkGapSize: 3,
+ chunkSequence: []uint32{0, 5}, // Gap of 5 > maxGapSize of 3
+ expectedStatus: []clusterv1.SyncStatus{clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER},
+ expectedBuffered: 0,
+ },
+ {
+ name: "buffer_full_rejection",
+ enableChunkReordering: true,
+ maxChunkBufferSize: 2,
+ maxChunkGapSize: 10,
+ chunkSequence: []uint32{0, 3, 4, 5}, // Fill buffer with chunks 3,4 then try to add 5
+ expectedStatus: []clusterv1.SyncStatus{clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER},
+ expectedBuffered: 2, // Chunks 3,4 get buffered, 5 rejected
+ },
+ {
+ name: "duplicate_chunk_ignored",
+ enableChunkReordering: true,
+ maxChunkBufferSize: 10,
+ maxChunkGapSize: 5,
+ chunkSequence: []uint32{0, 1, 0}, // Duplicate chunk 0
+ expectedStatus: []clusterv1.SyncStatus{clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED},
+ expectedBuffered: 0,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Create a mock server instance with specified configuration
+ s := &server{
+ log: logger.GetLogger("test-server-" + tt.name),
+ chunkedSyncHandlers: make(map[bus.Topic]queue.ChunkedSyncHandler),
+ enableChunkReordering: tt.enableChunkReordering,
+ maxChunkBufferSize: tt.maxChunkBufferSize,
+ maxChunkGapSize: tt.maxChunkGapSize,
+ }
+
+ // Register a mock handler
+ mockHandler := &MockChunkedSyncHandler{}
+ s.chunkedSyncHandlers[data.TopicStreamPartSync] = mockHandler
+
+ // Create mock stream
+ mockStream := &MockSyncPartStream{}
+
+ // Create a session
+ session := &syncSession{
+ sessionID: "test-session",
+ startTime: time.Now(),
+ chunksReceived: 0,
+ metadata: &clusterv1.SyncMetadata{
+ Topic: data.TopicStreamPartSync.String(),
+ Group: "test-group",
+ },
+ partsProgress: make(map[int]*partProgress),
+ }
+
+ // Process chunks in the specified sequence
+ bufferedCount := 0
+ for i, chunkIndex := range tt.chunkSequence {
+ req := &clusterv1.SyncPartRequest{
+ SessionId: session.sessionID,
+ ChunkIndex: chunkIndex,
+ ChunkData: []byte(fmt.Sprintf("test-data-%d", chunkIndex)),
+ ChunkChecksum: fmt.Sprintf("%x", crc32.ChecksumIEEE([]byte(fmt.Sprintf("test-data-%d", chunkIndex)))),
+ PartsInfo: []*clusterv1.PartInfo{
+ {
+ Id: uint64(chunkIndex),
+ Files: []*clusterv1.FileInfo{
+ {
+ Name: "test-file",
+ Offset: 0,
+ Size: uint32(len(fmt.Sprintf("test-data-%d", chunkIndex))),
+ },
+ },
+ },
+ },
+ }
+
+ responsesBefore := len(mockStream.sentResponses)
+ err := s.processChunk(mockStream, session, req)
+ require.NoError(t, err, "processChunk should not return error for chunk %d", chunkIndex)
+
+ // Verify at least one response was sent
+ responsesAfter := len(mockStream.sentResponses)
+ assert.True(t, responsesAfter > responsesBefore, "Should have sent at least one response for chunk %d", chunkIndex)
+
+ // For the last response sent, check if it matches expected status (if provided)
+ if i < len(tt.expectedStatus) {
+ lastResp := mockStream.sentResponses[responsesAfter-1]
+ assert.Equal(t, tt.expectedStatus[i], lastResp.Status, "Unexpected status for chunk %d", chunkIndex)
+ }
+
+ // Count buffered chunks
+ if session.chunkBuffer != nil && len(session.chunkBuffer.chunks) > bufferedCount {
+ bufferedCount = len(session.chunkBuffer.chunks)
+ }
+ }
+
+ // Verify expected number of chunks were buffered
+ assert.Equal(t, tt.expectedBuffered, bufferedCount, "Unexpected number of buffered chunks")
+ })
+ }
+}
+
+// MockChunkedSyncHandler implements queue.ChunkedSyncHandler for testing
+type MockChunkedSyncHandler struct{}
+
+func (m *MockChunkedSyncHandler) CreatePartHandler(ctx *queue.ChunkedSyncPartContext) (queue.PartHandler, error) {
+ return &MockChunkedSyncPartHandler{}, nil
+}
+
+func (m *MockChunkedSyncHandler) HandleFileChunk(ctx *queue.ChunkedSyncPartContext, chunk []byte) error {
+ return nil
+}
+
+// MockChunkedSyncPartHandler implements queue.PartHandler for testing
+type MockChunkedSyncPartHandler struct{}
+
+func (m *MockChunkedSyncPartHandler) FinishSync() error {
+ return nil
+}
+
+func (m *MockChunkedSyncPartHandler) Close() error {
+ return nil
+}
+
+// MockSyncPartStream implements clusterv1.ChunkedSyncService_SyncPartServer for testing
+type MockSyncPartStream struct {
+ sentResponses []*clusterv1.SyncPartResponse
+ ctx context.Context
+}
+
+func (m *MockSyncPartStream) Send(resp *clusterv1.SyncPartResponse) error {
+ m.sentResponses = append(m.sentResponses, resp)
+ return nil
+}
+
+func (m *MockSyncPartStream) Recv() (*clusterv1.SyncPartRequest, error) {
+ // Not needed for these tests, just return nil
+ return nil, nil
+}
+
+func (m *MockSyncPartStream) Context() context.Context {
+ if m.ctx == nil {
+ m.ctx = metadata.NewIncomingContext(context.Background(), metadata.MD{})
+ }
+ return m.ctx
+}
+
+func (m *MockSyncPartStream) SendMsg(msg interface{}) error { return nil }
+func (m *MockSyncPartStream) RecvMsg(msg interface{}) error { return nil }
+func (m *MockSyncPartStream) SetHeader(metadata.MD) error { return nil }
+func (m *MockSyncPartStream) SendHeader(metadata.MD) error { return nil }
+func (m *MockSyncPartStream) SetTrailer(metadata.MD) {}
+
+func TestChunkedSyncBufferTimeout(t *testing.T) {
+ // Initialize logger for tests
+ err := logger.Init(logger.Logging{
+ Env: "dev",
+ Level: "debug",
+ })
+ require.NoError(t, err)
+
+ s := &server{
+ log: logger.GetLogger("test-server-timeout"),
+ chunkedSyncHandlers: make(map[bus.Topic]queue.ChunkedSyncHandler),
+ enableChunkReordering: true,
+ maxChunkBufferSize: 10,
+ maxChunkGapSize: 5,
+ chunkBufferTimeout: 100 * time.Millisecond, // Short timeout for testing
+ }
+
+ session := &syncSession{
+ sessionID: "test-session-timeout",
+ startTime: time.Now(),
+ chunkBuffer: &chunkBuffer{
+ chunks: make(map[uint32]*clusterv1.SyncPartRequest),
+ expectedIndex: 1, // Waiting for chunk 1
+ maxBufferSize: 10,
+ bufferTimeout: 100 * time.Millisecond,
+ lastActivity: time.Now().Add(-200 * time.Millisecond), // Already timed out
+ },
+ }
+
+ // Add a chunk to the buffer to simulate missing chunks
+ session.chunkBuffer.chunks[2] = &clusterv1.SyncPartRequest{ChunkIndex: 2}
+
+ // Check buffer timeout
+ err = s.checkBufferTimeout(session)
+ assert.Error(t, err, "Should return timeout error")
+ assert.Contains(t, err.Error(), "buffer timeout", "Error should mention buffer timeout")
+}
+
+func TestChunkedSyncChecksumValidation(t *testing.T) {
+ // Initialize logger for tests
+ err := logger.Init(logger.Logging{
+ Env: "dev",
+ Level: "info",
+ })
+ require.NoError(t, err)
+
+ // Create a mock server instance to test the processChunk method directly
+ s := &server{
+ log: logger.GetLogger("test-server"),
+ chunkedSyncHandlers: make(map[bus.Topic]queue.ChunkedSyncHandler),
+ }
+
+ // Create a mock stream
+ mockStream := &MockSyncPartStream{}
+
+ t.Run("valid checksum should be accepted", func(t *testing.T) {
+ chunkData := []byte("test chunk data with valid checksum")
+ validChecksum := fmt.Sprintf("%x", crc32.ChecksumIEEE(chunkData))
+
+ session := &syncSession{
+ sessionID: "test-session-1",
+ chunksReceived: 0,
+ partsProgress: make(map[int]*partProgress),
+ metadata: &clusterv1.SyncMetadata{
+ Group: "test-group",
+ ShardId: 1,
+ Topic: data.TopicStreamPartSync.String(),
+ },
+ }
+
+ req := &clusterv1.SyncPartRequest{
+ SessionId: "test-session-1",
+ ChunkIndex: 0,
+ ChunkData: chunkData,
+ ChunkChecksum: validChecksum,
+ PartsInfo: []*clusterv1.PartInfo{
+ {
+ Id: 1,
+ Files: []*clusterv1.FileInfo{
+ {
+ Name: "test-file.dat",
+ Offset: 0,
+ Size: uint32(len(chunkData)),
+ },
+ },
+ CompressedSizeBytes: uint64(len(chunkData)),
+ UncompressedSizeBytes: uint64(len(chunkData)),
+ TotalCount: 1,
+ BlocksCount: 1,
+ MinTimestamp: time.Now().UnixMilli(),
+ MaxTimestamp: time.Now().UnixMilli(),
+ },
+ },
+ }
+
+ // Register a mock handler
+ mockHandler := &MockChunkedSyncHandler{}
+ s.chunkedSyncHandlers[data.TopicStreamPartSync] = mockHandler
+
+ // Test checksum validation - this should succeed
+ err := s.processChunk(mockStream, session, req)
+ require.NoError(t, err)
+
+ // Verify that the correct response was sent
+ assert.Len(t, mockStream.sentResponses, 1)
+ resp := mockStream.sentResponses[0]
+ assert.Equal(t, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, resp.Status)
+ assert.Empty(t, resp.Error)
+ assert.Equal(t, "test-session-1", resp.SessionId)
+ assert.Equal(t, uint32(0), resp.ChunkIndex)
+ })
+
+ t.Run("invalid checksum should be rejected", func(t *testing.T) {
+ chunkData := []byte("test chunk data with invalid checksum")
+ invalidChecksum := "deadbeef" // Intentionally wrong checksum
+
+ session := &syncSession{
+ sessionID: "test-session-2",
+ chunksReceived: 0,
+ partsProgress: make(map[int]*partProgress),
+ metadata: &clusterv1.SyncMetadata{
+ Group: "test-group",
+ ShardId: 1,
+ Topic: data.TopicStreamPartSync.String(),
+ },
+ }
+
+ req := &clusterv1.SyncPartRequest{
+ SessionId: "test-session-2",
+ ChunkIndex: 0,
+ ChunkData: chunkData,
+ ChunkChecksum: invalidChecksum,
+ PartsInfo: []*clusterv1.PartInfo{
+ {
+ Id: 1,
+ Files: []*clusterv1.FileInfo{
+ {
+ Name: "test-file.dat",
+ Offset: 0,
+ Size: uint32(len(chunkData)),
+ },
+ },
+ CompressedSizeBytes: uint64(len(chunkData)),
+ UncompressedSizeBytes: uint64(len(chunkData)),
+ TotalCount: 1,
+ BlocksCount: 1,
+ MinTimestamp: time.Now().UnixMilli(),
+ MaxTimestamp: time.Now().UnixMilli(),
+ },
+ },
+ }
+
+ mockStream := &MockSyncPartStream{}
+
+ // Test checksum validation - this should fail
+ err := s.processChunk(mockStream, session, req)
+ require.NoError(t, err) // processChunk doesn't return error, it sends response
+
+ // Verify that the correct error response was sent
+ assert.Len(t, mockStream.sentResponses, 1)
+ resp := mockStream.sentResponses[0]
+ assert.Equal(t, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_CHECKSUM_MISMATCH, resp.Status)
+ assert.NotEmpty(t, resp.Error)
+ assert.Contains(t, resp.Error, "checksum mismatch")
+ assert.Contains(t, resp.Error, "chunk 0")
+ assert.Equal(t, "test-session-2", resp.SessionId)
+ assert.Equal(t, uint32(0), resp.ChunkIndex)
+
+ // Verify the error message contains both expected and actual checksums
+ expectedChecksum := fmt.Sprintf("%x", crc32.ChecksumIEEE(chunkData))
+ assert.Contains(t, resp.Error, invalidChecksum) // sent checksum
+ assert.Contains(t, resp.Error, expectedChecksum) // calculated checksum
+ })
+
+ t.Run("out of order chunks should be rejected", func(t *testing.T) {
+ chunkData := []byte("out of order chunk")
+ validChecksum := fmt.Sprintf("%x", crc32.ChecksumIEEE(chunkData))
+
+ session := &syncSession{
+ sessionID: "test-session-3",
+ chunksReceived: 0, // Expecting chunk 0, but we'll send chunk 1
+ partsProgress: make(map[int]*partProgress),
+ metadata: &clusterv1.SyncMetadata{
+ Group: "test-group",
+ ShardId: 1,
+ Topic: data.TopicStreamPartSync.String(),
+ },
+ }
+
+ req := &clusterv1.SyncPartRequest{
+ SessionId: "test-session-3",
+ ChunkIndex: 1, // Wrong index - should be 0
+ ChunkData: chunkData,
+ ChunkChecksum: validChecksum,
+ PartsInfo: []*clusterv1.PartInfo{
+ {
+ Id: 1,
+ Files: []*clusterv1.FileInfo{
+ {
+ Name: "test-file.dat",
+ Offset: 0,
+ Size: uint32(len(chunkData)),
+ },
+ },
+ CompressedSizeBytes: uint64(len(chunkData)),
+ UncompressedSizeBytes: uint64(len(chunkData)),
+ TotalCount: 1,
+ BlocksCount: 1,
+ MinTimestamp: time.Now().UnixMilli(),
+ MaxTimestamp: time.Now().UnixMilli(),
+ },
+ },
+ }
+
+ mockStream := &MockSyncPartStream{}
+
+ // Test out of order validation - this should fail
+ err := s.processChunk(mockStream, session, req)
+ require.NoError(t, err) // processChunk doesn't return error, it sends response
+
+ // Verify that the correct error response was sent
+ assert.Len(t, mockStream.sentResponses, 1)
+ resp := mockStream.sentResponses[0]
+ assert.Equal(t, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, resp.Status)
+ assert.NotEmpty(t, resp.Error)
+ assert.Contains(t, resp.Error, "out of order chunk")
+ assert.Contains(t, resp.Error, "expected 0")
+ assert.Contains(t, resp.Error, "got 1")
+ })
+}
+
+func TestChunkedSyncOutOfOrderBasic(t *testing.T) {
+ // Initialize logger for tests
+ err := logger.Init(logger.Logging{
+ Env: "dev",
+ Level: "info",
+ })
+ require.NoError(t, err)
+
+ // Create server with strict sequential mode (reordering disabled)
+ s := &server{
+ log: logger.GetLogger("test-server"),
+ chunkedSyncHandlers: make(map[bus.Topic]queue.ChunkedSyncHandler),
+ enableChunkReordering: false, // Strict sequential mode
+ }
+
+ // Create mock stream
+ mockStream := &MockSyncPartStream{}
+
+ // Create a session
+ session := &syncSession{
+ sessionID: "test-session",
+ startTime: time.Now(),
+ chunksReceived: 0,
+ partsProgress: make(map[int]*partProgress),
+ metadata: &clusterv1.SyncMetadata{
+ Group: "test-group",
+ ShardId: 1,
+ Topic: data.TopicStreamPartSync.String(),
+ },
+ }
+
+ // Register a mock handler
+ mockHandler := &MockChunkedSyncHandler{}
+ s.chunkedSyncHandlers[data.TopicStreamPartSync] = mockHandler
+
+ t.Run("out of order chunk should be rejected in strict mode", func(t *testing.T) {
+ // Send chunk 0 first (should succeed)
+ req := &clusterv1.SyncPartRequest{
+ SessionId: "test-session",
+ ChunkIndex: 0,
+ ChunkData: []byte("chunk-0-data"),
+ ChunkChecksum: fmt.Sprintf("%x", crc32.ChecksumIEEE([]byte("chunk-0-data"))),
+ PartsInfo: []*clusterv1.PartInfo{
+ {
+ Id: 1,
+ Files: []*clusterv1.FileInfo{
+ {
+ Name: "test-file.dat",
+ Offset: 0,
+ Size: 13,
+ },
+ },
+ },
+ },
+ }
+
+ err := s.processChunk(mockStream, session, req)
+ require.NoError(t, err) // processChunk doesn't return error, it sends response
+
+ // Verify first chunk was accepted
+ assert.Len(t, mockStream.sentResponses, 1)
+ resp := mockStream.sentResponses[0]
+ assert.Equal(t, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, resp.Status)
+
+ // Now send chunk 2 (out of order - missing chunk 1)
+ req2 := &clusterv1.SyncPartRequest{
+ SessionId: "test-session",
+ ChunkIndex: 2,
+ ChunkData: []byte("chunk-2-data"),
+ ChunkChecksum: fmt.Sprintf("%x", crc32.ChecksumIEEE([]byte("chunk-2-data"))),
+ PartsInfo: []*clusterv1.PartInfo{
+ {
+ Id: 1,
+ Files: []*clusterv1.FileInfo{
+ {
+ Name: "test-file.dat",
+ Offset: 0,
+ Size: 13,
+ },
+ },
+ },
+ },
+ }
+
+ err = s.processChunk(mockStream, session, req2)
+ require.NoError(t, err) // processChunk doesn't return error, it sends response
+
+ // Verify that the correct error response was sent
+ assert.Len(t, mockStream.sentResponses, 2)
+ resp = mockStream.sentResponses[1]
+ assert.Equal(t, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, resp.Status)
+ assert.NotEmpty(t, resp.Error)
+ assert.Contains(t, resp.Error, "out of order chunk")
+ assert.Contains(t, resp.Error, "expected 1")
+ assert.Contains(t, resp.Error, "got 2")
+ })
+}
\ No newline at end of file
diff --git a/banyand/queue/sub/server.go b/banyand/queue/sub/server.go
index 6e246531..d5349b7e 100644
--- a/banyand/queue/sub/server.go
+++ b/banyand/queue/sub/server.go
@@ -67,28 +67,35 @@ var (
type server struct {
clusterv1.UnimplementedServiceServer
+ clusterv1.UnimplementedChunkedSyncServiceServer
streamv1.UnimplementedStreamServiceServer
databasev1.UnimplementedSnapshotServiceServer
- creds credentials.TransportCredentials
- omr observability.MetricsRegistry
- metrics *metrics
- ser *grpclib.Server
- listeners map[bus.Topic][]bus.MessageListener
- topicMap map[string]bus.Topic
- log *logger.Logger
- httpSrv *http.Server
- clientCloser context.CancelFunc
- httpAddr string
- addr string
- host string
- certFile string
- keyFile string
- flagNamePrefix string
- maxRecvMsgSize run.Bytes
- listenersLock sync.RWMutex
- port uint32
- httpPort uint32
- tls bool
+ creds credentials.TransportCredentials
+ omr observability.MetricsRegistry
+ metrics *metrics
+ ser *grpclib.Server
+ listeners map[bus.Topic][]bus.MessageListener
+ topicMap map[string]bus.Topic
+ chunkedSyncHandlers map[bus.Topic]queue.ChunkedSyncHandler
+ log *logger.Logger
+ httpSrv *http.Server
+ clientCloser context.CancelFunc
+ httpAddr string
+ addr string
+ host string
+ certFile string
+ keyFile string
+ flagNamePrefix string
+ maxRecvMsgSize run.Bytes
+ listenersLock sync.RWMutex
+ port uint32
+ httpPort uint32
+ tls bool
+ // Chunk ordering configuration
+ enableChunkReordering bool
+ maxChunkBufferSize uint32
+ chunkBufferTimeout time.Duration
+ maxChunkGapSize uint32
}
// NewServer returns a new gRPC server.
@@ -99,13 +106,19 @@ func NewServer(omr observability.MetricsRegistry) queue.Server {
// NewServerWithPorts returns a new gRPC server with specified ports.
func NewServerWithPorts(omr observability.MetricsRegistry, flagNamePrefix string, port, httpPort uint32) queue.Server {
return &server{
- listeners: make(map[bus.Topic][]bus.MessageListener),
- topicMap: make(map[string]bus.Topic),
- omr: omr,
- maxRecvMsgSize: defaultRecvSize,
- flagNamePrefix: flagNamePrefix,
- port: port,
- httpPort: httpPort,
+ listeners: make(map[bus.Topic][]bus.MessageListener),
+ topicMap: make(map[string]bus.Topic),
+ chunkedSyncHandlers: make(map[bus.Topic]queue.ChunkedSyncHandler),
+ omr: omr,
+ maxRecvMsgSize: defaultRecvSize,
+ flagNamePrefix: flagNamePrefix,
+ port: port,
+ httpPort: httpPort,
+ // Default chunk ordering configuration
+ enableChunkReordering: true,
+ maxChunkBufferSize: 10,
+ chunkBufferTimeout: 5 * time.Second,
+ maxChunkGapSize: 5,
}
}
@@ -142,6 +155,11 @@ func (s *server) FlagSet() *run.FlagSet {
fs.StringVar(&s.host, prefixFlag("grpc-host"), "", "the host of banyand listens")
fs.Uint32Var(&s.port, prefixFlag("grpc-port"), s.port, "the port of banyand listens")
fs.Uint32Var(&s.httpPort, prefixFlag("http-port"), s.httpPort, "the port of banyand http api listens")
+ // Chunk ordering configuration flags
+ fs.BoolVar(&s.enableChunkReordering, prefixFlag("enable-chunk-reordering"), s.enableChunkReordering, "enable out-of-order chunk handling")
+ fs.Uint32Var(&s.maxChunkBufferSize, prefixFlag("max-chunk-buffer-size"), s.maxChunkBufferSize, "maximum chunks to buffer for reordering")
+ fs.DurationVar(&s.chunkBufferTimeout, prefixFlag("chunk-buffer-timeout"), s.chunkBufferTimeout, "maximum time to wait for missing chunks")
+ fs.Uint32Var(&s.maxChunkGapSize, prefixFlag("max-chunk-gap-size"), s.maxChunkGapSize, "maximum gap in chunk sequence")
return fs
}
@@ -202,6 +220,7 @@ func (s *server) Serve() run.StopNotify {
)
s.ser = grpclib.NewServer(opts...)
clusterv1.RegisterServiceServer(s.ser, s)
+ clusterv1.RegisterChunkedSyncServiceServer(s.ser, s)
grpc_health_v1.RegisterHealthServer(s.ser, health.NewServer())
databasev1.RegisterSnapshotServiceServer(s.ser, s)
streamv1.RegisterStreamServiceServer(s.ser, &streamService{ser: s})
@@ -290,6 +309,11 @@ func (s *server) GracefulStop() {
}
}
+// RegisterChunkedSyncHandler implements queue.Server.
+func (s *server) RegisterChunkedSyncHandler(topic bus.Topic, handler queue.ChunkedSyncHandler) {
+ s.chunkedSyncHandlers[topic] = handler
+}
+
type metrics struct {
totalStarted meter.Counter
totalFinished meter.Counter
@@ -300,6 +324,13 @@ type metrics struct {
totalMsgReceivedErr meter.Counter
totalMsgSent meter.Counter
totalMsgSentErr meter.Counter
+
+ // Chunk ordering metrics
+ outOfOrderChunksReceived meter.Counter
+ chunksBuffered meter.Counter
+ bufferTimeouts meter.Counter
+ largeGapsRejected meter.Counter
+ bufferCapacityExceeded meter.Counter
}
func newMetrics(factory *observability.Factory) *metrics {
@@ -312,5 +343,31 @@ func newMetrics(factory *observability.Factory) *metrics {
totalMsgReceivedErr: factory.NewCounter("total_msg_received_err", "topic"),
totalMsgSent: factory.NewCounter("total_msg_sent", "topic"),
totalMsgSentErr: factory.NewCounter("total_msg_sent_err", "topic"),
+
+ // Chunk ordering metrics
+ outOfOrderChunksReceived: factory.NewCounter("out_of_order_chunks_received", "session_id"),
+ chunksBuffered: factory.NewCounter("chunks_buffered", "session_id"),
+ bufferTimeouts: factory.NewCounter("buffer_timeouts", "session_id"),
+ largeGapsRejected: factory.NewCounter("large_gaps_rejected", "session_id"),
+ bufferCapacityExceeded: factory.NewCounter("buffer_capacity_exceeded", "session_id"),
+ }
+}
+
+// updateChunkOrderMetrics updates chunk ordering metrics.
+func (s *server) updateChunkOrderMetrics(event string, sessionID string) {
+ if s.metrics == nil {
+ return // Skip metrics if not initialized (e.g., during tests)
+ }
+ switch event {
+ case "out_of_order_received":
+ s.metrics.outOfOrderChunksReceived.Inc(1, sessionID)
+ case "chunk_buffered":
+ s.metrics.chunksBuffered.Inc(1, sessionID)
+ case "buffer_timeout":
+ s.metrics.bufferTimeouts.Inc(1, sessionID)
+ case "gap_too_large":
+ s.metrics.largeGapsRejected.Inc(1, sessionID)
+ case "buffer_full":
+ s.metrics.bufferCapacityExceeded.Inc(1, sessionID)
}
}
diff --git a/banyand/queue/test/chunked_sync_basic_test.go b/banyand/queue/test/chunked_sync_basic_test.go
new file mode 100644
index 00000000..ac504ea9
--- /dev/null
+++ b/banyand/queue/test/chunked_sync_basic_test.go
@@ -0,0 +1,115 @@
+// Package test provides testing utilities and test cases for the queue package.
+package test
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+)
+
+// TestChunkedSyncBasic validates the chunked streaming data synchronization.
+func TestChunkedSyncBasic(t *testing.T) {
+ setup := setupChunkedSyncTest(t, "chunked-sync-test")
+ defer cleanupTestSetup(setup)
+
+ testFileName := "test-file.txt"
+ testContent := []byte("Hello, BanyanDB chunked sync test!")
+ testFiles := map[string][]byte{testFileName: testContent}
+ dataPart := createTestDataPart(testFiles, "test-group", 1)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, []queue.StreamingPartData{dataPart})
+ require.NoError(t, err, "Failed to send data part")
+ require.NotNil(t, result, "Sync result should not be nil")
+
+ time.Sleep(500 * time.Millisecond)
+
+ assert.True(t, result.Success, "Sync should be successful")
+ assert.Equal(t, uint32(1), result.PartsCount, "Should have synced 1 part")
+ assert.Greater(t, result.ChunksCount, uint32(0), "Should have sent at least 1 chunk")
+ assert.Greater(t, result.TotalBytes, uint64(0), "Should have sent some bytes")
+ assert.NotEmpty(t, result.SessionID, "Session ID should not be empty")
+
+ receivedFiles := setup.MockHandler.GetReceivedFiles()
+ require.Len(t, receivedFiles, 1, "Expected exactly one received file")
+
+ receivedContent, exists := receivedFiles[testFileName]
+ require.True(t, exists, "Expected file %s to be received", testFileName)
+ assert.Equal(t, testContent, receivedContent, "File content should match")
+
+ chunksCount := setup.MockHandler.GetReceivedChunksCount()
+ assert.Greater(t, chunksCount, 0, "Should have received at least one chunk")
+
+ completedParts := setup.MockHandler.GetCompletedParts()
+ assert.Len(t, completedParts, 1, "Should have completed one part")
+ if len(completedParts) > 0 {
+ assert.True(t, completedParts[0].finishSyncCalled, "FinishSync should have been called")
+ }
+
+ t.Logf("Test completed successfully:")
+ t.Logf(" - Session ID: %s", result.SessionID)
+ t.Logf(" - Total bytes sent: %d", result.TotalBytes)
+ t.Logf(" - Chunks sent in total: %d", result.ChunksCount)
+ t.Logf(" - Duration: %d ms", result.DurationMs)
+ t.Logf(" - Chunks received in file-based sync: %d", chunksCount)
+ t.Logf(" - Files received: %d", len(receivedFiles))
+}
+
+// TestChunkedSyncMultipleFiles tests syncing multiple files in a single part.
+func TestChunkedSyncMultipleFiles(t *testing.T) {
+ setup := setupChunkedSyncTest(t, "chunked-sync-multi-test")
+ defer cleanupTestSetup(setup)
+
+ testFiles := map[string][]byte{
+ "primary.data": []byte("Primary data content for testing"),
+ "timestamps.data": []byte("Timestamp data: 1234567890"),
+ "metadata.json": []byte(`{"version": 1, "type": "test"}`),
+ }
+
+ dataPart := createTestDataPart(testFiles, "test-group-multi", 2)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, []queue.StreamingPartData{dataPart})
+ require.NoError(t, err, "Failed to send data part")
+ require.True(t, result.Success, "Sync should be successful")
+
+ time.Sleep(500 * time.Millisecond)
+
+ receivedFiles := setup.MockHandler.GetReceivedFiles()
+ require.Len(t, receivedFiles, len(testFiles), "Should receive all files")
+
+ for fileName, expectedContent := range testFiles {
+ receivedContent, exists := receivedFiles[fileName]
+ require.True(t, exists, "File %s should be received", fileName)
+ assert.Equal(t, expectedContent, receivedContent, "Content for file %s should match", fileName)
+ }
+
+ t.Logf("Multi-file test completed: %d files synced successfully", len(receivedFiles))
+}
+
+// TestChunkedSyncEmptyData tests syncing with empty data.
+func TestChunkedSyncEmptyData(t *testing.T) {
+ setup := setupChunkedSyncTest(t, "chunked-sync-empty-test")
+ defer cleanupTestSetup(setup)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, []queue.StreamingPartData{})
+ require.NoError(t, err, "Empty sync should not fail")
+ require.NotNil(t, result, "Result should not be nil")
+
+ assert.True(t, result.Success, "Empty sync should be successful")
+ assert.Equal(t, uint32(0), result.PartsCount, "Parts count should be 0")
+ assert.Equal(t, uint32(0), result.ChunksCount, "Chunks count should be 0")
+ assert.Empty(t, result.SessionID, "Session ID should be empty for no data")
+}
diff --git a/banyand/queue/test/chunked_sync_common.go b/banyand/queue/test/chunked_sync_common.go
new file mode 100644
index 00000000..9e2069b1
--- /dev/null
+++ b/banyand/queue/test/chunked_sync_common.go
@@ -0,0 +1,387 @@
+// 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 test
+
+import (
+ "context"
+ "fmt"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/spf13/cobra"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/banyand/queue/pub"
+ "github.com/apache/skywalking-banyandb/banyand/queue/sub"
+ "github.com/apache/skywalking-banyandb/pkg/bytes"
+ "github.com/apache/skywalking-banyandb/pkg/run"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+ "github.com/apache/skywalking-banyandb/pkg/test/helpers"
+)
+
+// Setup holds the test environment configuration and components.
+type Setup struct {
+ Server queue.Server
+ Client queue.Client
+ Closer run.Unit
+ ChunkedClient queue.ChunkedSyncClient
+ MockHandler *MockChunkedSyncHandler
+ DeferFn func()
+ NodeName string
+ NodeAddr string
+ TestGroup run.Group
+ GRPCPort uint32
+ HTTPPort uint32
+}
+
+// setupChunkedSyncTest creates a complete test environment for chunked sync testing.
+func setupChunkedSyncTest(t *testing.T, testName string) *Setup {
+ return setupChunkedSyncTestWithChunkSize(t, testName, 1024)
+}
+
+// setupChunkedSyncTestWithChunkSize creates a complete test environment for chunked sync testing with custom chunk size.
+func setupChunkedSyncTestWithChunkSize(t *testing.T, testName string, chunkSize uint32) *Setup {
+ ports, err := test.AllocateFreePorts(2)
+ require.NoError(t, err, "Failed to allocate free ports")
+
+ grpcPort := uint32(ports[0])
+ httpPort := uint32(ports[1])
+
+ omr := observability.BypassRegistry
+
+ testGroup := run.NewGroup(testName)
+ closer, deferFn := run.NewTester(testName + "-closer")
+
+ server := sub.NewServerWithPorts(omr, testName, grpcPort, httpPort)
+
+ mockHandler := NewMockChunkedSyncHandler()
+
+ server.RegisterChunkedSyncHandler(data.TopicStreamPartSync, mockHandler)
+
+ testGroup.Register(closer, server)
+
+ cmd := &cobra.Command{
+ Use: testName,
+ FParseErrWhitelist: cobra.FParseErrWhitelist{
+ UnknownFlags: true, // Ignore unknown flags
+ },
+ Run: func(_ *cobra.Command, _ []string) {
+ err = testGroup.Run(context.Background())
+ if err != nil {
+ t.Logf("Server group failed: %v", err)
+ }
+ },
+ }
+ cmd.Flags().AddFlagSet(testGroup.RegisterFlags().FlagSet)
+ go func() {
+ require.NoError(t, cmd.Execute())
+ }()
+
+ assert.Eventually(t, func() bool {
+ errInternal := helpers.HealthCheck(fmt.Sprintf("localhost:%d", grpcPort), 10*time.Second, 10*time.Second, grpc.WithTransportCredentials(insecure.NewCredentials()))()
+ return errInternal == nil
+ }, flags.EventuallyTimeout, 100*time.Millisecond)
+
+ client := pub.NewWithoutMetadata()
+
+ nodeAddr := fmt.Sprintf("localhost:%d", grpcPort)
+ nodeName := testName + "-node"
+ node := schema.Metadata{
+ TypeMeta: schema.TypeMeta{
+ Name: nodeName,
+ Kind: schema.KindNode,
+ },
+ Spec: &databasev1.Node{
+ Metadata: &commonv1.Metadata{
+ Name: nodeName,
+ },
+ Roles: []databasev1.Role{databasev1.Role_ROLE_DATA},
+ GrpcAddress: nodeAddr,
+ },
+ }
+
+ client.OnAddOrUpdate(node)
+
+ var chunkedClient queue.ChunkedSyncClient
+ assert.Eventually(t, func() bool {
+ var errInternal error
+ chunkedClient, errInternal = client.NewChunkedSyncClient(nodeName, chunkSize)
+ if err != nil {
+ t.Logf("NewChunkedSyncClient error: %v", err)
+ }
+ return errInternal == nil
+ }, flags.EventuallyTimeout, 100*time.Millisecond)
+
+ return &Setup{
+ Server: server,
+ Client: client,
+ MockHandler: mockHandler,
+ TestGroup: testGroup,
+ Closer: closer,
+ DeferFn: deferFn,
+ GRPCPort: grpcPort,
+ HTTPPort: httpPort,
+ NodeName: nodeName,
+ NodeAddr: nodeAddr,
+ ChunkedClient: chunkedClient,
+ }
+}
+
+// cleanupTestSetup cleans up the test environment.
+func cleanupTestSetup(setup *Setup) {
+ if setup.ChunkedClient != nil {
+ setup.ChunkedClient.Close()
+ }
+ if setup.DeferFn != nil {
+ setup.DeferFn()
+ }
+}
+
+// createTestDataPart creates a StreamingPartData for testing.
+func createTestDataPart(files map[string][]byte, group string, shardID uint32) queue.StreamingPartData {
+ var fileInfos []queue.FileInfo
+ var totalSize uint64
+
+ for fileName, content := range files {
+ var buf bytes.Buffer
+ if _, err := buf.Write(content); err != nil {
+ panic(fmt.Sprintf("failed to write content to buffer: %v", err))
+ }
+ fileInfos = append(fileInfos, queue.FileInfo{
+ Name: fileName,
+ Reader: buf.SequentialRead(),
+ })
+ totalSize += uint64(len(content))
+ }
+
+ return queue.StreamingPartData{
+ ID: 1,
+ Files: fileInfos,
+ Group: group,
+ ShardID: shardID,
+ Topic: data.TopicStreamPartSync.String(),
+ CompressedSizeBytes: totalSize,
+ UncompressedSizeBytes: totalSize,
+ TotalCount: uint64(len(files)),
+ BlocksCount: 1,
+ MinTimestamp: time.Now().UnixMilli(),
+ MaxTimestamp: time.Now().UnixMilli(),
+ }
+}
+
+// createTestDataPartWithMetadata creates a StreamingPartData for testing with specific metadata values.
+func createTestDataPartWithMetadata(files map[string][]byte, group string, shardID uint32,
+ compressedSize, uncompressedSize, totalCount, blocksCount uint64,
+ minTimestamp, maxTimestamp int64, partID uint64,
+) queue.StreamingPartData {
+ var fileInfos []queue.FileInfo
+
+ for fileName, content := range files {
+ var buf bytes.Buffer
+ if _, err := buf.Write(content); err != nil {
+ panic(fmt.Sprintf("failed to write content to buffer: %v", err))
+ }
+ fileInfos = append(fileInfos, queue.FileInfo{
+ Name: fileName,
+ Reader: buf.SequentialRead(),
+ })
+ }
+
+ return queue.StreamingPartData{
+ ID: partID,
+ Files: fileInfos,
+ Group: group,
+ ShardID: shardID,
+ Topic: data.TopicStreamPartSync.String(),
+ CompressedSizeBytes: compressedSize,
+ UncompressedSizeBytes: uncompressedSize,
+ TotalCount: totalCount,
+ BlocksCount: blocksCount,
+ MinTimestamp: minTimestamp,
+ MaxTimestamp: maxTimestamp,
+ }
+}
+
+// MockPartHandler implements queue.PartHandler for testing.
+type MockPartHandler struct {
+ receivedFiles map[string][]byte
+ syncHandler *MockChunkedSyncHandler
+ receivedChunks [][]byte
+ finishSyncCalled bool
+ closeCalled bool
+}
+
+// FinishSync marks the sync as finished and closes the handler.
+func (m *MockPartHandler) FinishSync() error {
+ m.finishSyncCalled = true
+
+ if m.syncHandler != nil {
+ m.syncHandler.mu.Lock()
+ m.syncHandler.completedParts = append(m.syncHandler.completedParts, m)
+ m.syncHandler.mu.Unlock()
+ }
+ return m.Close()
+}
+
+// Close marks the handler as closed.
+func (m *MockPartHandler) Close() error {
+ m.closeCalled = true
+ return nil
+}
+
+// GetReceivedFiles returns a copy of the received files.
+func (m *MockPartHandler) GetReceivedFiles() map[string][]byte {
+ result := make(map[string][]byte)
+ for k, v := range m.receivedFiles {
+ result[k] = make([]byte, len(v))
+ copy(result[k], v)
+ }
+ return result
+}
+
+// GetReceivedChunksCount returns the number of received chunks.
+func (m *MockPartHandler) GetReceivedChunksCount() int {
+ return len(m.receivedChunks)
+}
+
+// MockChunkedSyncHandler implements queue.ChunkedSyncHandler for testing.
+type MockChunkedSyncHandler struct {
+ completedParts []*MockPartHandler
+ receivedContexts []*queue.ChunkedSyncPartContext
+ mu sync.Mutex
+}
+
+// NewMockChunkedSyncHandler creates a new mock chunked sync handler.
+func NewMockChunkedSyncHandler() *MockChunkedSyncHandler {
+ return &MockChunkedSyncHandler{
+ completedParts: make([]*MockPartHandler, 0),
+ receivedContexts: make([]*queue.ChunkedSyncPartContext, 0),
+ }
+}
+
+// HandleFileChunk processes incoming file chunks for testing.
+func (m *MockChunkedSyncHandler) HandleFileChunk(ctx *queue.ChunkedSyncPartContext, chunk []byte) error {
+ currentPart := ctx.Handler.(*MockPartHandler)
+
+ if currentPart != nil {
+ chunkCopy := make([]byte, len(chunk))
+ copy(chunkCopy, chunk)
+ currentPart.receivedChunks = append(currentPart.receivedChunks, chunkCopy)
+
+ if currentPart.receivedFiles[ctx.FileName] == nil {
+ currentPart.receivedFiles[ctx.FileName] = make([]byte, 0)
+ }
+ currentPart.receivedFiles[ctx.FileName] = append(currentPart.receivedFiles[ctx.FileName], chunkCopy...)
+ }
+
+ return nil
+}
+
+// CreatePartHandler creates a new part handler for testing.
+func (m *MockChunkedSyncHandler) CreatePartHandler(ctx *queue.ChunkedSyncPartContext) (queue.PartHandler, error) {
+ m.mu.Lock()
+ contextCopy := &queue.ChunkedSyncPartContext{
+ ID: ctx.ID,
+ Group: ctx.Group,
+ ShardID: ctx.ShardID,
+ CompressedSizeBytes: ctx.CompressedSizeBytes,
+ UncompressedSizeBytes: ctx.UncompressedSizeBytes,
+ TotalCount: ctx.TotalCount,
+ BlocksCount: ctx.BlocksCount,
+ MinTimestamp: ctx.MinTimestamp,
+ MaxTimestamp: ctx.MaxTimestamp,
+ }
+ m.receivedContexts = append(m.receivedContexts, contextCopy)
+ m.mu.Unlock()
+
+ partHandler := &MockPartHandler{
+ receivedFiles: make(map[string][]byte),
+ syncHandler: m,
+ receivedChunks: make([][]byte, 0),
+ }
+ contextCopy.Handler = partHandler
+ return partHandler, nil
+}
+
+// GetCompletedParts returns all completed parts.
+func (m *MockChunkedSyncHandler) GetCompletedParts() []*MockPartHandler {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ result := make([]*MockPartHandler, len(m.completedParts))
+ copy(result, m.completedParts)
+ return result
+}
+
+// GetReceivedFiles returns all received files across all parts.
+func (m *MockChunkedSyncHandler) GetReceivedFiles() map[string][]byte {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ result := make(map[string][]byte)
+ for _, part := range m.completedParts {
+ for fileName, content := range part.GetReceivedFiles() {
+ result[fileName] = content
+ }
+ }
+ return result
+}
+
+// GetReceivedChunksCount returns the total number of received chunks.
+func (m *MockChunkedSyncHandler) GetReceivedChunksCount() int {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ total := 0
+ for _, part := range m.completedParts {
+ total += part.GetReceivedChunksCount()
+ }
+ return total
+}
+
+// GetReceivedContexts returns all received contexts.
+func (m *MockChunkedSyncHandler) GetReceivedContexts() []*queue.ChunkedSyncPartContext {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ result := make([]*queue.ChunkedSyncPartContext, len(m.receivedContexts))
+ for i, ctx := range m.receivedContexts {
+ contextCopy := &queue.ChunkedSyncPartContext{
+ ID: ctx.ID,
+ Group: ctx.Group,
+ ShardID: ctx.ShardID,
+ FileName: ctx.FileName,
+ CompressedSizeBytes: ctx.CompressedSizeBytes,
+ UncompressedSizeBytes: ctx.UncompressedSizeBytes,
+ TotalCount: ctx.TotalCount,
+ BlocksCount: ctx.BlocksCount,
+ MinTimestamp: ctx.MinTimestamp,
+ MaxTimestamp: ctx.MaxTimestamp,
+ }
+ result[i] = contextCopy
+ }
+ return result
+}
diff --git a/banyand/queue/test/chunked_sync_metadata_test.go b/banyand/queue/test/chunked_sync_metadata_test.go
new file mode 100644
index 00000000..5b4fc29a
--- /dev/null
+++ b/banyand/queue/test/chunked_sync_metadata_test.go
@@ -0,0 +1,280 @@
+// 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 test
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+)
+
+// TestChunkedSyncMetadataVerification verifies that all metadata fields are properly transmitted from client to server.
+func TestChunkedSyncMetadataVerification(t *testing.T) {
+ setup := setupChunkedSyncTest(t, "chunked-sync-metadata-test")
+ defer cleanupTestSetup(setup)
+
+ testFiles := map[string][]byte{
+ "test-metadata.json": []byte(`{"test": "metadata verification"}`),
+ "test-data.bin": []byte("test binary data for metadata verification"),
+ }
+
+ expectedMetadata := struct {
+ Group string
+ ID uint64
+ CompressedSizeBytes uint64
+ UncompressedSizeBytes uint64
+ TotalCount uint64
+ BlocksCount uint64
+ MinTimestamp int64
+ MaxTimestamp int64
+ ShardID uint32
+ }{
+ ID: 12345,
+ Group: "test-metadata-group",
+ ShardID: 42,
+ CompressedSizeBytes: 1024,
+ UncompressedSizeBytes: 2048,
+ TotalCount: 100,
+ BlocksCount: 5,
+ MinTimestamp: 1640995200000,
+ MaxTimestamp: 1641081600000,
+ }
+
+ dataPart := createTestDataPartWithMetadata(
+ testFiles,
+ expectedMetadata.Group,
+ expectedMetadata.ShardID,
+ expectedMetadata.CompressedSizeBytes,
+ expectedMetadata.UncompressedSizeBytes,
+ expectedMetadata.TotalCount,
+ expectedMetadata.BlocksCount,
+ expectedMetadata.MinTimestamp,
+ expectedMetadata.MaxTimestamp,
+ expectedMetadata.ID,
+ )
+
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, []queue.StreamingPartData{dataPart})
+ require.NoError(t, err, "Failed to send data part with metadata")
+ require.NotNil(t, result, "Sync result should not be nil")
+
+ time.Sleep(500 * time.Millisecond)
+
+ assert.True(t, result.Success, "Sync should be successful")
+ assert.Equal(t, uint32(1), result.PartsCount, "Should have synced 1 part")
+ assert.Greater(t, result.ChunksCount, uint32(0), "Should have sent at least 1 chunk")
+ assert.Greater(t, result.TotalBytes, uint64(0), "Should have sent some bytes")
+
+ receivedContexts := setup.MockHandler.GetReceivedContexts()
+ require.Len(t, receivedContexts, 1, "Should have received exactly one context")
+
+ receivedContext := receivedContexts[0]
+
+ assert.Equal(t, expectedMetadata.ID, receivedContext.ID, "Part ID should match")
+ assert.Equal(t, expectedMetadata.Group, receivedContext.Group, "Group should match")
+ assert.Equal(t, expectedMetadata.ShardID, receivedContext.ShardID, "Shard ID should match")
+ assert.Equal(t, expectedMetadata.CompressedSizeBytes, receivedContext.CompressedSizeBytes, "Compressed size should match")
+ assert.Equal(t, expectedMetadata.UncompressedSizeBytes, receivedContext.UncompressedSizeBytes, "Uncompressed size should match")
+ assert.Equal(t, expectedMetadata.TotalCount, receivedContext.TotalCount, "Total count should match")
+ assert.Equal(t, expectedMetadata.BlocksCount, receivedContext.BlocksCount, "Blocks count should match")
+ assert.Equal(t, expectedMetadata.MinTimestamp, receivedContext.MinTimestamp, "Min timestamp should match")
+ assert.Equal(t, expectedMetadata.MaxTimestamp, receivedContext.MaxTimestamp, "Max timestamp should match")
+
+ receivedFiles := setup.MockHandler.GetReceivedFiles()
+ require.Len(t, receivedFiles, len(testFiles), "Should receive all test files")
+
+ for fileName, expectedContent := range testFiles {
+ receivedContent, exists := receivedFiles[fileName]
+ require.True(t, exists, "File %s should be received", fileName)
+ assert.Equal(t, expectedContent, receivedContent, "File content for %s should match", fileName)
+ }
+
+ completedParts := setup.MockHandler.GetCompletedParts()
+ assert.Len(t, completedParts, 1, "Should have completed one part")
+ if len(completedParts) > 0 {
+ assert.True(t, completedParts[0].finishSyncCalled, "FinishSync should have been called")
+ assert.True(t, completedParts[0].closeCalled, "Close should have been called")
+ }
+
+ t.Logf("Metadata verification test completed successfully:")
+ t.Logf(" - Session ID: %s", result.SessionID)
+ t.Logf(" - Part ID: %d", receivedContext.ID)
+ t.Logf(" - Group: %s", receivedContext.Group)
+ t.Logf(" - Shard ID: %d", receivedContext.ShardID)
+ t.Logf(" - Compressed Size: %d bytes", receivedContext.CompressedSizeBytes)
+ t.Logf(" - Uncompressed Size: %d bytes", receivedContext.UncompressedSizeBytes)
+ t.Logf(" - Total Count: %d", receivedContext.TotalCount)
+ t.Logf(" - Blocks Count: %d", receivedContext.BlocksCount)
+ t.Logf(" - Min Timestamp: %d", receivedContext.MinTimestamp)
+ t.Logf(" - Max Timestamp: %d", receivedContext.MaxTimestamp)
+ t.Logf(" - Files received: %d", len(receivedFiles))
+ t.Logf(" - Total bytes sent: %d", result.TotalBytes)
+}
+
+// TestChunkedSyncMultiplePartsMetadataVerification verifies metadata transmission for multiple parts.
+func TestChunkedSyncMultiplePartsMetadataVerification(t *testing.T) {
+ setup := setupChunkedSyncTest(t, "chunked-sync-multi-metadata-test")
+ defer cleanupTestSetup(setup)
+
+ part1Files := map[string][]byte{"part1.txt": []byte("part 1 data")}
+ part2Files := map[string][]byte{"part2.txt": []byte("part 2 data")}
+ part3Files := map[string][]byte{"part3.txt": []byte("part 3 data")}
+
+ partsMetadata := []struct {
+ Group string
+ ID uint64
+ CompressedSizeBytes uint64
+ UncompressedSizeBytes uint64
+ TotalCount uint64
+ BlocksCount uint64
+ MinTimestamp int64
+ MaxTimestamp int64
+ ShardID uint32
+ }{
+ {
+ ID: 1001,
+ Group: "test-group-multi",
+ ShardID: 42,
+ CompressedSizeBytes: 512,
+ UncompressedSizeBytes: 1024,
+ TotalCount: 50,
+ BlocksCount: 2,
+ MinTimestamp: 1640995200000,
+ MaxTimestamp: 1640998800000,
+ },
+ {
+ ID: 1002,
+ Group: "test-group-multi",
+ ShardID: 42,
+ CompressedSizeBytes: 768,
+ UncompressedSizeBytes: 1536,
+ TotalCount: 75,
+ BlocksCount: 3,
+ MinTimestamp: 1640998800000,
+ MaxTimestamp: 1641002400000,
+ },
+ {
+ ID: 1003,
+ Group: "test-group-multi",
+ ShardID: 42,
+ CompressedSizeBytes: 1024,
+ UncompressedSizeBytes: 2048,
+ TotalCount: 100,
+ BlocksCount: 4,
+ MinTimestamp: 1641002400000,
+ MaxTimestamp: 1641006000000,
+ },
+ }
+
+ dataParts := []queue.StreamingPartData{
+ createTestDataPartWithMetadata(
+ part1Files,
+ partsMetadata[0].Group,
+ partsMetadata[0].ShardID,
+ partsMetadata[0].CompressedSizeBytes,
+ partsMetadata[0].UncompressedSizeBytes,
+ partsMetadata[0].TotalCount,
+ partsMetadata[0].BlocksCount,
+ partsMetadata[0].MinTimestamp,
+ partsMetadata[0].MaxTimestamp,
+ partsMetadata[0].ID,
+ ),
+ createTestDataPartWithMetadata(
+ part2Files,
+ partsMetadata[1].Group,
+ partsMetadata[1].ShardID,
+ partsMetadata[1].CompressedSizeBytes,
+ partsMetadata[1].UncompressedSizeBytes,
+ partsMetadata[1].TotalCount,
+ partsMetadata[1].BlocksCount,
+ partsMetadata[1].MinTimestamp,
+ partsMetadata[1].MaxTimestamp,
+ partsMetadata[1].ID,
+ ),
+ createTestDataPartWithMetadata(
+ part3Files,
+ partsMetadata[2].Group,
+ partsMetadata[2].ShardID,
+ partsMetadata[2].CompressedSizeBytes,
+ partsMetadata[2].UncompressedSizeBytes,
+ partsMetadata[2].TotalCount,
+ partsMetadata[2].BlocksCount,
+ partsMetadata[2].MinTimestamp,
+ partsMetadata[2].MaxTimestamp,
+ partsMetadata[2].ID,
+ ),
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, dataParts)
+ require.NoError(t, err, "Failed to send multiple parts with metadata")
+ require.True(t, result.Success, "Multiple parts sync should be successful")
+
+ time.Sleep(500 * time.Millisecond)
+
+ assert.Equal(t, uint32(3), result.PartsCount, "Should have synced 3 parts")
+ assert.Greater(t, result.ChunksCount, uint32(0), "Should have sent at least 1 chunk")
+
+ receivedContexts := setup.MockHandler.GetReceivedContexts()
+ require.Len(t, receivedContexts, 3, "Should have received exactly 3 contexts")
+
+ for i, expectedMetadata := range partsMetadata {
+ receivedContext := receivedContexts[i]
+
+ assert.Equal(t, expectedMetadata.ID, receivedContext.ID, "Part %d ID should match", i+1)
+ assert.Equal(t, expectedMetadata.Group, receivedContext.Group, "Part %d Group should match", i+1)
+ assert.Equal(t, expectedMetadata.ShardID, receivedContext.ShardID, "Part %d Shard ID should match", i+1)
+ assert.Equal(t, expectedMetadata.CompressedSizeBytes, receivedContext.CompressedSizeBytes, "Part %d Compressed size should match", i+1)
+ assert.Equal(t, expectedMetadata.UncompressedSizeBytes, receivedContext.UncompressedSizeBytes, "Part %d Uncompressed size should match", i+1)
+ assert.Equal(t, expectedMetadata.TotalCount, receivedContext.TotalCount, "Part %d Total count should match", i+1)
+ assert.Equal(t, expectedMetadata.BlocksCount, receivedContext.BlocksCount, "Part %d Blocks count should match", i+1)
+ assert.Equal(t, expectedMetadata.MinTimestamp, receivedContext.MinTimestamp, "Part %d Min timestamp should match", i+1)
+ assert.Equal(t, expectedMetadata.MaxTimestamp, receivedContext.MaxTimestamp, "Part %d Max timestamp should match", i+1)
+ }
+
+ receivedFiles := setup.MockHandler.GetReceivedFiles()
+ expectedFileCount := len(part1Files) + len(part2Files) + len(part3Files)
+ require.Len(t, receivedFiles, expectedFileCount, "Should receive all files from all parts")
+
+ completedParts := setup.MockHandler.GetCompletedParts()
+ assert.Len(t, completedParts, 3, "Should have completed three parts")
+ for i, part := range completedParts {
+ assert.True(t, part.finishSyncCalled, "FinishSync should have been called for part %d", i+1)
+ assert.True(t, part.closeCalled, "Close should have been called for part %d", i+1)
+ }
+
+ t.Logf("Multiple parts metadata verification test completed successfully:")
+ t.Logf(" - Parts synced: %d", result.PartsCount)
+ t.Logf(" - Contexts received: %d", len(receivedContexts))
+ t.Logf(" - Files received: %d", len(receivedFiles))
+ t.Logf(" - Total bytes sent: %d", result.TotalBytes)
+
+ for i, ctx := range receivedContexts {
+ t.Logf(" - Part %d: ID=%d, Group=%s, ShardID=%d, CompressedSize=%d, TotalCount=%d",
+ i+1, ctx.ID, ctx.Group, ctx.ShardID, ctx.CompressedSizeBytes, ctx.TotalCount)
+ }
+}
diff --git a/banyand/queue/test/chunked_sync_performance_test.go b/banyand/queue/test/chunked_sync_performance_test.go
new file mode 100644
index 00000000..ee80ae44
--- /dev/null
+++ b/banyand/queue/test/chunked_sync_performance_test.go
@@ -0,0 +1,464 @@
+// 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 test
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+)
+
+// TestChunkedSyncLargeData tests syncing with data larger than chunk size.
+func TestChunkedSyncLargeData(t *testing.T) {
+ if testing.Short() {
+ t.Skip("Skipping large data test in short mode")
+ }
+
+ setup := setupChunkedSyncTestWithChunkSize(t, "chunked-sync-large-test", 512)
+ defer cleanupTestSetup(setup)
+
+ const largeDataSize = 2048
+ largeContent := make([]byte, largeDataSize)
+ for i := range largeContent {
+ largeContent[i] = byte(i % 256)
+ }
+
+ testFiles := map[string][]byte{"large-file.data": largeContent}
+ dataPart := createTestDataPart(testFiles, "test-group-large", 3)
+
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, []queue.StreamingPartData{dataPart})
+ require.NoError(t, err, "Failed to send large data part")
+ require.True(t, result.Success, "Large data sync should be successful")
+
+ time.Sleep(500 * time.Millisecond)
+
+ receivedFiles := setup.MockHandler.GetReceivedFiles()
+ require.Len(t, receivedFiles, 1, "Should receive the large file")
+
+ receivedContent := receivedFiles["large-file.data"]
+ assert.Equal(t, largeContent, receivedContent, "Large file content should match exactly")
+
+ assert.Greater(t, result.ChunksCount, uint32(1), "Large data should require multiple chunks")
+ assert.Greater(t, setup.MockHandler.GetReceivedChunksCount(), 1, "Should have received multiple chunks")
+
+ t.Logf("Large data test completed: %d bytes in %d chunks", result.TotalBytes, result.ChunksCount)
+}
+
+// TestChunkedSyncMultiplePartsSmallFiles tests syncing multiple parts with multiple small files.
+func TestChunkedSyncMultiplePartsSmallFiles(t *testing.T) {
+ setup := setupChunkedSyncTestWithChunkSize(t, "chunked-sync-multi-parts-small", 1024)
+ defer cleanupTestSetup(setup)
+
+ part1Files := map[string][]byte{
+ "part1-config.json": []byte(`{"part": 1, "type": "config", "data": "small"}`),
+ "part1-log.txt": []byte("Part 1 log entry: test data"),
+ "part1-meta.yaml": []byte("metadata:\n part: 1\n size: small"),
+ }
+
+ part2Files := map[string][]byte{
+ "part2-config.json": []byte(`{"part": 2, "type": "config", "data": "small"}`),
+ "part2-log.txt": []byte("Part 2 log entry: different data"),
+ "part2-meta.yaml": []byte("metadata:\n part: 2\n size: small"),
+ }
+
+ part3Files := map[string][]byte{
+ "part3-config.json": []byte(`{"part": 3, "type": "config", "data": "small"}`),
+ "part3-log.txt": []byte("Part 3 log entry: final data"),
+ "part3-meta.yaml": []byte("metadata:\n part: 3\n size: small"),
+ }
+
+ dataParts := []queue.StreamingPartData{
+ createTestDataPart(part1Files, "test-group-multi-parts", 1),
+ createTestDataPart(part2Files, "test-group-multi-parts", 2),
+ createTestDataPart(part3Files, "test-group-multi-parts", 3),
+ }
+
+ dataParts[1].ID = 2
+ dataParts[2].ID = 3
+
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, dataParts)
+ require.NoError(t, err, "Failed to send multiple parts")
+ require.True(t, result.Success, "Multiple parts sync should be successful")
+
+ time.Sleep(500 * time.Millisecond)
+
+ assert.Equal(t, uint32(3), result.PartsCount, "Should have synced 3 parts")
+ assert.Greater(t, result.ChunksCount, uint32(0), "Should have sent at least 1 chunk")
+ assert.Greater(t, result.TotalBytes, uint64(0), "Should have sent some bytes")
+
+ receivedFiles := setup.MockHandler.GetReceivedFiles()
+ expectedFileCount := len(part1Files) + len(part2Files) + len(part3Files)
+ require.Len(t, receivedFiles, expectedFileCount, "Should receive all files from all parts")
+
+ allExpectedFiles := make(map[string][]byte)
+ for k, v := range part1Files {
+ allExpectedFiles[k] = v
+ }
+ for k, v := range part2Files {
+ allExpectedFiles[k] = v
+ }
+ for k, v := range part3Files {
+ allExpectedFiles[k] = v
+ }
+
+ for fileName, expectedContent := range allExpectedFiles {
+ receivedContent, exists := receivedFiles[fileName]
+ require.True(t, exists, "File %s should be received", fileName)
+ assert.Equal(t, expectedContent, receivedContent, "Content for file %s should match", fileName)
+ }
+
+ completedParts := setup.MockHandler.GetCompletedParts()
+ assert.Len(t, completedParts, 3, "Should have completed three parts")
+ for i, part := range completedParts {
+ assert.True(t, part.finishSyncCalled, "FinishSync should have been called for part %d", i+1)
+ assert.True(t, part.closeCalled, "Close should have been called for part %d", i+1)
+ }
+
+ chunksCount := setup.MockHandler.GetReceivedChunksCount()
+ assert.GreaterOrEqual(t, chunksCount, 3, "Should have received at least 3 chunks (one per part minimum)")
+
+ t.Logf("Multiple parts with small files test completed:")
+ t.Logf(" - Parts synced: %d", result.PartsCount)
+ t.Logf(" - Files synced: %d", len(receivedFiles))
+ t.Logf(" - Chunks sent in total: %d", result.ChunksCount)
+ t.Logf(" - Chunks received in file-based sync: %d", chunksCount)
+ t.Logf(" - Total bytes: %d", result.TotalBytes)
+}
+
+// TestChunkedSyncMultiplePartsBigFiles tests syncing multiple parts with multiple big files.
+func TestChunkedSyncMultiplePartsBigFiles(t *testing.T) {
+ if testing.Short() {
+ t.Skip("Skipping multiple parts with big files test in short mode")
+ }
+
+ setup := setupChunkedSyncTestWithChunkSize(t, "chunked-sync-multi-parts-big", 512)
+ defer cleanupTestSetup(setup)
+
+ const bigFileSize = 1024
+
+ createBigFileContent := func(partNum int, fileName string) []byte {
+ content := make([]byte, bigFileSize)
+ for i := range content {
+ content[i] = byte((partNum*100 + len(fileName) + i) % 256)
+ }
+ return content
+ }
+
+ part1Files := map[string][]byte{
+ "part1-large-data.bin": createBigFileContent(1, "part1-large-data.bin"),
+ "part1-bulk-export.dat": createBigFileContent(1, "part1-bulk-export.dat"),
+ }
+
+ part2Files := map[string][]byte{
+ "part2-large-data.bin": createBigFileContent(2, "part2-large-data.bin"),
+ "part2-bulk-export.dat": createBigFileContent(2, "part2-bulk-export.dat"),
+ }
+
+ part3Files := map[string][]byte{
+ "part3-large-data.bin": createBigFileContent(3, "part3-large-data.bin"),
+ "part3-bulk-export.dat": createBigFileContent(3, "part3-bulk-export.dat"),
+ }
+
+ dataParts := []queue.StreamingPartData{
+ createTestDataPart(part1Files, "test-group-multi-parts-big", 1),
+ createTestDataPart(part2Files, "test-group-multi-parts-big", 2),
+ createTestDataPart(part3Files, "test-group-multi-parts-big", 3),
+ }
+
+ dataParts[1].ID = 2
+ dataParts[2].ID = 3
+
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, dataParts)
+ require.NoError(t, err, "Failed to send multiple parts with big files")
+ require.True(t, result.Success, "Multiple parts with big files sync should be successful")
+
+ time.Sleep(1 * time.Second)
+
+ assert.Equal(t, uint32(3), result.PartsCount, "Should have synced 3 parts")
+ assert.Greater(t, result.ChunksCount, uint32(6), "Should have sent multiple chunks per file (at least 6 total)")
+ assert.Equal(t, result.TotalBytes, uint64(6*bigFileSize), "Should have sent bytes for all big files")
+
+ receivedFiles := setup.MockHandler.GetReceivedFiles()
+ expectedFileCount := len(part1Files) + len(part2Files) + len(part3Files)
+ require.Len(t, receivedFiles, expectedFileCount, "Should receive all files from all parts")
+
+ allExpectedFiles := make(map[string][]byte)
+ for k, v := range part1Files {
+ allExpectedFiles[k] = v
+ }
+ for k, v := range part2Files {
+ allExpectedFiles[k] = v
+ }
+ for k, v := range part3Files {
+ allExpectedFiles[k] = v
+ }
+
+ for fileName, expectedContent := range allExpectedFiles {
+ receivedContent, exists := receivedFiles[fileName]
+ require.True(t, exists, "File %s should be received", fileName)
+ assert.Equal(t, expectedContent, receivedContent, "Content for file %s should match exactly", fileName)
+ assert.Equal(t, bigFileSize, len(receivedContent), "File %s should have correct size", fileName)
+ }
+
+ completedParts := setup.MockHandler.GetCompletedParts()
+ assert.Len(t, completedParts, 3, "Should have completed three parts")
+ for i, part := range completedParts {
+ assert.True(t, part.finishSyncCalled, "FinishSync should have been called for part %d", i+1)
+ assert.True(t, part.closeCalled, "Close should have been called for part %d", i+1)
+ }
+
+ chunksCount := setup.MockHandler.GetReceivedChunksCount()
+ expectedMinChunks := expectedFileCount * 2
+ assert.GreaterOrEqual(t, chunksCount, expectedMinChunks, "Should have received multiple chunks per big file")
+
+ totalReceivedBytes := uint64(0)
+ for _, part := range completedParts {
+ for _, chunk := range part.receivedChunks {
+ totalReceivedBytes += uint64(len(chunk))
+ }
+ }
+ assert.Equal(t, result.TotalBytes, totalReceivedBytes, "Total received bytes should match sent bytes")
+
+ t.Logf("Multiple parts with big files test completed:")
+ t.Logf(" - Parts synced: %d", result.PartsCount)
+ t.Logf(" - Files synced: %d", len(receivedFiles))
+ t.Logf(" - Chunks sent in total: %d", result.ChunksCount)
+ t.Logf(" - Chunks received in file-based sync: %d", chunksCount)
+ t.Logf(" - Total bytes: %d", result.TotalBytes)
+ t.Logf(" - Average chunks per file: %.2f", float64(chunksCount)/float64(expectedFileCount))
+}
+
+// TestChunkedSyncMultiplePartsTinyFiles tests syncing multiple parts with tiny files that can be batched into single chunks.
+func TestChunkedSyncMultiplePartsTinyFiles(t *testing.T) {
+ setup := setupChunkedSyncTestWithChunkSize(t, "chunked-sync-multi-parts-tiny", 2048)
+ defer cleanupTestSetup(setup)
+
+ part1Files := map[string][]byte{
+ "part1-a.txt": []byte("A"),
+ "part1-b.txt": []byte("B"),
+ "part1-c.txt": []byte("C"),
+ "part1-d.txt": []byte("D"),
+ "part1-e.txt": []byte("E"),
+ }
+
+ part2Files := map[string][]byte{
+ "part2-a.txt": []byte("AA"),
+ "part2-b.txt": []byte("BB"),
+ "part2-c.txt": []byte("CC"),
+ "part2-d.txt": []byte("DD"),
+ "part2-e.txt": []byte("EE"),
+ }
+
+ part3Files := map[string][]byte{
+ "part3-a.txt": []byte("AAA"),
+ "part3-b.txt": []byte("BBB"),
+ "part3-c.txt": []byte("CCC"),
+ "part3-d.txt": []byte("DDD"),
+ "part3-e.txt": []byte("EEE"),
+ }
+
+ dataParts := []queue.StreamingPartData{
+ createTestDataPart(part1Files, "test-group-multi-parts-tiny", 1),
+ createTestDataPart(part2Files, "test-group-multi-parts-tiny", 2),
+ createTestDataPart(part3Files, "test-group-multi-parts-tiny", 3),
+ }
+
+ dataParts[1].ID = 2
+ dataParts[2].ID = 3
+
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, dataParts)
+ require.NoError(t, err, "Failed to send multiple parts with tiny files")
+ require.True(t, result.Success, "Multiple parts with tiny files sync should be successful")
+
+ time.Sleep(500 * time.Millisecond)
+
+ assert.Equal(t, uint32(3), result.PartsCount, "Should have synced 3 parts")
+ assert.Greater(t, result.ChunksCount, uint32(0), "Should have sent at least 1 chunk")
+ assert.Greater(t, result.TotalBytes, uint64(0), "Should have sent some bytes")
+
+ receivedFiles := setup.MockHandler.GetReceivedFiles()
+ expectedFileCount := len(part1Files) + len(part2Files) + len(part3Files)
+ require.Len(t, receivedFiles, expectedFileCount, "Should receive all files from all parts")
+
+ allExpectedFiles := make(map[string][]byte)
+ for k, v := range part1Files {
+ allExpectedFiles[k] = v
+ }
+ for k, v := range part2Files {
+ allExpectedFiles[k] = v
+ }
+ for k, v := range part3Files {
+ allExpectedFiles[k] = v
+ }
+
+ for fileName, expectedContent := range allExpectedFiles {
+ receivedContent, exists := receivedFiles[fileName]
+ require.True(t, exists, "File %s should be received", fileName)
+ assert.Equal(t, expectedContent, receivedContent, "Content for file %s should match", fileName)
+ }
+
+ completedParts := setup.MockHandler.GetCompletedParts()
+ assert.Len(t, completedParts, 3, "Should have completed three parts")
+ for i, part := range completedParts {
+ assert.True(t, part.finishSyncCalled, "FinishSync should have been called for part %d", i+1)
+ assert.True(t, part.closeCalled, "Close should have been called for part %d", i+1)
+ }
+
+ chunksCount := setup.MockHandler.GetReceivedChunksCount()
+ totalFileBytes := uint64(0)
+ for _, content := range allExpectedFiles {
+ totalFileBytes += uint64(len(content))
+ }
+
+ expectedMaxChunks := uint32((totalFileBytes + 2047) / 2048)
+ assert.LessOrEqual(t, result.ChunksCount, expectedMaxChunks+1, "Should use fewer chunks due to batching")
+
+ t.Logf("Multiple parts with tiny files test completed:")
+ t.Logf(" - Parts synced: %d", result.PartsCount)
+ t.Logf(" - Files synced: %d", len(receivedFiles))
+ t.Logf(" - Chunks sent in total: %d", result.ChunksCount)
+ t.Logf(" - Chunks received in file-based sync: %d", chunksCount)
+ t.Logf(" - Total bytes: %d", result.TotalBytes)
+ t.Logf(" - Average files per chunk: %.2f", float64(expectedFileCount)/float64(chunksCount))
+}
+
+// TestChunkedSyncMixedFileSizes tests syncing with mixed file sizes that can be efficiently batched.
+func TestChunkedSyncMixedFileSizes(t *testing.T) {
+ setup := setupChunkedSyncTestWithChunkSize(t, "chunked-sync-mixed-sizes", 1024)
+ defer cleanupTestSetup(setup)
+
+ part1Files := map[string][]byte{
+ "part1-small1.txt": []byte("small1"),
+ "part1-small2.txt": []byte("small2"),
+ "part1-medium.txt": []byte(string(make([]byte, 200))),
+ "part1-large.txt": []byte(string(make([]byte, 800))),
+ "part1-tiny.txt": []byte("t"),
+ }
+
+ part2Files := map[string][]byte{
+ "part2-small1.txt": []byte("small1"),
+ "part2-small2.txt": []byte("small2"),
+ "part2-medium.txt": []byte(string(make([]byte, 300))),
+ "part2-large.txt": []byte(string(make([]byte, 900))),
+ "part2-tiny.txt": []byte("t"),
+ }
+
+ part3Files := map[string][]byte{
+ "part3-small1.txt": []byte("small1"),
+ "part3-small2.txt": []byte("small2"),
+ "part3-medium.txt": []byte(string(make([]byte, 250))),
+ "part3-large.txt": []byte(string(make([]byte, 850))),
+ "part3-tiny.txt": []byte("t"),
+ }
+
+ dataParts := []queue.StreamingPartData{
+ createTestDataPart(part1Files, "test-group-mixed-sizes", 1),
+ createTestDataPart(part2Files, "test-group-mixed-sizes", 2),
+ createTestDataPart(part3Files, "test-group-mixed-sizes", 3),
+ }
+
+ dataParts[1].ID = 2
+ dataParts[2].ID = 3
+
+ ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+ defer cancel()
+
+ result, err := setup.ChunkedClient.SyncStreamingParts(ctx, dataParts)
+ require.NoError(t, err, "Failed to send multiple parts with mixed file sizes")
+ require.True(t, result.Success, "Multiple parts with mixed file sizes sync should be successful")
+
+ time.Sleep(500 * time.Millisecond)
+
+ assert.Equal(t, uint32(3), result.PartsCount, "Should have synced 3 parts")
+ assert.Greater(t, result.ChunksCount, uint32(0), "Should have sent at least 1 chunk")
+ assert.Greater(t, result.TotalBytes, uint64(0), "Should have sent some bytes")
+
+ receivedFiles := setup.MockHandler.GetReceivedFiles()
+ expectedFileCount := len(part1Files) + len(part2Files) + len(part3Files)
+ require.Len(t, receivedFiles, expectedFileCount, "Should receive all files from all parts")
+
+ allExpectedFiles := make(map[string][]byte)
+ for k, v := range part1Files {
+ allExpectedFiles[k] = v
+ }
+ for k, v := range part2Files {
+ allExpectedFiles[k] = v
+ }
+ for k, v := range part3Files {
+ allExpectedFiles[k] = v
+ }
+
+ for fileName, expectedContent := range allExpectedFiles {
+ receivedContent, exists := receivedFiles[fileName]
+ require.True(t, exists, "File %s should be received", fileName)
+ assert.Equal(t, expectedContent, receivedContent, "Content for file %s should match", fileName)
+ }
+
+ completedParts := setup.MockHandler.GetCompletedParts()
+ assert.Len(t, completedParts, 3, "Should have completed three parts")
+ for i, part := range completedParts {
+ assert.True(t, part.finishSyncCalled, "FinishSync should have been called for part %d", i+1)
+ assert.True(t, part.closeCalled, "Close should have been called for part %d", i+1)
+ }
+
+ chunksCount := setup.MockHandler.GetReceivedChunksCount()
+ totalFileBytes := uint64(0)
+ for _, content := range allExpectedFiles {
+ totalFileBytes += uint64(len(content))
+ }
+
+ expectedLargeFileChunks := uint32(3)
+ remainingBytes := totalFileBytes - uint64(800+900+850)
+ expectedRemainingChunks := uint32((remainingBytes + 1023) / 1024)
+ expectedMinChunks := expectedLargeFileChunks + expectedRemainingChunks
+
+ assert.GreaterOrEqual(t, result.ChunksCount, expectedMinChunks, "Should use at least expected minimum chunks")
+ assert.LessOrEqual(t, result.ChunksCount, expectedMinChunks+3, "Should not use significantly more chunks than expected")
+
+ partChunkCounts := make([]int, 3)
+ for i, part := range completedParts {
+ partChunkCounts[i] = len(part.receivedChunks)
+ }
+
+ t.Logf("Mixed file sizes test completed:")
+ t.Logf(" - Parts synced: %d", result.PartsCount)
+ t.Logf(" - Files synced: %d", len(receivedFiles))
+ t.Logf(" - Chunks sent in total: %d", result.ChunksCount)
+ t.Logf(" - Chunks received in file-based sync: %d", chunksCount)
+ t.Logf(" - Total bytes: %d", result.TotalBytes)
+ t.Logf(" - Chunks per part: %v", partChunkCounts)
+ t.Logf(" - Average files per chunk: %.2f", float64(expectedFileCount)/float64(result.ChunksCount))
+ t.Logf(" - Chunking efficiency: %.2f%%", float64(totalFileBytes)/float64(result.ChunksCount*1024)*100)
+}
diff --git a/banyand/stream/block_writer.go b/banyand/stream/block_writer.go
index 33fc0764..48e352c8 100644
--- a/banyand/stream/block_writer.go
+++ b/banyand/stream/block_writer.go
@@ -89,6 +89,14 @@ func (sw *writers) reset() {
}
}
+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)
+}
+
func (sw *writers) totalBytesWritten() uint64 {
n := sw.metaWriter.bytesWritten + sw.primaryWriter.bytesWritten +
sw.timestampsWriter.bytesWritten
@@ -310,3 +318,22 @@ func releaseBlockWriter(bsw *blockWriter) {
}
var blockWriterPool = pool.Register[*blockWriter]("stream-blockWriter")
+
+func generateWriters() *writers {
+ v := writersPool.Get()
+ if v == nil {
+ return &writers{
+ tagFamilyMetadataWriters: make(map[string]*writer),
+ tagFamilyWriters: make(map[string]*writer),
+ tagFamilyFilterWriters: make(map[string]*writer),
+ }
+ }
+ return v
+}
+
+func releaseWriters(sw *writers) {
+ sw.reset()
+ writersPool.Put(sw)
+}
+
+var writersPool = pool.Register[*writers]("stream-writers")
diff --git a/banyand/stream/part.go b/banyand/stream/part.go
index 15620714..4bce9479 100644
--- a/banyand/stream/part.go
+++ b/banyand/stream/part.go
@@ -37,10 +37,18 @@ import (
)
const (
+ // Streaming file names (without extensions).
+ streamPrimaryName = "primary"
+ streamMetaName = "meta"
+ streamTimestampsName = "timestamps"
+ streamTagFamiliesPrefix = "tf:"
+ streamTagMetadataPrefix = "tfm:"
+ streamTagFilterPrefix = "tff:"
+
metadataFilename = "metadata.json"
- primaryFilename = "primary.bin"
- metaFilename = "meta.bin"
- timestampsFilename = "timestamps.bin"
+ primaryFilename = streamPrimaryName + ".bin"
+ metaFilename = streamMetaName + ".bin"
+ timestampsFilename = streamTimestampsName + ".bin"
elementIndexFilename = "idx"
tagFamiliesMetadataFilenameExt = ".tfm"
tagFamiliesFilenameExt = ".tf"
diff --git a/banyand/stream/svc_liaison.go b/banyand/stream/svc_liaison.go
index 0239077d..a70aaed2 100644
--- a/banyand/stream/svc_liaison.go
+++ b/banyand/stream/svc_liaison.go
@@ -125,6 +125,10 @@ func (s *liaison) PreRun(ctx context.Context) error {
streamDataNodeRegistry := grpc.NewClusterNodeRegistry(data.TopicStreamPartSync, s.option.tire2Client, s.dataNodeSelector)
s.schemaRepo = newLiaisonSchemaRepo(s.dataPath, s, streamDataNodeRegistry)
s.writeListener = setUpWriteQueueCallback(s.l, &s.schemaRepo, s.maxDiskUsagePercent, s.option.tire2Client)
+
+ // Register chunked sync handler for stream data
+ s.pipeline.RegisterChunkedSyncHandler(data.TopicStreamPartSync, setUpChunkedSyncCallback(s.l, &s.schemaRepo))
+
return s.pipeline.Subscribe(data.TopicStreamWrite, s.writeListener)
}
diff --git a/banyand/stream/svc_standalone.go b/banyand/stream/svc_standalone.go
index f9f6a9fa..ab1e93fa 100644
--- a/banyand/stream/svc_standalone.go
+++ b/banyand/stream/svc_standalone.go
@@ -164,10 +164,8 @@ func (s *standalone) PreRun(ctx context.Context) error {
if err != nil {
return err
}
- err = s.pipeline.Subscribe(data.TopicStreamPartSync, setUpSyncCallback(s.l, &s.schemaRepo))
- if err != nil {
- return err
- }
+ s.pipeline.RegisterChunkedSyncHandler(data.TopicStreamPartSync, setUpChunkedSyncCallback(s.l, &s.schemaRepo))
+
err = s.pipeline.Subscribe(data.TopicStreamSeriesIndexWrite, setUpSeriesIndexCallback(s.l, &s.schemaRepo))
if err != nil {
return err
diff --git a/banyand/stream/syncer.go b/banyand/stream/syncer.go
index 8f7060d8..44ce0ea3 100644
--- a/banyand/stream/syncer.go
+++ b/banyand/stream/syncer.go
@@ -23,9 +23,8 @@ import (
"time"
"github.com/apache/skywalking-banyandb/api/data"
- "github.com/apache/skywalking-banyandb/pkg/bus"
- "github.com/apache/skywalking-banyandb/pkg/convert"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/compress/zstd"
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
@@ -60,6 +59,7 @@ func (tst *tsTable) syncLoop(syncCh chan *syncIntroduction, flusherNotifier watc
}
tst.l.Logger.Warn().Err(err).Msgf("cannot sync snapshot: %d", curSnapshot.epoch)
tst.incTotalSyncLoopErr(1)
+ time.Sleep(2 * time.Second)
return false
}
epoch = curSnapshot.epoch
@@ -76,6 +76,61 @@ func (tst *tsTable) syncLoop(syncCh chan *syncIntroduction, flusherNotifier watc
}
}
+func createPartFileReaders(part *part) ([]queue.FileInfo, func()) {
+ var files []queue.FileInfo
+
+ buf := bigValuePool.Generate()
+ // Stream metadata
+ for i := range part.primaryBlockMetadata {
+ buf.Buf = part.primaryBlockMetadata[i].marshal(buf.Buf)
+ }
+ bb := bigValuePool.Generate()
+ bb.Buf = zstd.Compress(bb.Buf[:0], buf.Buf, 1)
+ bigValuePool.Release(buf)
+ files = append(files,
+ queue.FileInfo{
+ Name: streamMetaName,
+ Reader: bb.SequentialRead(),
+ },
+ queue.FileInfo{
+ Name: streamPrimaryName,
+ Reader: part.primary.SequentialRead(),
+ },
+ queue.FileInfo{
+ Name: streamTimestampsName,
+ Reader: part.timestamps.SequentialRead(),
+ },
+ )
+
+ // Stream tag families data
+ if part.tagFamilies != nil {
+ for name, reader := range part.tagFamilies {
+ files = append(files, queue.FileInfo{
+ Name: fmt.Sprintf("%s%s", streamTagFamiliesPrefix, name),
+ Reader: reader.SequentialRead(),
+ })
+ }
+
+ for name, reader := range part.tagFamilyMetadata {
+ files = append(files, queue.FileInfo{
+ Name: fmt.Sprintf("%s%s", streamTagMetadataPrefix, name),
+ Reader: reader.SequentialRead(),
+ })
+ }
+
+ for name, reader := range part.tagFamilyFilter {
+ files = append(files, queue.FileInfo{
+ Name: fmt.Sprintf("%s%s", streamTagFilterPrefix, name),
+ Reader: reader.SequentialRead(),
+ })
+ }
+ }
+
+ return files, func() {
+ bigValuePool.Release(bb)
+ }
+}
+
func (tst *tsTable) syncSnapshot(curSnapshot *snapshot, syncCh chan *syncIntroduction) error {
// Get all parts from the current snapshot
var partsToSync []*part
@@ -103,41 +158,63 @@ func (tst *tsTable) syncSnapshot(curSnapshot *snapshot, syncCh chan *syncIntrodu
}
}
- // Send parts through pipeline
+ // Use chunked sync with streaming for better memory efficiency
ctx := context.Background()
+ releaseFuncs := make([]func(), 0, len(partsToSync))
+ defer func() {
+ for _, release := range releaseFuncs {
+ release()
+ }
+ }()
for _, node := range nodes {
+ // Get chunked sync client for this node
+ chunkedClient, err := tst.option.tire2Client.NewChunkedSyncClient(node, 1024*1024)
+ if err != nil {
+ return fmt.Errorf("failed to create chunked sync client for node %s: %w", node, err)
+ }
+ defer chunkedClient.Close()
+
+ // Prepare streaming parts data for chunked sync
+ var streamingParts []queue.StreamingPartData
for _, part := range partsToSync {
- // Convert part to memPart
- memPart := generateMemPart()
- memPart.mustInitFromPart(part)
-
- // Marshal memPart to []byte
- partData, err := memPart.Marshal()
- releaseMemPart(memPart)
- if err != nil {
- return fmt.Errorf("failed to marshal part %d: %w", part.partMetadata.ID, err)
- }
+ // Create streaming reader for the part
+ files, release := createPartFileReaders(part)
+ releaseFuncs = append(releaseFuncs, release)
+
+ // Create streaming part sync data
+ streamingParts = append(streamingParts, queue.StreamingPartData{
+ ID: part.partMetadata.ID,
+ Group: tst.group,
+ ShardID: uint32(tst.shardID),
+ Topic: data.TopicStreamPartSync.String(),
+ Files: files,
+ CompressedSizeBytes: part.partMetadata.CompressedSizeBytes,
+ UncompressedSizeBytes: part.partMetadata.UncompressedSizeBytes,
+ TotalCount: part.partMetadata.TotalCount,
+ BlocksCount: part.partMetadata.BlocksCount,
+ MinTimestamp: part.partMetadata.MinTimestamp,
+ MaxTimestamp: part.partMetadata.MaxTimestamp,
+ })
+ }
- // Encode group, shardID and prepend to partData
- combinedData := make([]byte, 0, len(partData)+len(tst.group)+4)
- combinedData = encoding.EncodeBytes(combinedData, convert.StringToBytes(tst.group))
- combinedData = encoding.Uint32ToBytes(combinedData, uint32(tst.shardID))
- combinedData = append(combinedData, partData...)
-
- // Create message with combined data (group + shardID + partData)
- message := bus.NewMessageWithNode(bus.MessageID(time.Now().UnixNano()), node, combinedData)
- f, err := tst.option.tire2Client.Publish(ctx, data.TopicStreamPartSync, message)
- if err != nil {
- return fmt.Errorf("failed to publish part %d to pipeline: %w", part.partMetadata.ID, err)
- }
- if f != nil {
- _, err = f.GetAll()
- if err != nil {
- tst.l.Warn().Err(err).Msgf("failed to get all sync results: %v", err)
- }
- }
+ // Sync parts using chunked transfer with streaming
+ result, err := chunkedClient.SyncStreamingParts(ctx, streamingParts)
+ if err != nil {
+ return fmt.Errorf("failed to sync streaming parts to node %s: %w", node, err)
+ }
+
+ if !result.Success {
+ return fmt.Errorf("chunked sync partially failed: %v", result.ErrorMessage)
}
+ tst.l.Info().
+ Str("node", node).
+ Str("session", result.SessionID).
+ Uint64("bytes", result.TotalBytes).
+ Int64("duration_ms", result.DurationMs).
+ Uint32("chunks", result.ChunksCount).
+ Uint32("parts", result.PartsCount).
+ Msg("chunked sync completed successfully")
}
// Construct syncIntroduction to remove synced parts from snapshot
diff --git a/banyand/stream/wqueue_test.go b/banyand/stream/wqueue_test.go
index 49f94fc1..da48678a 100644
--- a/banyand/stream/wqueue_test.go
+++ b/banyand/stream/wqueue_test.go
@@ -18,6 +18,7 @@
package stream
import (
+ "context"
"testing"
"time"
@@ -43,11 +44,20 @@ func Test_newWriteQueue(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockTire2Client := queue.NewMockClient(ctrl)
+
+ // Mock Publish method
mockTire2Client.EXPECT().
Publish(gomock.Any(), gomock.Any(), gomock.Any()).
Return(bus.Future(nil), nil).
AnyTimes()
+ // Mock NewChunkedSyncClient method
+ mockChunkedSyncClient := &mockChunkedSyncClient{}
+ mockTire2Client.EXPECT().
+ NewChunkedSyncClient(gomock.Any(), gomock.Any()).
+ Return(mockChunkedSyncClient, nil).
+ AnyTimes()
+
// Create tsTable using newWriteQueue
opts := option{
tire2Client: mockTire2Client,
@@ -88,3 +98,21 @@ func Test_newWriteQueue(t *testing.T) {
// Verify that the tire2Client was properly set
assert.Equal(t, mockTire2Client, tst.option.tire2Client)
}
+
+// mockChunkedSyncClient implements queue.ChunkedSyncClient for testing
+type mockChunkedSyncClient struct{}
+
+func (m *mockChunkedSyncClient) SyncStreamingParts(ctx context.Context, parts []queue.StreamingPartData) (*queue.SyncResult, error) {
+ return &queue.SyncResult{
+ Success: true,
+ SessionID: "mock-session",
+ TotalBytes: 0,
+ DurationMs: 0,
+ ChunksCount: 0,
+ PartsCount: uint32(len(parts)),
+ }, nil
+}
+
+func (m *mockChunkedSyncClient) Close() error {
+ return nil
+}
diff --git a/banyand/stream/write_data.go b/banyand/stream/write_data.go
index d39343e2..20da4773 100644
--- a/banyand/stream/write_data.go
+++ b/banyand/stream/write_data.go
@@ -18,268 +18,120 @@
package stream
import (
- "context"
+ "fmt"
+ "strings"
"time"
"github.com/apache/skywalking-banyandb/api/common"
- "github.com/apache/skywalking-banyandb/pkg/bus"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
- "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
-type syncCallback struct {
- l *logger.Logger
- schemaRepo *schemaRepo
+type syncPartContext struct {
+ tsTable *tsTable
+ writers *writers
+ memPart *memPart
}
-func setUpSyncCallback(l *logger.Logger, schemaRepo *schemaRepo) bus.MessageListener {
- return &syncCallback{
- l: l,
- schemaRepo: schemaRepo,
- }
+func (s *syncPartContext) FinishSync() error {
+ s.tsTable.mustAddMemPart(s.memPart)
+ return s.Close()
}
-func (s *syncCallback) CheckHealth() *common.Error {
+func (s *syncPartContext) Close() error {
+ s.writers.MustClose()
+ releaseWriters(s.writers)
+ s.writers = nil
+ s.memPart = nil
+ s.tsTable = nil
return nil
}
-func (s *syncCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
- data, ok := message.Data().([]byte)
- if !ok {
- s.l.Warn().Msg("invalid sync message data type")
- return
- }
-
- // Decode group name
- tail, groupBytes, err := encoding.DecodeBytes(data)
- if err != nil {
- s.l.Error().Err(err).Msg("failed to decode group name from sync message")
- return
- }
- group := string(groupBytes)
-
- // Decode shardID (uint32)
- if len(tail) < 4 {
- s.l.Error().Msg("insufficient data for shardID in sync message")
- return
- }
- shardID := encoding.BytesToUint32(tail[:4])
- tail = tail[4:]
-
- // Decode memPart
- memPart := generateMemPart()
-
- if err = memPart.Unmarshal(tail); err != nil {
- s.l.Error().Err(err).Msg("failed to unmarshal memPart from sync message")
- return
- }
-
- // Get the group schema from schemaRepo by the string group
- tsdb, err := s.schemaRepo.loadTSDB(group)
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
- return
- }
-
- // Get the segment using memPart.partMetadata.MinTimestamp
- segmentTime := time.Unix(0, memPart.partMetadata.MinTimestamp)
- segment, err := tsdb.CreateSegmentIfNotExist(segmentTime)
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Time("segmentTime", segmentTime).Msg("failed to create segment")
- return
- }
- defer segment.DecRef()
-
- // Get Shard from a segment using shardID
- tsTable, err := segment.CreateTSTableIfNotExist(common.ShardID(shardID))
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Uint32("shardID", shardID).Msg("failed to create ts table")
- return
- }
-
- tsdb.Tick(memPart.partMetadata.MaxTimestamp)
- // Use tsTable.mustAddMemPart to introduce memPart to tsTable
- tsTable.mustAddMemPart(memPart)
-
- if e := s.l.Debug(); e.Enabled() {
- e.
- Str("group", group).
- Uint32("shardID", shardID).
- Uint64("partID", memPart.partMetadata.ID).
- Msg("successfully stored memPart to tsTable")
- }
-
- return
-}
-
-type seriesIndexCallback struct {
+type syncCallback struct {
l *logger.Logger
schemaRepo *schemaRepo
}
-func setUpSeriesIndexCallback(l *logger.Logger, schemaRepo *schemaRepo) bus.MessageListener {
- return &seriesIndexCallback{
+func setUpChunkedSyncCallback(l *logger.Logger, schemaRepo *schemaRepo) queue.ChunkedSyncHandler {
+ return &syncCallback{
l: l,
schemaRepo: schemaRepo,
}
}
-func (s *seriesIndexCallback) CheckHealth() *common.Error {
+func (s *syncCallback) CheckHealth() *common.Error {
return nil
}
-func (s *seriesIndexCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
- data, ok := message.Data().([]byte)
- if !ok {
- s.l.Warn().Msg("invalid series index message data type")
- return
- }
-
- // Decode group name (first part of the message)
- tail, groupBytes, err := encoding.DecodeBytes(data)
+// CreatePartHandler implements queue.ChunkedSyncHandler.
+func (s *syncCallback) CreatePartHandler(ctx *queue.ChunkedSyncPartContext) (queue.PartHandler, error) {
+ tsdb, err := s.schemaRepo.loadTSDB(ctx.Group)
if err != nil {
- s.l.Error().Err(err).Msg("failed to decode group name from series index message")
- return
- }
- group := string(groupBytes)
-
- // Decode timestamp (uint64) - second part of the message
- if len(tail) < 8 {
- s.l.Error().Msg("insufficient data for timestamp in series index message")
- return
- }
- timestamp := encoding.BytesToInt64(tail[:8])
- tail = tail[8:]
-
- // Unmarshal Documents from the remaining data
- var documents index.Documents
- if err = documents.Unmarshal(tail); err != nil {
- s.l.Error().Err(err).Msg("failed to unmarshal documents from series index message")
- return
+ s.l.Error().Err(err).Str("group", ctx.Group).Msg("failed to load TSDB for group")
+ return nil, err
}
-
- if len(documents) == 0 {
- s.l.Warn().Msg("empty documents in series index message")
- return
- }
-
- // Get TSDB by group name using schemaRepo
- tsdb, err := s.schemaRepo.loadTSDB(group)
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
- return
- }
-
- // Use the timestamp to get the segment directly
- segmentTime := time.Unix(0, timestamp)
+ segmentTime := time.Unix(0, ctx.MinTimestamp)
segment, err := tsdb.CreateSegmentIfNotExist(segmentTime)
if err != nil {
- s.l.Error().Err(err).Str("group", group).Time("segmentTime", segmentTime).Msg("failed to create segment")
- return
+ s.l.Error().Err(err).Str("group", ctx.Group).Time("segmentTime", segmentTime).Msg("failed to create segment")
+ return nil, err
}
defer segment.DecRef()
-
- // Insert all documents into the segment's index
- if err := segment.IndexDB().Insert(documents); err != nil {
- s.l.Error().Err(err).Str("group", group).Int("documentCount", len(documents)).Msg("failed to insert documents to index")
- return
+ tsTable, err := segment.CreateTSTableIfNotExist(common.ShardID(ctx.ShardID))
+ if err != nil {
+ s.l.Error().Err(err).Str("group", ctx.Group).Uint32("shardID", ctx.ShardID).Msg("failed to create ts table")
+ return nil, err
}
- s.l.Info().
- Str("group", group).
- Int("documentCount", len(documents)).
- Time("segmentTime", segmentTime).
- Msg("successfully inserted documents to series index")
-
- return
-}
-
-type localIndexCallback struct {
- l *logger.Logger
- schemaRepo *schemaRepo
+ 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
}
-func setUpLocalIndexCallback(l *logger.Logger, schemaRepo *schemaRepo) bus.MessageListener {
- return &localIndexCallback{
- l: l,
- schemaRepo: schemaRepo,
+// HandleFileChunk implements queue.ChunkedSyncHandler for streaming file chunks.
+func (s *syncCallback) HandleFileChunk(ctx *queue.ChunkedSyncPartContext, chunk []byte) error {
+ if ctx.Handler == nil {
+ return fmt.Errorf("part handler is nil")
+ }
+ partCtx := ctx.Handler.(*syncPartContext)
+
+ // Select the appropriate writer based on the filename and write the chunk.
+ fileName := ctx.FileName
+ switch {
+ case fileName == streamMetaName:
+ partCtx.writers.metaWriter.MustWrite(chunk)
+ case fileName == streamPrimaryName:
+ partCtx.writers.primaryWriter.MustWrite(chunk)
+ case fileName == streamTimestampsName:
+ partCtx.writers.timestampsWriter.MustWrite(chunk)
+ case strings.HasPrefix(fileName, streamTagFamiliesPrefix):
+ tagName := fileName[len(streamTagFamiliesPrefix):]
+ _, tagWriter, _ := partCtx.writers.getWriters(tagName)
+ tagWriter.MustWrite(chunk)
+ case strings.HasPrefix(fileName, streamTagMetadataPrefix):
+ tagName := fileName[len(streamTagMetadataPrefix):]
+ tagMetadataWriter, _, _ := partCtx.writers.getWriters(tagName)
+ tagMetadataWriter.MustWrite(chunk)
+ case strings.HasPrefix(fileName, streamTagFilterPrefix):
+ tagName := fileName[len(streamTagFilterPrefix):]
+ _, _, tagFilterWriter := partCtx.writers.getWriters(tagName)
+ tagFilterWriter.MustWrite(chunk)
+ default:
+ s.l.Warn().Str("fileName", fileName).Msg("unknown file type in chunked sync")
+ return fmt.Errorf("unknown file type: %s", fileName)
}
-}
-func (l *localIndexCallback) CheckHealth() *common.Error {
return nil
}
-
-func (l *localIndexCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
- data, ok := message.Data().([]byte)
- if !ok {
- l.l.Warn().Msg("invalid local index message data type")
- return
- }
-
- // Decode group name (first part of the message)
- tail, groupBytes, err := encoding.DecodeBytes(data)
- if err != nil {
- l.l.Error().Err(err).Msg("failed to decode group name from local index message")
- return
- }
- group := string(groupBytes)
-
- // Decode shardID (uint32)
- if len(tail) < 4 {
- l.l.Error().Msg("insufficient data for shardID in local index message")
- return
- }
- shardID := encoding.BytesToUint32(tail[:4])
- tail = tail[4:]
-
- // Unmarshal Documents from the remaining data
- var documents index.Documents
- if err = documents.Unmarshal(tail); err != nil {
- l.l.Error().Err(err).Msg("failed to unmarshal documents from local index message")
- return
- }
-
- if len(documents) == 0 {
- l.l.Warn().Msg("empty documents in local index message")
- return
- }
-
- // Get TSDB by group name using schemaRepo
- tsdb, err := l.schemaRepo.loadTSDB(group)
- if err != nil {
- l.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
- return
- }
-
- // Use the first document's timestamp to get the segment
- firstDocTime := time.Unix(0, documents[0].Timestamp)
- segment, err := tsdb.CreateSegmentIfNotExist(firstDocTime)
- if err != nil {
- l.l.Error().Err(err).Str("group", group).Time("docTime", firstDocTime).Msg("failed to create segment for documents")
- return
- }
- defer segment.DecRef()
-
- // Get the tsTable using shardID
- tsTable, err := segment.CreateTSTableIfNotExist(common.ShardID(shardID))
- if err != nil {
- l.l.Error().Err(err).Str("group", group).Uint32("shardID", shardID).Msg("failed to create ts table")
- return
- }
-
- // Insert documents into tsTable.index
- if err := tsTable.Index().Write(documents); err != nil {
- l.l.Error().Err(err).Str("group", group).Uint32("shardID", shardID).Int("documentCount", len(documents)).Msg("failed to insert documents to tsTable index")
- return
- }
-
- l.l.Info().
- Str("group", group).
- Uint32("shardID", shardID).
- Int("documentCount", len(documents)).
- Msg("successfully inserted documents to tsTable index")
-
- return
-}
diff --git a/banyand/stream/write_data.go b/banyand/stream/write_index.go
similarity index 72%
copy from banyand/stream/write_data.go
copy to banyand/stream/write_index.go
index d39343e2..868619c8 100644
--- a/banyand/stream/write_data.go
+++ b/banyand/stream/write_index.go
@@ -28,91 +28,6 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
)
-type syncCallback struct {
- l *logger.Logger
- schemaRepo *schemaRepo
-}
-
-func setUpSyncCallback(l *logger.Logger, schemaRepo *schemaRepo) bus.MessageListener {
- return &syncCallback{
- l: l,
- schemaRepo: schemaRepo,
- }
-}
-
-func (s *syncCallback) CheckHealth() *common.Error {
- return nil
-}
-
-func (s *syncCallback) Rev(_ context.Context, message bus.Message) (resp bus.Message) {
- data, ok := message.Data().([]byte)
- if !ok {
- s.l.Warn().Msg("invalid sync message data type")
- return
- }
-
- // Decode group name
- tail, groupBytes, err := encoding.DecodeBytes(data)
- if err != nil {
- s.l.Error().Err(err).Msg("failed to decode group name from sync message")
- return
- }
- group := string(groupBytes)
-
- // Decode shardID (uint32)
- if len(tail) < 4 {
- s.l.Error().Msg("insufficient data for shardID in sync message")
- return
- }
- shardID := encoding.BytesToUint32(tail[:4])
- tail = tail[4:]
-
- // Decode memPart
- memPart := generateMemPart()
-
- if err = memPart.Unmarshal(tail); err != nil {
- s.l.Error().Err(err).Msg("failed to unmarshal memPart from sync message")
- return
- }
-
- // Get the group schema from schemaRepo by the string group
- tsdb, err := s.schemaRepo.loadTSDB(group)
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Msg("failed to load TSDB for group")
- return
- }
-
- // Get the segment using memPart.partMetadata.MinTimestamp
- segmentTime := time.Unix(0, memPart.partMetadata.MinTimestamp)
- segment, err := tsdb.CreateSegmentIfNotExist(segmentTime)
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Time("segmentTime", segmentTime).Msg("failed to create segment")
- return
- }
- defer segment.DecRef()
-
- // Get Shard from a segment using shardID
- tsTable, err := segment.CreateTSTableIfNotExist(common.ShardID(shardID))
- if err != nil {
- s.l.Error().Err(err).Str("group", group).Uint32("shardID", shardID).Msg("failed to create ts table")
- return
- }
-
- tsdb.Tick(memPart.partMetadata.MaxTimestamp)
- // Use tsTable.mustAddMemPart to introduce memPart to tsTable
- tsTable.mustAddMemPart(memPart)
-
- if e := s.l.Debug(); e.Enabled() {
- e.
- Str("group", group).
- Uint32("shardID", shardID).
- Uint64("partID", memPart.partMetadata.ID).
- Msg("successfully stored memPart to tsTable")
- }
-
- return
-}
-
type seriesIndexCallback struct {
l *logger.Logger
schemaRepo *schemaRepo
diff --git a/docs/api-reference.md b/docs/api-reference.md
index d940ff57..d6a82d28 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -7,11 +7,22 @@
- [Status](#banyandb-model-v1-Status)
- [banyandb/cluster/v1/rpc.proto](#banyandb_cluster_v1_rpc-proto)
+ - [FileInfo](#banyandb-cluster-v1-FileInfo)
- [HealthCheckRequest](#banyandb-cluster-v1-HealthCheckRequest)
- [HealthCheckResponse](#banyandb-cluster-v1-HealthCheckResponse)
+ - [PartInfo](#banyandb-cluster-v1-PartInfo)
+ - [PartResult](#banyandb-cluster-v1-PartResult)
- [SendRequest](#banyandb-cluster-v1-SendRequest)
- [SendResponse](#banyandb-cluster-v1-SendResponse)
+ - [SyncCompletion](#banyandb-cluster-v1-SyncCompletion)
+ - [SyncMetadata](#banyandb-cluster-v1-SyncMetadata)
+ - [SyncPartRequest](#banyandb-cluster-v1-SyncPartRequest)
+ - [SyncPartResponse](#banyandb-cluster-v1-SyncPartResponse)
+ - [SyncResult](#banyandb-cluster-v1-SyncResult)
+ - [SyncStatus](#banyandb-cluster-v1-SyncStatus)
+
+ - [ChunkedSyncService](#banyandb-cluster-v1-ChunkedSyncService)
- [Service](#banyandb-cluster-v1-Service)
- [banyandb/common/v1/common.proto](#banyandb_common_v1_common-proto)
@@ -327,6 +338,23 @@ Status is the response status for write
+<a name="banyandb-cluster-v1-FileInfo"></a>
+
+### FileInfo
+Information about an individual file within a part
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| name | [string](#string) | | File identifier (e.g., "primary", "timestamps", "tagFamilies:seriesId") |
+| offset | [uint32](#uint32) | | Byte offset within the part where this file starts |
+| size | [uint32](#uint32) | | Size of this file in bytes |
+
+
+
+
+
+
<a name="banyandb-cluster-v1-HealthCheckRequest"></a>
### HealthCheckRequest
@@ -359,6 +387,45 @@ Status is the response status for write
+<a name="banyandb-cluster-v1-PartInfo"></a>
+
+### PartInfo
+Information about a part contained within a chunk
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| id | [uint64](#uint64) | | Unique identifier for this part |
+| files | [FileInfo](#banyandb-cluster-v1-FileInfo) | repeated | Information about individual files within this part |
+| compressed_size_bytes | [uint64](#uint64) | | Compressed size in bytes from partMetadata |
+| uncompressed_size_bytes | [uint64](#uint64) | | Uncompressed size in bytes from partMetadata |
+| total_count | [uint64](#uint64) | | Total count from partMetadata |
+| blocks_count | [uint64](#uint64) | | Blocks count from partMetadata |
+| min_timestamp | [int64](#int64) | | Minimum timestamp from partMetadata |
+| max_timestamp | [int64](#int64) | | Maximum timestamp from partMetadata |
+
+
+
+
+
+
+<a name="banyandb-cluster-v1-PartResult"></a>
+
+### PartResult
+Result for individual parts
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| success | [bool](#bool) | | Whether this part was processed successfully |
+| error | [string](#string) | | Error message if processing failed |
+| bytes_processed | [uint32](#uint32) | | Number of bytes processed for this part |
+
+
+
+
+
+
<a name="banyandb-cluster-v1-SendRequest"></a>
### SendRequest
@@ -394,13 +461,135 @@ Status is the response status for write
+
+<a name="banyandb-cluster-v1-SyncCompletion"></a>
+
+### SyncCompletion
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| total_bytes_sent | [uint64](#uint64) | | Total bytes sent for validation |
+| total_parts_sent | [uint32](#uint32) | | Total number of parts sent |
+| total_chunks | [uint32](#uint32) | | Total number of chunks in this sync |
+
+
+
+
+
+
+<a name="banyandb-cluster-v1-SyncMetadata"></a>
+
+### SyncMetadata
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| group | [string](#string) | | Group name (stream/measure) |
+| shard_id | [uint32](#uint32) | | Shard identifier |
+| topic | [string](#string) | | Sync topic (stream-part-sync or measure-part-sync) |
+| timestamp | [int64](#int64) | | Timestamp when sync started |
+| total_parts | [uint32](#uint32) | | Total number of parts being synced |
+
+
+
+
+
+
+<a name="banyandb-cluster-v1-SyncPartRequest"></a>
+
+### SyncPartRequest
+Chunked Sync Service Messages
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| session_id | [string](#string) | | Unique session identifier for this sync operation |
+| chunk_index | [uint32](#uint32) | | Current chunk index (0-based) |
+| chunk_data | [bytes](#bytes) | | Actual chunk data |
+| chunk_checksum | [string](#string) | | CRC32 checksum for this chunk |
+| parts_info | [PartInfo](#banyandb-cluster-v1-PartInfo) | repeated | Information about parts contained in this chunk |
+| metadata | [SyncMetadata](#banyandb-cluster-v1-SyncMetadata) | | Sent with first chunk (chunk_index = 0) |
+| completion | [SyncCompletion](#banyandb-cluster-v1-SyncCompletion) | | Sent with last chunk to finalize |
+
+
+
+
+
+
+<a name="banyandb-cluster-v1-SyncPartResponse"></a>
+
+### SyncPartResponse
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| session_id | [string](#string) | | |
+| chunk_index | [uint32](#uint32) | | |
+| status | [SyncStatus](#banyandb-cluster-v1-SyncStatus) | | |
+| error | [string](#string) | | |
+| sync_result | [SyncResult](#banyandb-cluster-v1-SyncResult) | | Final result when sync completes |
+
+
+
+
+
+
+<a name="banyandb-cluster-v1-SyncResult"></a>
+
+### SyncResult
+
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| success | [bool](#bool) | | Whether entire sync was successful |
+| total_bytes_received | [uint64](#uint64) | | Total bytes received |
+| duration_ms | [int64](#int64) | | Time taken for sync in milliseconds |
+| chunks_received | [uint32](#uint32) | | Number of chunks successfully received |
+| parts_received | [uint32](#uint32) | | Number of parts successfully received |
+| parts_results | [PartResult](#banyandb-cluster-v1-PartResult) | repeated | Results for each part |
+
+
+
+
+
+
+<a name="banyandb-cluster-v1-SyncStatus"></a>
+
+### SyncStatus
+
+
+| Name | Number | Description |
+| ---- | ------ | ----------- |
+| SYNC_STATUS_UNSPECIFIED | 0 | Unspecified status |
+| SYNC_STATUS_CHUNK_RECEIVED | 1 | Chunk received and validated successfully |
+| SYNC_STATUS_CHUNK_CHECKSUM_MISMATCH | 2 | Chunk checksum validation failed |
+| SYNC_STATUS_CHUNK_OUT_OF_ORDER | 3 | Chunk received out of expected order |
+| SYNC_STATUS_SESSION_NOT_FOUND | 4 | Session ID not recognized |
+| SYNC_STATUS_SYNC_COMPLETE | 5 | Entire sync operation completed successfully |
+
+
+<a name="banyandb-cluster-v1-ChunkedSyncService"></a>
+
+### ChunkedSyncService
+New streaming sync service for chunked data transfer
+
+| Method Name | Request Type | Response Type | Description |
+| ----------- | ------------ | ------------- | ------------|
+| SyncPart | [SyncPartRequest](#banyandb-cluster-v1-SyncPartRequest) stream | [SyncPartResponse](#banyandb-cluster-v1-SyncPartResponse) stream | Synchronize part data using chunked transfer |
+
+
<a name="banyandb-cluster-v1-Service"></a>
### Service