Skip to content

M
milvus

milvus / milvus
10 个月 前同步成功

通知 260
Star 22476
Fork 2472
M
milvus

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

未验证 提交 23baecd7 编写于 作者: W wei liu 提交者： GitHub
浏览文件
操作

set sealed segment to unreadable before sync target version (#26338) 

Signed-off-by: NWei Liu <wei.liu@zilliz.com>
上级 376642b9
隐藏空白更改
内联 并排
Showing with 79 addition and 49 deletion
internal/querynodev2/delegator/delegator_data_test.go
浏览文件 @ 23baecd7
......@@ -377,9 +377,10 @@ func (s *DelegatorDataSuite) TestLoadSegments() {
s.Equal(int64(1), sealed[0].NodeID)
s.ElementsMatch([]SegmentEntry{
{
SegmentID: 100,
NodeID: 1,
PartitionID: 500,
SegmentID: 100,
NodeID: 1,
PartitionID: 500,
TargetVersion: unreadableTargetVersion,
},
}, sealed[0].Segments)
})
......@@ -453,14 +454,16 @@ func (s *DelegatorDataSuite) TestLoadSegments() {
s.Equal(int64(1), sealed[0].NodeID)
s.ElementsMatch([]SegmentEntry{
{
SegmentID: 100,
NodeID: 1,
PartitionID: 500,
SegmentID: 100,
NodeID: 1,
PartitionID: 500,
TargetVersion: unreadableTargetVersion,
},
{
SegmentID: 200,
NodeID: 1,
PartitionID: 500,
SegmentID: 200,
NodeID: 1,
PartitionID: 500,
TargetVersion: unreadableTargetVersion,
},
}, sealed[0].Segments)
})
......@@ -657,9 +660,10 @@ func (s *DelegatorDataSuite) TestReleaseSegment() {
s.Equal(int64(1), sealed[0].NodeID)
s.ElementsMatch([]SegmentEntry{
{
SegmentID: 1000,
NodeID: 1,
PartitionID: 500,
SegmentID: 1000,
NodeID: 1,
PartitionID: 500,
TargetVersion: unreadableTargetVersion,
},
}, sealed[0].Segments)
......
internal/querynodev2/delegator/delegator_test.go
浏览文件 @ 23baecd7
......@@ -192,10 +192,11 @@ func (s *DelegatorSuite) TestGetSegmentInfo() {
NodeID: 1,
Segments: []SegmentEntry{
{
NodeID: 1,
SegmentID: 1001,
PartitionID: 500,
Version: 2001,
NodeID: 1,
SegmentID: 1001,
PartitionID: 500,
Version: 2001,
TargetVersion: unreadableTargetVersion,
},
},
},
......
internal/querynodev2/delegator/distribution.go
浏览文件 @ 23baecd7
......@@ -35,6 +35,9 @@ const (
// for growing segment which not exist in target, and it's start position < max sealed dml position
redundantTargetVersion = int64(-1)
// for sealed segment which loaded by load segment request, should become readable after sync target version
unreadableTargetVersion = int64(-2)
)
var (
......@@ -186,6 +189,9 @@ func (d *distribution) AddDistributions(entries ...SegmentEntry) {
if ok {
// remain the target version for already loaded segment to void skipping this segment when executing search
entry.TargetVersion = oldEntry.TargetVersion
} else {
// waiting for sync target version, to become readable
entry.TargetVersion = unreadableTargetVersion
}
d.sealedSegments[entry.SegmentID] = entry
d.offlines.Remove(entry.SegmentID)
......
internal/querynodev2/delegator/distribution_test.go
浏览文件 @ 23baecd7
......@@ -64,12 +64,14 @@ func (s *DistributionSuite) TestAddDistribution() {
NodeID: 1,
Segments: []SegmentEntry{
{
NodeID: 1,
SegmentID: 1,
NodeID: 1,
SegmentID: 1,
TargetVersion: unreadableTargetVersion,
},
{
NodeID: 1,
SegmentID: 2,
NodeID: 1,
SegmentID: 2,
TargetVersion: unreadableTargetVersion,
},
},
},
......@@ -93,8 +95,9 @@ func (s *DistributionSuite) TestAddDistribution() {
NodeID: 1,
Segments: []SegmentEntry{
{
NodeID: 1,
SegmentID: 1,
NodeID: 1,
SegmentID: 1,
TargetVersion: unreadableTargetVersion,
},
},
},
......@@ -122,13 +125,15 @@ func (s *DistributionSuite) TestAddDistribution() {
NodeID: 1,
Segments: []SegmentEntry{
{
NodeID: 1,
SegmentID: 1,
NodeID: 1,
SegmentID: 1,
TargetVersion: unreadableTargetVersion,
},
{
NodeID: 1,
SegmentID: 3,
NodeID: 1,
SegmentID: 3,
TargetVersion: unreadableTargetVersion,
},
},
},
......@@ -136,8 +141,9 @@ func (s *DistributionSuite) TestAddDistribution() {
NodeID: 2,
Segments: []SegmentEntry{
{
NodeID: 2,
SegmentID: 2,
NodeID: 2,
SegmentID: 2,
TargetVersion: unreadableTargetVersion,
},
},
},
......@@ -157,8 +163,8 @@ func (s *DistributionSuite) TestAddDistribution() {
{
NodeID: 1,
Segments: []SegmentEntry{
{NodeID: 1, SegmentID: 1},
{NodeID: 1, SegmentID: 2},
{NodeID: 1, SegmentID: 1, TargetVersion: unreadableTargetVersion},
{NodeID: 1, SegmentID: 2, TargetVersion: unreadableTargetVersion},
},
},
},
......@@ -287,13 +293,13 @@ func (s *DistributionSuite) TestRemoveDistribution() {
{
NodeID: 1,
Segments: []SegmentEntry{
{NodeID: 1, SegmentID: 3},
{NodeID: 1, SegmentID: 3, TargetVersion: unreadableTargetVersion},
},
},
{
NodeID: 2,
Segments: []SegmentEntry{
{NodeID: 2, SegmentID: 2},
{NodeID: 2, SegmentID: 2, TargetVersion: unreadableTargetVersion},
},
},
},
......@@ -322,14 +328,14 @@ func (s *DistributionSuite) TestRemoveDistribution() {
{
NodeID: 1,
Segments: []SegmentEntry{
{NodeID: 1, SegmentID: 1},
{NodeID: 1, SegmentID: 3},
{NodeID: 1, SegmentID: 1, TargetVersion: unreadableTargetVersion},
{NodeID: 1, SegmentID: 3, TargetVersion: unreadableTargetVersion},
},
},
{
NodeID: 2,
Segments: []SegmentEntry{
{NodeID: 2, SegmentID: 2},
{NodeID: 2, SegmentID: 2, TargetVersion: unreadableTargetVersion},
},
},
},
......@@ -358,13 +364,13 @@ func (s *DistributionSuite) TestRemoveDistribution() {
{
NodeID: 1,
Segments: []SegmentEntry{
{NodeID: 1, SegmentID: 3},
{NodeID: 1, SegmentID: 3, TargetVersion: unreadableTargetVersion},
},
},
{
NodeID: 2,
Segments: []SegmentEntry{
{NodeID: 2, SegmentID: 2},
{NodeID: 2, SegmentID: 2, TargetVersion: unreadableTargetVersion},
},
},
},
......@@ -395,13 +401,21 @@ func (s *DistributionSuite) TestRemoveDistribution() {
{
NodeID: 1,
Segments: []SegmentEntry{
{NodeID: 1, SegmentID: 3},
{
NodeID: 1,
SegmentID: 3,
TargetVersion: unreadableTargetVersion,
},
},
},
{
NodeID: 2,
Segments: []SegmentEntry{
{NodeID: 2, SegmentID: 2},
{
NodeID: 2,
SegmentID: 2,
TargetVersion: unreadableTargetVersion,
},
},
},
},
......@@ -476,12 +490,14 @@ func (s *DistributionSuite) TestPeek() {
NodeID: 1,
Segments: []SegmentEntry{
{
NodeID: 1,
SegmentID: 1,
NodeID: 1,
SegmentID: 1,
TargetVersion: unreadableTargetVersion,
},
{
NodeID: 1,
SegmentID: 2,
NodeID: 1,
SegmentID: 2,
TargetVersion: unreadableTargetVersion,
},
},
},
......@@ -508,13 +524,15 @@ func (s *DistributionSuite) TestPeek() {
NodeID: 1,
Segments: []SegmentEntry{
{
NodeID: 1,
SegmentID: 1,
NodeID: 1,
SegmentID: 1,
TargetVersion: unreadableTargetVersion,
},
{
NodeID: 1,
SegmentID: 3,
NodeID: 1,
SegmentID: 3,
TargetVersion: unreadableTargetVersion,
},
},
},
......@@ -522,8 +540,9 @@ func (s *DistributionSuite) TestPeek() {
NodeID: 2,
Segments: []SegmentEntry{
{
NodeID: 2,
SegmentID: 2,
NodeID: 2,
SegmentID: 2,
TargetVersion: unreadableTargetVersion,
},
},
},
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或