enhance: Resolve issues integrating loon FFI (#45918)

Related to #44956

- Update milvus-storage version to ba7df7b for chunk reader fix
- Pass manifest path to index build request in DataCoord/DataNode
- Add null chunk assertion with detailed debug info in
ManifestGroupTranslator
- Fix memory corruption by removing premature transaction handle
destruction
- Clean up log message in ChunkedSegmentSealedImpl

---------

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

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<storagev2translator::ManifestGroupTranslator>(

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));

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}")

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)

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))
 

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,

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
 }

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