Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
milvus
milvus
提交
884f7904
M
milvus
项目概览
milvus
/
milvus
大约 1 年 前同步成功
通知
261
Star
22476
Fork
2472
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
M
milvus
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
884f7904
编写于
5月 06, 2023
作者:
C
congqixia
提交者:
GitHub
5月 06, 2023
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Re-calculate segment entry num when binlog field is zero (#23883)
Signed-off-by:
N
Congqi Xia
<
congqi.xia@zilliz.com
>
上级
8ffa8a28
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
125 addition
and
0 deletion
+125
-0
internal/querynodev2/segments/segment_loader.go
internal/querynodev2/segments/segment_loader.go
+70
-0
internal/querynodev2/segments/segment_loader_test.go
internal/querynodev2/segments/segment_loader_test.go
+55
-0
未找到文件。
internal/querynodev2/segments/segment_loader.go
浏览文件 @
884f7904
...
...
@@ -375,6 +375,11 @@ func (loader *segmentLoader) loadSegment(ctx context.Context,
if
err
:=
loader
.
loadSealedSegmentFields
(
ctx
,
segment
,
fieldBinlogs
,
loadInfo
);
err
!=
nil
{
return
err
}
// https://github.com/milvus-io/milvus/23654
// legacy entry num = 0
if
err
:=
loader
.
patchEntryNumber
(
ctx
,
segment
,
loadInfo
);
err
!=
nil
{
return
err
}
}
else
{
if
err
:=
loader
.
loadGrowingSegmentFields
(
ctx
,
segment
,
loadInfo
.
BinlogPaths
);
err
!=
nil
{
return
err
...
...
@@ -779,6 +784,71 @@ func (loader *segmentLoader) LoadDeltaLogs(ctx context.Context, segment *LocalSe
return
nil
}
func
(
loader
*
segmentLoader
)
patchEntryNumber
(
ctx
context
.
Context
,
segment
*
LocalSegment
,
loadInfo
*
querypb
.
SegmentLoadInfo
)
error
{
var
needReset
bool
segment
.
fieldIndexes
.
Range
(
func
(
fieldID
int64
,
info
*
IndexedFieldInfo
)
bool
{
for
_
,
info
:=
range
info
.
FieldBinlog
.
GetBinlogs
()
{
if
info
.
GetEntriesNum
()
==
0
{
needReset
=
true
return
false
}
}
return
true
})
if
!
needReset
{
return
nil
}
log
.
Warn
(
"legacy segment binlog found, start to patch entry num"
,
zap
.
Int64
(
"segmentID"
,
segment
.
segmentID
))
rowIDField
:=
lo
.
FindOrElse
(
loadInfo
.
BinlogPaths
,
nil
,
func
(
binlog
*
datapb
.
FieldBinlog
)
bool
{
return
binlog
.
GetFieldID
()
==
common
.
RowIDField
})
if
rowIDField
==
nil
{
return
errors
.
New
(
"rowID field binlog not found"
)
}
counts
:=
make
([]
int64
,
0
,
len
(
rowIDField
.
GetBinlogs
()))
for
_
,
binlog
:=
range
rowIDField
.
GetBinlogs
()
{
bs
,
err
:=
loader
.
cm
.
Read
(
ctx
,
binlog
.
LogPath
)
if
err
!=
nil
{
return
err
}
// get binlog entry num from rowID field
// since header does not store entry numb, we have to read all data here
reader
,
err
:=
storage
.
NewBinlogReader
(
bs
)
if
err
!=
nil
{
return
err
}
er
,
err
:=
reader
.
NextEventReader
()
if
err
!=
nil
{
return
err
}
rowIDs
,
err
:=
er
.
GetInt64FromPayload
()
if
err
!=
nil
{
return
err
}
counts
=
append
(
counts
,
int64
(
len
(
rowIDs
)))
}
var
err
error
segment
.
fieldIndexes
.
Range
(
func
(
fieldID
int64
,
info
*
IndexedFieldInfo
)
bool
{
if
len
(
info
.
FieldBinlog
.
GetBinlogs
())
!=
len
(
counts
)
{
err
=
errors
.
New
(
"rowID & index binlog number not matched"
)
return
false
}
for
i
,
binlog
:=
range
info
.
FieldBinlog
.
GetBinlogs
()
{
binlog
.
EntriesNum
=
counts
[
i
]
}
return
true
})
return
err
}
// JoinIDPath joins ids to path format.
func
JoinIDPath
(
ids
...
UniqueID
)
string
{
idStr
:=
make
([]
string
,
0
,
len
(
ids
))
...
...
internal/querynodev2/segments/segment_loader_test.go
浏览文件 @
884f7904
...
...
@@ -353,6 +353,61 @@ func (suite *SegmentLoaderSuite) TestLoadWithMmap() {
suite
.
NoError
(
err
)
}
func
(
suite
*
SegmentLoaderSuite
)
TestPatchEntryNum
()
{
ctx
:=
context
.
Background
()
segmentID
:=
suite
.
segmentID
binlogs
,
statsLogs
,
err
:=
SaveBinLog
(
ctx
,
suite
.
collectionID
,
suite
.
partitionID
,
segmentID
,
100
,
suite
.
schema
,
suite
.
chunkManager
,
)
suite
.
NoError
(
err
)
vecFields
:=
funcutil
.
GetVecFieldIDs
(
suite
.
schema
)
indexInfo
,
err
:=
GenAndSaveIndex
(
suite
.
collectionID
,
suite
.
partitionID
,
segmentID
,
vecFields
[
0
],
100
,
IndexFaissIVFFlat
,
L2
,
suite
.
chunkManager
,
)
suite
.
NoError
(
err
)
loadInfo
:=
&
querypb
.
SegmentLoadInfo
{
SegmentID
:
segmentID
,
PartitionID
:
suite
.
partitionID
,
CollectionID
:
suite
.
collectionID
,
BinlogPaths
:
binlogs
,
Statslogs
:
statsLogs
,
IndexInfos
:
[]
*
querypb
.
FieldIndexInfo
{
indexInfo
},
}
// mock legacy binlog entry num is zero case
for
_
,
fieldLog
:=
range
binlogs
{
for
_
,
binlog
:=
range
fieldLog
.
GetBinlogs
()
{
binlog
.
EntriesNum
=
0
}
}
segments
,
err
:=
suite
.
loader
.
Load
(
ctx
,
suite
.
collectionID
,
SegmentTypeSealed
,
0
,
loadInfo
)
suite
.
Require
()
.
NoError
(
err
)
suite
.
Require
()
.
Equal
(
1
,
len
(
segments
))
segment
:=
segments
[
0
]
info
:=
segment
.
GetIndex
(
vecFields
[
0
])
suite
.
Require
()
.
NotNil
(
info
)
for
_
,
binlog
:=
range
info
.
FieldBinlog
.
GetBinlogs
()
{
suite
.
Greater
(
binlog
.
EntriesNum
,
int64
(
0
))
}
}
func
TestSegmentLoader
(
t
*
testing
.
T
)
{
suite
.
Run
(
t
,
&
SegmentLoaderSuite
{})
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录