diff --git a/internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp b/internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp index 0d65e2ba1b..a80e300d9a 100644 --- a/internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp +++ b/internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp @@ -401,11 +401,9 @@ ChunkedSegmentSealedImpl::LoadColumnGroup( auto chunk_reader = std::move(chunk_reader_result).ValueOrDie(); - LOG_INFO( - "[StorageV2] segment {} loads manifest cg index {} with field ids " - "{} ", - this->get_segment_id(), - index); + LOG_INFO("[StorageV2] segment {} loads manifest cg index {}", + this->get_segment_id(), + index); auto translator = std::make_unique( diff --git a/internal/core/src/segcore/storagev2translator/ManifestGroupTranslator.cpp b/internal/core/src/segcore/storagev2translator/ManifestGroupTranslator.cpp index fce1e57f49..c9bf3aaf4b 100644 --- a/internal/core/src/segcore/storagev2translator/ManifestGroupTranslator.cpp +++ b/internal/core/src/segcore/storagev2translator/ManifestGroupTranslator.cpp @@ -165,6 +165,13 @@ ManifestGroupTranslator::get_cells( auto chunks = read_result.ValueOrDie(); for (size_t i = 0; i < chunks.size(); ++i) { auto& chunk = chunks[i]; + AssertInfo(chunk != nullptr, + "chunk is null, idx = {}, group index = {}, segment id = " + "{}, parallel degree = {}", + i, + column_group_index_, + segment_id_, + parallel_degree); auto cid = cids[i]; auto group_chunk = load_group_chunk(chunk, cid); cells.emplace_back(cid, std::move(group_chunk)); diff --git a/internal/core/thirdparty/milvus-storage/CMakeLists.txt b/internal/core/thirdparty/milvus-storage/CMakeLists.txt index f0c6873dc5..800c1fd5d9 100644 --- a/internal/core/thirdparty/milvus-storage/CMakeLists.txt +++ b/internal/core/thirdparty/milvus-storage/CMakeLists.txt @@ -14,7 +14,7 @@ # Update milvus-storage_VERSION for the first occurrence milvus_add_pkg_config("milvus-storage") set_property(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} PROPERTY INCLUDE_DIRECTORIES "") -set( milvus-storage_VERSION 302143c) +set( milvus-storage_VERSION ba7df7b) set( GIT_REPOSITORY "https://github.com/milvus-io/milvus-storage.git") message(STATUS "milvus-storage repo: ${GIT_REPOSITORY}") message(STATUS "milvus-storage version: ${milvus-storage_VERSION}") diff --git a/internal/datacoord/task_index.go b/internal/datacoord/task_index.go index ae03cca514..1812716b54 100644 --- a/internal/datacoord/task_index.go +++ b/internal/datacoord/task_index.go @@ -327,6 +327,7 @@ func (it *indexBuildTask) prepareJobRequest(ctx context.Context, segment *Segmen TaskSlot: it.taskSlot, LackBinlogRows: segIndex.NumRows - totalRows, InsertLogs: segment.GetBinlogs(), + Manifest: segment.GetManifestPath(), } WrapPluginContext(segment.GetCollectionID(), schema.GetProperties(), req) diff --git a/internal/datanode/index/task_index.go b/internal/datanode/index/task_index.go index cf78f61042..8e22f847d0 100644 --- a/internal/datanode/index/task_index.go +++ b/internal/datanode/index/task_index.go @@ -310,6 +310,7 @@ func (it *indexBuildTask) Execute(ctx context.Context) error { it.req.GetCollectionID(), it.req.GetPartitionID(), it.req.GetSegmentID()) + buildIndexParams.Manifest = it.req.GetManifest() } log.Info("create index", zap.Any("buildIndexParams", buildIndexParams)) diff --git a/internal/storage/record_reader.go b/internal/storage/record_reader.go index dd5525e0fe..a242773fe8 100644 --- a/internal/storage/record_reader.go +++ b/internal/storage/record_reader.go @@ -3,6 +3,7 @@ package storage import ( "fmt" "io" + "strconv" "github.com/apache/arrow/go/v17/arrow" "github.com/apache/arrow/go/v17/arrow/array" @@ -214,7 +215,12 @@ func NewManifestReader(manifest string, neededColumns := make([]string, 0, len(allFields)) for i, field := range allFields { field2Col[field.FieldID] = i - neededColumns = append(neededColumns, field.Name) + // Use field id here or external field + if field.ExternalField != "" { + neededColumns = append(neededColumns, field.ExternalField) + } else { + neededColumns = append(neededColumns, strconv.FormatInt(field.FieldID, 10)) + } } prr := &ManifestReader{ manifest: manifest, diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 5b622bf3e7..ddeb748e17 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -70,13 +70,17 @@ func ConvertToArrowSchema(schema *schemapb.CollectionSchema, useFieldID bool) (* func ConvertToArrowField(field *schemapb.FieldSchema, dataType arrow.DataType, useFieldID bool) arrow.Field { f := arrow.Field{ - Name: field.GetName(), Type: dataType, Metadata: arrow.NewMetadata([]string{packed.ArrowFieldIdMetadataKey}, []string{strconv.Itoa(int(field.GetFieldID()))}), Nullable: field.GetNullable(), } - if useFieldID { - field.Name = fmt.Sprintf("%d", field.GetFieldID()) + // external field name has higher priority + if field.GetExternalField() != "" { + f.Name = field.GetExternalField() + } else if useFieldID { // use fieldID as name when specified + f.Name = fmt.Sprintf("%d", field.GetFieldID()) + } else { + f.Name = field.GetName() } return f } diff --git a/internal/storagev2/packed/packed_writer_ffi.go b/internal/storagev2/packed/packed_writer_ffi.go index a9387159de..2da5c56a77 100644 --- a/internal/storagev2/packed/packed_writer_ffi.go +++ b/internal/storagev2/packed/packed_writer_ffi.go @@ -163,8 +163,6 @@ func (pw *FFIPackedWriter) Close() (string, error) { return "", err } - defer C.transaction_destroy(transationHandle) - var readVersion C.int64_t // TODO: not atomic, need to get version from transaction