milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix compaction target segment rowNum is always 0 (#20937) 

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
pull/20914/head
congqixia 2022-12-01 20:33:17 +08:00 committed by GitHub
parent e5c9da7152
commit f745d7f489
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 22 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/datanode/binlog_io.go
View File

@ -285,7 +285,7 @@ func (b *binlogIO) genInsertBlobs(data *InsertData, partID, segID UniqueID, meta
kvs[key] = value kvs[key] = value
inpaths[fID] = &datapb.FieldBinlog{ inpaths[fID] = &datapb.FieldBinlog{
FieldID: fID, FieldID: fID,
Binlogs: []*datapb.Binlog{{LogSize: int64(fileLen), LogPath: key}}, Binlogs: []*datapb.Binlog{{LogSize: int64(fileLen), LogPath: key, EntriesNum: blob.RowNum}},
} }
} }
@ -302,7 +302,7 @@ func (b *binlogIO) genInsertBlobs(data *InsertData, partID, segID UniqueID, meta
kvs[key] = value kvs[key] = value
statspaths[fID] = &datapb.FieldBinlog{ statspaths[fID] = &datapb.FieldBinlog{
FieldID: fID, FieldID: fID,
Binlogs: []*datapb.Binlog{{LogSize: int64(fileLen), LogPath: key}}, Binlogs: []*datapb.Binlog{{LogSize: int64(fileLen), LogPath: key, EntriesNum: blob.RowNum}},
} }
} }

18
internal/storage/data_codec.go
View File

@ -64,9 +64,10 @@ const InvalidUniqueID = UniqueID(-1)
// Blob is a pack of key&value // Blob is a pack of key&value
type Blob struct { type Blob struct {
Key string Key string
Value []byte Value []byte
Size int64 Size int64
RowNum int64
} }
// BlobList implements sort.Interface for a list of Blob // BlobList implements sort.Interface for a list of Blob
@ -277,6 +278,7 @@ func (insertCodec *InsertCodec) Serialize(partitionID UniqueID, segmentID Unique
if timeFieldData.RowNum() <= 0 { if timeFieldData.RowNum() <= 0 {
return nil, nil, fmt.Errorf("there's no data in InsertData") return nil, nil, fmt.Errorf("there's no data in InsertData")
} }
rowNum := int64(timeFieldData.RowNum())
ts := timeFieldData.(*Int64FieldData).Data ts := timeFieldData.(*Int64FieldData).Data
startTs := ts[0] startTs := ts[0]
@ -420,8 +422,9 @@ func (insertCodec *InsertCodec) Serialize(partitionID UniqueID, segmentID Unique
} }
blobKey := fmt.Sprintf("%d", field.FieldID) blobKey := fmt.Sprintf("%d", field.FieldID)
blobs = append(blobs, &Blob{ blobs = append(blobs, &Blob{
Key: blobKey, Key: blobKey,
Value: buffer, Value: buffer,
RowNum: rowNum,
}) })
eventWriter.Close() eventWriter.Close()
writer.Close() writer.Close()
@ -435,8 +438,9 @@ func (insertCodec *InsertCodec) Serialize(partitionID UniqueID, segmentID Unique
} }
statsBuffer := statsWriter.GetBuffer() statsBuffer := statsWriter.GetBuffer()
statsBlobs = append(statsBlobs, &Blob{ statsBlobs = append(statsBlobs, &Blob{
Key: blobKey, Key: blobKey,
Value: statsBuffer, Value: statsBuffer,
RowNum: rowNum,
}) })
} }
} }

18
internal/storage/index_data_codec_test.go
View File

@ -118,19 +118,19 @@ func TestIndexCodec(t *testing.T) {
indexCodec := NewIndexCodec() indexCodec := NewIndexCodec()
blobs := []*Blob{ blobs := []*Blob{
{ {
"12345", Key: "12345",
[]byte{1, 2, 3, 4, 5, 6, 7, 1, 2, 3, 4, 5, 6, 7}, Value: []byte{1, 2, 3, 4, 5, 6, 7, 1, 2, 3, 4, 5, 6, 7},
14, Size: 14,
}, },
{ {
"6666", Key: "6666",
[]byte{6, 6, 6, 6, 6, 1, 2, 3, 4, 5, 6, 7}, Value: []byte{6, 6, 6, 6, 6, 1, 2, 3, 4, 5, 6, 7},
12, Size: 12,
}, },
{ {
"8885", Key: "8885",
[]byte{8, 8, 8, 8, 8, 8, 8, 8, 2, 3, 4, 5, 6, 7}, Value: []byte{8, 8, 8, 8, 8, 8, 8, 8, 2, 3, 4, 5, 6, 7},
14, Size: 14,
}, },
} }
indexParams := map[string]string{ indexParams := map[string]string{