enhance: enable Loon FFI by default and support storage v3 in file managers (#47984)

Related to #44956

Enable useLoonFFI config by default and extend DiskFileManagerImpl and
MemFileManagerImpl to handle STORAGE_V3 using the same code paths as
STORAGE_V2 for caching raw data, optional fields to disk and memory.

---------

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

configs/milvus.yaml

@@ -1027,7 +1027,7 @@ common:
       splitByAvgSize:
         enabled: true # enable split by average size policy in storage v2
         threshold: 1024 # split by average size policy threshold(in bytes) in storage v2
-    useLoonFFI: false
+    useLoonFFI: true
   traceLogMode: 0 # trace request info
   bloomFilterSize: 100000 # bloom filter initial size
   bloomFilterType: BlockedBloomFilter # bloom filter type, support BasicBloomFilter and BlockedBloomFilter

internal/compaction/common.go

@@ -21,6 +21,7 @@ import (
 	"io"
 
 	"github.com/apache/arrow/go/v17/arrow/array"
+	"github.com/cockroachdb/errors"
 	"go.uber.org/zap"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
@@ -109,8 +110,13 @@ func readDeltalogsV2(
 	manifestPath string,
 	option ...storage.RwOption,
 ) ([]storage.PrimaryKey, []typeutil.Timestamp, error) {
+	log := log.Ctx(ctx)
 	reader, err := storage.NewDeltalogReaderFromManifest(pkType, manifestPath, option...)
 	if err != nil {
+		if errors.Is(err, io.EOF) {
+			log.Info("new detalog reader returns EOF, no deltalog found")
+			return []storage.PrimaryKey{}, []typeutil.Timestamp{}, nil
+		}
 		return nil, nil, err
 	}
 	defer reader.Close()
@@ -120,7 +126,7 @@ func readDeltalogsV2(
 		return nil, nil, err
 	}
 
-	log.Ctx(ctx).Info("read V2 deltalogs from manifest", zap.Int("entries", len(pks)))
+	log.Info("read V2 deltalogs from manifest", zap.Int("entries", len(pks)))
 	return pks, tss, nil
 }
 

internal/compaction/common_test.go

@@ -18,7 +18,6 @@ package compaction
 
 import (
 	"context"
-	"io"
 	"path/filepath"
 	"testing"
 
@@ -508,6 +507,6 @@ func (s *CommonSuite) TestComposeDeleteFromDeltalogsV2() {
 
 		segment := &datapb.CompactionSegmentBinlogs{Manifest: manifestPath}
 		_, err := ComposeDeleteFromDeltalogs(ctx, schemapb.DataType_Int64, segment, options...)
-		s.ErrorIs(err, io.EOF)
+		s.NoError(err)
 	})
 }

internal/compaction/params_test.go

@@ -47,6 +47,7 @@ func TestGetJSONParams(t *testing.T) {
 		PreferSegmentSizeRatio:    paramtable.Get().DataCoordCfg.ClusteringCompactionPreferSegmentSizeRatio.GetAsFloat(),
 		BloomFilterApplyBatchSize: paramtable.Get().CommonCfg.BloomFilterApplyBatchSize.GetAsInt(),
 		StorageConfig:             CreateStorageConfig(),
+		UseLoonFFI:                paramtable.Get().CommonCfg.UseLoonFFI.GetAsBool(),
 	}, result)
 }
 
@@ -86,7 +87,8 @@ func TestGetParamsFromJSON_EmptyJSON(t *testing.T) {
 	result, err := ParseParamsFromJSON(emptyJSON)
 	assert.NoError(t, err)
 	assert.Equal(t, Params{
-		StorageVersion:            storage.StorageV2,
+		StorageVersion:            storage.StorageV3,
+		UseLoonFFI:                true,
 		BinLogMaxSize:             paramtable.Get().DataNodeCfg.BinLogMaxSize.GetAsUint64(),
 		UseMergeSort:              paramtable.Get().DataNodeCfg.UseMergeSort.GetAsBool(),
 		MaxSegmentMergeSort:       paramtable.Get().DataNodeCfg.MaxSegmentMergeSort.GetAsInt(),

internal/core/src/segcore/SegmentGrowingImpl.cpp

@@ -2010,7 +2010,7 @@ SegmentGrowingImpl::LoadColumnsGroups(std::string manifest_path) {
     auto column_groups = std::make_shared<milvus_storage::api::ColumnGroups>(
         loon_manifest->columnGroups());
 
-    auto arrow_schema = schema_->ConvertToArrowSchema();
+    auto arrow_schema = schema_->ConvertToLoonArrowSchema();
     reader_ = milvus_storage::api::Reader::create(
         column_groups, arrow_schema, nullptr, *properties);
 

internal/core/src/storage/DiskFileManagerImpl.cpp

@@ -506,7 +506,7 @@ DiskFileManagerImpl::CacheRawDataToDisk(const Config& config) {
     auto storage_version =
         index::GetValueFromConfig<int64_t>(config, STORAGE_VERSION_KEY)
             .value_or(0);
-    if (storage_version == STORAGE_V2) {
+    if (storage_version == STORAGE_V2 || storage_version == STORAGE_V3) {
         return cache_raw_data_to_disk_storage_v2<DataType>(config);
     }
     return cache_raw_data_to_disk_internal<DataType>(config);
@@ -1081,7 +1081,7 @@ DiskFileManagerImpl::CacheOptFieldToDisk(const Config& config) {
     }
 
     std::vector<std::vector<std::string>> remote_files_storage_v2;
-    if (storage_version == STORAGE_V2) {
+    if (storage_version == STORAGE_V2 || storage_version == STORAGE_V3) {
         auto segment_insert_files =
             index::GetValueFromConfig<std::vector<std::vector<std::string>>>(
                 config, SEGMENT_INSERT_FILES_KEY);
@@ -1123,7 +1123,7 @@ DiskFileManagerImpl::CacheOptFieldToDisk(const Config& config) {
 
         std::vector<FieldDataPtr> field_datas;
         // fetch scalar data from storage v2
-        if (storage_version == STORAGE_V2) {
+        if (storage_version == STORAGE_V2 || storage_version == STORAGE_V3) {
             field_datas = GetFieldDatasFromStorageV2(remote_files_storage_v2,
                                                      field_id,
                                                      field_type,

internal/core/src/storage/MemFileManagerImpl.cpp

@@ -334,7 +334,7 @@ MemFileManagerImpl::CacheOptFieldToMemory(const Config& config) {
     auto storage_version =
         index::GetValueFromConfig<int64_t>(config, STORAGE_VERSION_KEY)
             .value_or(0);
-    if (storage_version == STORAGE_V2) {
+    if (storage_version == STORAGE_V2 || storage_version == STORAGE_V3) {
         return cache_opt_field_memory_v2(config);
     }
     return cache_opt_field_memory(config);

internal/core/src/storage/Util.cpp

@@ -1588,8 +1588,8 @@ GetFieldDatasFromManifest(
             continue;
         }
 
-        auto chunked_array =
-            std::make_shared<arrow::ChunkedArray>(batch->column(0));
+        auto chunked_array = std::make_shared<arrow::ChunkedArray>(
+            batch->GetColumnByName(field_id_str));
         auto field_data = CreateFieldData(data_type.value(),
                                           element_type.value(),
                                           batch->schema()->field(0)->nullable(),

internal/datanode/compactor/clustering_compactor_test.go

@@ -64,6 +64,7 @@ func (s *ClusteringCompactionTaskSuite) SetupSuite() {
 
 func (s *ClusteringCompactionTaskSuite) setupTest() {
 	paramtable.Get().Save(paramtable.Get().CommonCfg.StorageType.Key, "local")
+	paramtable.Get().Save(paramtable.Get().CommonCfg.UseLoonFFI.Key, "false")
 
 	s.mockBinlogIO = mock_util.NewMockBinlogIO(s.T())
 
@@ -117,6 +118,7 @@ func (s *ClusteringCompactionTaskSuite) SetupSubTest() {
 
 func (s *ClusteringCompactionTaskSuite) TearDownTest() {
 	paramtable.Get().Reset(paramtable.Get().CommonCfg.StorageType.Key)
+	paramtable.Get().Reset(paramtable.Get().CommonCfg.UseLoonFFI.Key)
 }
 
 func (s *ClusteringCompactionTaskSuite) TestWrongCompactionType() {

internal/datanode/compactor/mix_compactor_storage_v2_test.go

@@ -56,17 +56,22 @@ type MixCompactionTaskStorageV2Suite struct {
 }
 
 func (s *MixCompactionTaskStorageV2Suite) SetupTest() {
-	s.setupTest()
-	paramtable.Get().Save("common.storageType", "local")
+	paramtable.Get().Save(paramtable.Get().CommonCfg.StorageType.Key, "local")
+	paramtable.Get().Save(paramtable.Get().CommonCfg.UseLoonFFI.Key, "false")
+	paramtable.Get().Save(paramtable.Get().LocalStorageCfg.Path.Key, s.T().TempDir())
 	initcore.InitStorageV2FileSystem(paramtable.Get())
+	s.setupTest()
 	s.task.compactionParams = compaction.GenParams()
 }
 
 func (s *MixCompactionTaskStorageV2Suite) TearDownTest() {
-	paramtable.Get().Reset("common.storageType")
+	paramtable.Get().Reset(paramtable.Get().CommonCfg.StorageType.Key)
+	paramtable.Get().Reset(paramtable.Get().CommonCfg.UseLoonFFI.Key)
+	paramtable.Get().Reset(paramtable.Get().LocalStorageCfg.Path.Key)
 	os.RemoveAll(paramtable.Get().LocalStorageCfg.Path.GetValue() + "insert_log")
 	os.RemoveAll(paramtable.Get().LocalStorageCfg.Path.GetValue() + "delta_log")
 	os.RemoveAll(paramtable.Get().LocalStorageCfg.Path.GetValue() + "stats_log")
+	initcore.CleanArrowFileSystemSingleton()
 }
 
 func (s *MixCompactionTaskStorageV2Suite) TestCompactDupPK() {

internal/datanode/compactor/sort_compaction.go

@@ -413,19 +413,21 @@ func (t *sortCompactionTask) Compact() (*datapb.CompactionPlanResult, error) {
 		return res, nil
 	}
 	stepStart = time.Now()
-	textStatsLogs, err := t.createTextIndex(ctx,
-		t.collectionID, t.partitionID, targetSegemntID, t.GetPlanID(),
-		res.GetSegments()[0].GetInsertLogs())
-	if err != nil {
-		log.Warn("failed to create text indexes", zap.Int64("targetSegmentID", targetSegemntID),
-			zap.Error(err))
-		return &datapb.CompactionPlanResult{
-			PlanID: t.GetPlanID(),
-			State:  datapb.CompactionTaskState_failed,
-		}, nil
+	for _, resultSegment := range res.GetSegments() {
+		textStatsLogs, err := t.createTextIndex(ctx,
+			t.collectionID, t.partitionID, targetSegemntID, t.GetPlanID(),
+			resultSegment)
+		if err != nil {
+			log.Warn("failed to create text indexes", zap.Int64("targetSegmentID", targetSegemntID),
+				zap.Error(err))
+			return &datapb.CompactionPlanResult{
+				PlanID: t.GetPlanID(),
+				State:  datapb.CompactionTaskState_failed,
+			}, nil
+		}
+		resultSegment.TextStatsLogs = textStatsLogs
 	}
 	createTextIndexCost := time.Since(stepStart)
-	res.Segments[0].TextStatsLogs = textStatsLogs
 
 	totalCost := time.Since(compactStart)
 	log.Info("compact done", zap.Int64("targetSegmentID", targetSegemntID),
@@ -478,7 +480,7 @@ func (t *sortCompactionTask) createTextIndex(ctx context.Context,
 	partitionID int64,
 	segmentID int64,
 	taskID int64,
-	insertBinlogs []*datapb.FieldBinlog,
+	segment *datapb.CompactionSegment,
 ) (map[int64]*datapb.TextIndexStats, error) {
 	log := log.Ctx(ctx).With(
 		zap.Int64("collectionID", collectionID),
@@ -486,7 +488,7 @@ func (t *sortCompactionTask) createTextIndex(ctx context.Context,
 		zap.Int64("segmentID", segmentID),
 	)
 
-	fieldBinlogs := lo.GroupBy(insertBinlogs, func(binlog *datapb.FieldBinlog) int64 {
+	fieldBinlogs := lo.GroupBy(segment.GetInsertLogs(), func(binlog *datapb.FieldBinlog) int64 {
 		return binlog.GetFieldID()
 	})
 
@@ -559,7 +561,7 @@ func (t *sortCompactionTask) createTextIndex(ctx context.Context,
 				StorageConfig:             newStorageConfig,
 				CurrentScalarIndexVersion: t.plan.GetCurrentScalarIndexVersion(),
 				StorageVersion:            t.storageVersion,
-				Manifest:                  t.manifest,
+				Manifest:                  segment.GetManifest(),
 			}
 
 			if len(analyzerExtraInfo) > 0 {
@@ -568,7 +570,7 @@ func (t *sortCompactionTask) createTextIndex(ctx context.Context,
 
 			if t.storageVersion == storage.StorageV2 || t.storageVersion == storage.StorageV3 {
 				buildIndexParams.SegmentInsertFiles = util.GetSegmentInsertFiles(
-					insertBinlogs,
+					segment.GetInsertLogs(),
 					t.compactionParams.StorageConfig,
 					collectionID,
 					partitionID,

internal/datanode/compactor/sort_compaction_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/milvus-io/milvus/internal/mocks"
 	"github.com/milvus-io/milvus/internal/mocks/flushcommon/mock_util"
 	"github.com/milvus-io/milvus/internal/storage"
+	"github.com/milvus-io/milvus/internal/util/initcore"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/etcdpb"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
@@ -93,10 +94,18 @@ func (s *SortCompactionTaskSuite) setupTest() {
 }
 
 func (s *SortCompactionTaskSuite) SetupTest() {
+	paramtable.Get().Save(paramtable.Get().CommonCfg.StorageType.Key, "local")
+	paramtable.Get().Save(paramtable.Get().CommonCfg.UseLoonFFI.Key, "false")
+	paramtable.Get().Save(paramtable.Get().LocalStorageCfg.Path.Key, s.T().TempDir())
+	initcore.InitStorageV2FileSystem(paramtable.Get())
 	s.setupTest()
 }
 
 func (s *SortCompactionTaskSuite) TearDownTest() {
+	paramtable.Get().Reset(paramtable.Get().CommonCfg.StorageType.Key)
+	paramtable.Get().Reset(paramtable.Get().CommonCfg.UseLoonFFI.Key)
+	paramtable.Get().Reset(paramtable.Get().LocalStorageCfg.Path.Key)
+	initcore.CleanArrowFileSystemSingleton()
 }
 
 func (s *SortCompactionTaskSuite) TestNewSortCompactionTask() {

internal/datanode/importv2/task.go

@@ -150,6 +150,7 @@ func UpdateSegmentInfo(info *datapb.ImportSegmentInfo) UpdateAction {
 				segmentsInfo[segment].Statslogs = mergeFn(segmentsInfo[segment].Statslogs, info.GetStatslogs())
 				segmentsInfo[segment].Deltalogs = mergeFn(segmentsInfo[segment].Deltalogs, info.GetDeltalogs())
 				segmentsInfo[segment].Bm25Logs = mergeFn(segmentsInfo[segment].Bm25Logs, info.GetBm25Logs())
+				segmentsInfo[segment].ManifestPath = info.GetManifestPath()
 				return
 			}
 			segmentsInfo[segment] = info

internal/util/initcore/test_util.go

@@ -0,0 +1,36 @@
+//go:build test
+// +build test
+
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package initcore
+
+/*
+#cgo pkg-config: milvus_core
+
+#include <stdlib.h>
+#include <stdint.h>
+#include <stdbool.h>
+#include "common/init_c.h"
+#include "segcore/arrow_fs_c.h"
+*/
+import "C"
+
+// CleanArrowFileSystemSingleton is an utility clean internal segcore fs singleton
+func CleanArrowFileSystemSingleton() {
+	C.CleanArrowFileSystemSingleton()
+}

pkg/util/paramtable/component_param.go

@@ -976,7 +976,7 @@ Large numeric passwords require double quotes to avoid yaml parsing precision is
 	p.UseLoonFFI = ParamItem{
 		Key:          "common.storage.useLoonFFI",
 		Version:      "2.6.7",
-		DefaultValue: "false",
+		DefaultValue: "true",
 		Export:       true,
 	}
 	p.UseLoonFFI.Init(base.mgr)

tests/restful_client_v2/testcases/test_jobs_operation.py

@@ -875,6 +875,7 @@ class TestCreateImportJob(TestBase):
     @pytest.mark.parametrize("enable_dynamic_schema", [True])
     @pytest.mark.parametrize("nb", [3000])
     @pytest.mark.parametrize("dim", [128])
+    @pytest.mark.skip(reason="default storage is v3, cannot use collection to create v2 import")
     def test_job_import_binlog_file_type(self, nb, dim, insert_round, auto_id,
                                                       is_partition_key, enable_dynamic_schema):
         """