Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Refine bulkload code #17737

Closed
wants to merge 1 commit into from
Closed

Refine bulkload code #17737

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 16 additions & 21 deletions internal/core/src/pb/common.pb.cc
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 0 additions & 3 deletions internal/core/src/pb/common.pb.h
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

16 changes: 8 additions & 8 deletions internal/datanode/data_node.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -971,14 +971,14 @@ func (node *DataNode) AddSegment(ctx context.Context, req *datapb.AddSegmentRequ
}

func importFlushReqFunc(node *DataNode, req *datapb.ImportTaskRequest, res *rootcoordpb.ImportResult, schema *schemapb.CollectionSchema, ts Timestamp) importutil.ImportFlushFunc {
return func(fields map[storage.FieldID]storage.FieldData, shardNum int) error {
if shardNum >= len(req.GetImportTask().GetChannelNames()) {
return func(fields map[storage.FieldID]storage.FieldData, shardID int) error {
if shardID >= len(req.GetImportTask().GetChannelNames()) {
log.Error("import task returns invalid shard number",
zap.Int("shard num", shardNum),
zap.Int("shard num", shardID),
zap.Int("# of channels", len(req.GetImportTask().GetChannelNames())),
zap.Any("channel names", req.GetImportTask().GetChannelNames()),
)
return fmt.Errorf("syncSegmentID Failed: invalid shard number %d", shardNum)
return fmt.Errorf("syncSegmentID Failed: invalid shard ID %d", shardID)
}

tr := timerecord.NewTimeRecorder("import callback function")
Expand All @@ -993,10 +993,10 @@ func importFlushReqFunc(node *DataNode, req *datapb.ImportTaskRequest, res *root
}

// ask DataCoord to alloc a new segment
log.Info("import task flush segment", zap.Any("ChannelNames", req.ImportTask.ChannelNames), zap.Int("shardNum", shardNum))
log.Info("import task flush segment", zap.Any("ChannelNames", req.ImportTask.ChannelNames), zap.Int("shardID", shardID))
segReqs := []*datapb.SegmentIDRequest{
{
ChannelName: req.ImportTask.ChannelNames[shardNum],
ChannelName: req.ImportTask.ChannelNames[shardID],
Count: uint32(rowNum),
CollectionID: req.GetImportTask().GetCollectionId(),
PartitionID: req.GetImportTask().GetPartitionId(),
Expand Down Expand Up @@ -1078,8 +1078,8 @@ func importFlushReqFunc(node *DataNode, req *datapb.ImportTaskRequest, res *root
kvs[key] = blob.Value[:]
field2Insert[fieldID] = &datapb.Binlog{
EntriesNum: data.size,
TimestampFrom: 0, //TODO
TimestampTo: 0, //TODO,
TimestampFrom: ts,
TimestampTo: ts,
LogPath: key,
LogSize: int64(len(blob.Value)),
}
Expand Down
13 changes: 5 additions & 8 deletions internal/proto/common.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,14 +262,11 @@ enum ConsistencyLevel {
}

enum ImportState {
ImportPending = 0;
ImportFailed = 1;
ImportStarted = 2;
ImportDownloaded = 3;
ImportParsed = 4;
ImportPersisted = 5;
ImportCompleted = 6;
ImportAllocSegment = 10;
ImportPending = 0; // the task in in pending list of rootCoord, waiting to be executed
ImportFailed = 1; // the task failed for some reason, get detail reason from GetImportStateResponse.infos
ImportStarted = 2; // the task has been sent to datanode to execute
ImportPersisted = 5; // all data files have been parsed and data already persisted
ImportAllocSegment = 10; // internal useage, no need to expose in sdk
}

enum ResourceType {
Expand Down
Loading