(skywalking-banyandb) branch main updated: fix(sidx): use MinTimestamp/MaxTimestamp instead of SegmentID in streaming sync (#1060)

[ha...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git


The following commit(s) were added to refs/heads/main by this push:
     new b8e1f72d3 fix(sidx): use MinTimestamp/MaxTimestamp instead of SegmentID in streaming sync (#1060)
b8e1f72d3 is described below

commit b8e1f72d3dd4c7026e752a17e875211ead4ccb21
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sat Apr 11 07:57:50 2026 +0800

    fix(sidx): use MinTimestamp/MaxTimestamp instead of SegmentID in streaming sync (#1060)
    
    * fix(sidx): use MinTimestamp/MaxTimestamp instead of SegmentID in streaming sync
    
    SIDX StreamingParts was incorrectly setting MinTimestamp from
    partMetadata.SegmentID and omitting MaxTimestamp entirely. This caused
    the receiving node to reject parts with "invalid MinTimestamp 0" when
    SegmentID was zero, and previously created corrupt seg-19700101
    directories before validation was added in #1059.
---
 CHANGES.md                         |   1 +
 banyand/internal/sidx/sync.go      |  19 +++-
 banyand/internal/sidx/sync_test.go | 191 +++++++++++++++++++++++++++++++++++++
 3 files changed, 208 insertions(+), 3 deletions(-)

diff --git a/CHANGES.md b/CHANGES.md
index 48c2d04fb..939f46a83 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -29,6 +29,7 @@ Release Notes.
 - ui: fix query editor refresh/reset behavior and BydbQL keyword highlighting.
 - Disable the rotation task on warm and cold nodes to prevent incorrect segment boundaries during lifecycle migration.
 - Prevent epoch-dated segment directories (seg-19700101) from being created by zero timestamps in distributed sync paths.
+- Fix SIDX streaming sync sending SegmentID as MinTimestamp instead of the actual timestamp, causing sync failures on the receiving node.
 
 ### Chores
 
diff --git a/banyand/internal/sidx/sync.go b/banyand/internal/sidx/sync.go
index 465df54eb..96352364f 100644
--- a/banyand/internal/sidx/sync.go
+++ b/banyand/internal/sidx/sync.go
@@ -74,7 +74,7 @@ func (s *sidx) StreamingParts(partIDsToSync map[uint64]struct{}, group string, s
 			part := pw.p
 			files, release := createPartFileReaders(part)
 			releaseFuncs = append(releaseFuncs, release)
-			streamingParts = append(streamingParts, queue.StreamingPartData{
+			spd := queue.StreamingPartData{
 				ID:                    part.partMetadata.ID,
 				Group:                 group,
 				ShardID:               shardID,
@@ -84,11 +84,24 @@ func (s *sidx) StreamingParts(partIDsToSync map[uint64]struct{}, group string, s
 				UncompressedSizeBytes: part.partMetadata.UncompressedSizeBytes,
 				TotalCount:            part.partMetadata.TotalCount,
 				BlocksCount:           part.partMetadata.BlocksCount,
-				MinTimestamp:          part.partMetadata.SegmentID,
 				MinKey:                part.partMetadata.MinKey,
 				MaxKey:                part.partMetadata.MaxKey,
 				PartType:              name,
-			})
+			}
+			switch {
+			case part.partMetadata.MinTimestamp != nil:
+				spd.MinTimestamp = *part.partMetadata.MinTimestamp
+			case part.partMetadata.SegmentID > 0:
+				spd.MinTimestamp = part.partMetadata.SegmentID
+			default:
+				logger.Panicf("sidx streaming parts: %s, part %d has no valid timestamp (MinTimestamp=nil, SegmentID=0)", name, part.partMetadata.ID)
+			}
+			if part.partMetadata.MaxTimestamp != nil {
+				spd.MaxTimestamp = *part.partMetadata.MaxTimestamp
+			} else {
+				spd.MaxTimestamp = spd.MinTimestamp
+			}
+			streamingParts = append(streamingParts, spd)
 		}
 	}
 
diff --git a/banyand/internal/sidx/sync_test.go b/banyand/internal/sidx/sync_test.go
new file mode 100644
index 000000000..ff6a32435
--- /dev/null
+++ b/banyand/internal/sidx/sync_test.go
@@ -0,0 +1,191 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package sidx
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/apache/skywalking-banyandb/api/data"
+	"github.com/apache/skywalking-banyandb/banyand/observability"
+	"github.com/apache/skywalking-banyandb/banyand/protector"
+	"github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+// TestStreamingParts_Timestamps verifies that StreamingParts propagates
+// MinTimestamp and MaxTimestamp from partMetadata (not SegmentID) to
+// StreamingPartData. This prevents zero-timestamp segments from being
+// created on the receiving node during distributed sync.
+func TestStreamingParts_Timestamps(t *testing.T) {
+	reqs := []WriteRequest{
+		createTestWriteRequest(1, 100, "data1"),
+		createTestWriteRequest(1, 200, "data2"),
+	}
+
+	t.Run("with_timestamps_set", func(t *testing.T) {
+		sidxIface := createTestSIDX(t)
+		raw := sidxIface.(*sidx)
+		defer func() {
+			assert.NoError(t, raw.Close())
+		}()
+
+		minTS := int64(1700000000)
+		maxTS := int64(1700001000)
+		writeTestDataWithTimeRange(t, raw, reqs, 1, 1, &minTS, &maxTS)
+
+		flushIntro, err := raw.Flush(map[uint64]struct{}{1: {}})
+		require.NoError(t, err)
+		raw.IntroduceFlushed(flushIntro)
+		flushIntro.Release()
+
+		partIDs := map[uint64]struct{}{1: {}}
+		parts, releaseFuncs := raw.StreamingParts(partIDs, "test-group", 0, "test-sidx")
+		defer func() {
+			for _, release := range releaseFuncs {
+				release()
+			}
+		}()
+
+		require.Len(t, parts, 1)
+		assert.Equal(t, uint64(1), parts[0].ID)
+		assert.Equal(t, int64(1700000000), parts[0].MinTimestamp,
+			"MinTimestamp should come from partMetadata.MinTimestamp, not SegmentID")
+		assert.Equal(t, int64(1700001000), parts[0].MaxTimestamp,
+			"MaxTimestamp should come from partMetadata.MaxTimestamp")
+		assert.Equal(t, "test-group", parts[0].Group)
+		assert.Equal(t, uint32(0), parts[0].ShardID)
+		assert.Equal(t, data.TopicTracePartSync.String(), parts[0].Topic)
+		assert.Equal(t, "test-sidx", parts[0].PartType)
+	})
+
+	t.Run("nil_timestamps_fallback_to_segment_id", func(t *testing.T) {
+		sidxIface := createTestSIDX(t)
+		raw := sidxIface.(*sidx)
+		defer func() {
+			assert.NoError(t, raw.Close())
+		}()
+
+		// segmentID=1000, nil timestamps — should fall back to SegmentID
+		writeTestDataWithTimeRange(t, raw, reqs, 1000, 1, nil, nil)
+
+		flushIntro, err := raw.Flush(map[uint64]struct{}{1: {}})
+		require.NoError(t, err)
+		raw.IntroduceFlushed(flushIntro)
+		flushIntro.Release()
+
+		partIDs := map[uint64]struct{}{1: {}}
+		parts, releaseFuncs := raw.StreamingParts(partIDs, "test-group", 0, "test-sidx")
+		defer func() {
+			for _, release := range releaseFuncs {
+				release()
+			}
+		}()
+
+		require.Len(t, parts, 1)
+		assert.Equal(t, int64(1000), parts[0].MinTimestamp,
+			"MinTimestamp should fall back to SegmentID when partMetadata.MinTimestamp is nil")
+		assert.Equal(t, int64(1000), parts[0].MaxTimestamp,
+			"MaxTimestamp should fall back to MinTimestamp when partMetadata.MaxTimestamp is nil")
+	})
+
+	t.Run("nil_timestamps_zero_segment_id_panics", func(t *testing.T) {
+		sidxIface := createTestSIDX(t)
+		raw := sidxIface.(*sidx)
+		defer func() {
+			assert.NoError(t, raw.Close())
+		}()
+
+		// segmentID=0, nil timestamps — should panic
+		writeTestDataWithTimeRange(t, raw, reqs, 0, 1, nil, nil)
+
+		flushIntro, err := raw.Flush(map[uint64]struct{}{1: {}})
+		require.NoError(t, err)
+		raw.IntroduceFlushed(flushIntro)
+		flushIntro.Release()
+
+		partIDs := map[uint64]struct{}{1: {}}
+		assert.Panics(t, func() {
+			parts, releaseFuncs := raw.StreamingParts(partIDs, "test-group", 0, "test-sidx")
+			for _, release := range releaseFuncs {
+				release()
+			}
+			_ = parts
+		}, "Should panic when both MinTimestamp and SegmentID are zero")
+	})
+
+	t.Run("nil_snapshot_returns_nil", func(t *testing.T) {
+		sidxIface := createTestSIDX(t)
+		raw := sidxIface.(*sidx)
+		defer func() {
+			assert.NoError(t, raw.Close())
+		}()
+
+		partIDs := map[uint64]struct{}{1: {}}
+		parts, releaseFuncs := raw.StreamingParts(partIDs, "test-group", 0, "test-sidx")
+		defer func() {
+			for _, release := range releaseFuncs {
+				release()
+			}
+		}()
+		assert.Nil(t, parts)
+	})
+
+	t.Run("multiple_parts_sorted_by_id", func(t *testing.T) {
+		dir := t.TempDir()
+		fileSystem := fs.NewLocalFileSystem()
+		opts := NewDefaultOptions()
+		opts.Memory = protector.NewMemory(observability.NewBypassRegistry())
+		opts.Path = dir
+
+		sidxIface, err := NewSIDX(fileSystem, opts)
+		require.NoError(t, err)
+		raw := sidxIface.(*sidx)
+		defer func() {
+			assert.NoError(t, raw.Close())
+		}()
+
+		min1, max1 := int64(1700000000), int64(1700001000)
+		min2, max2 := int64(1800000000), int64(1800001000)
+		writeTestDataWithTimeRange(t, raw, reqs, 1, 2, &min1, &max1)
+		writeTestDataWithTimeRange(t, raw, reqs, 2, 3, &min2, &max2)
+
+		flushIntro, flushErr := raw.Flush(map[uint64]struct{}{2: {}, 3: {}})
+		require.NoError(t, flushErr)
+		raw.IntroduceFlushed(flushIntro)
+		flushIntro.Release()
+
+		partIDs := map[uint64]struct{}{2: {}, 3: {}}
+		parts, releaseFuncs := raw.StreamingParts(partIDs, "test-group", 0, "test-sidx")
+		defer func() {
+			for _, release := range releaseFuncs {
+				release()
+			}
+		}()
+
+		require.Len(t, parts, 2)
+		// Parts should be sorted by ID
+		assert.Equal(t, uint64(2), parts[0].ID)
+		assert.Equal(t, int64(1700000000), parts[0].MinTimestamp)
+		assert.Equal(t, int64(1700001000), parts[0].MaxTimestamp)
+		assert.Equal(t, uint64(3), parts[1].ID)
+		assert.Equal(t, int64(1800000000), parts[1].MinTimestamp)
+		assert.Equal(t, int64(1800001000), parts[1].MaxTimestamp)
+	})
+}