enhance: extract uncommitted non-log load changes

internal/core/src/exec/operator/MvccNode.cpp

@@ -74,6 +74,7 @@ PhyMvccNode::GetOutput() {
     }
 
     tracer::AddEvent(fmt::format("input_rows: {}", active_count_));
+    WaitPrefetch();
 
     // Visibility filtering disabled globally: skip all filtering.
     if (!segcore::SegcoreConfig::default_config()

internal/core/src/exec/operator/MvccNode.h

@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include <folly/futures/Future.h>
 #include <memory>
 #include <string>
 
@@ -70,7 +71,7 @@ class PhyMvccNode : public Operator {
     PrefetchAsync(const std::shared_ptr<folly::CPUThreadPoolExecutor>
                       prefetch_pool) override {
         auto self = std::static_pointer_cast<PhyMvccNode>(shared_from_this());
-        folly::via(prefetch_pool.get(), [self]() {
+        prefetch_future_ = folly::via(prefetch_pool.get(), [self]() {
             self->segment_->prefetch_chunks(
                 self->operator_context_->get_exec_context()
                     ->get_query_context()
@@ -79,14 +80,22 @@ class PhyMvccNode : public Operator {
         });
     }
 
+    void
+    WaitPrefetch() {
+        if (prefetch_future_.valid()) {
+            std::move(prefetch_future_).wait();
+        }
+    }
+
  private:
     const segcore::SegmentInternalInterface* segment_;
     milvus::Timestamp query_timestamp_;
     int64_t active_count_;
     bool is_finished_{false};
     bool is_source_node_{false};
     milvus::Timestamp collection_ttl_timestamp_;
+    folly::Future<folly::Unit> prefetch_future_;
 };
 
 }  // namespace exec
-}  // namespace milvus
+}  // namespace milvus

internal/core/src/storage/PrefetchThreadPool.h

@@ -6,7 +6,7 @@ inline std::shared_ptr<folly::CPUThreadPoolExecutor>
 GetPrefetchThreadPool() {
     static std::shared_ptr<folly::CPUThreadPoolExecutor> pool =
         std::make_shared<folly::CPUThreadPoolExecutor>(
-            64,
+            24,
             folly::CPUThreadPoolExecutor::makeDefaultPriorityQueue(1),
             std::make_shared<folly::NamedThreadFactory>("MILVUS_PREFETCH_"));
     return pool;

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 6f38add)
+set( milvus-storage_VERSION 0b35d53)
 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/optimistic_txn_persist.go

@@ -10,15 +10,15 @@ import (
 
 	"go.uber.org/zap"
 
-	"github.com/milvus-io/milvus/pkg/v2/log"
-	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
-	"github.com/milvus-io/milvus/pkg/v2/util/retry"
+	"github.com/milvus-io/milvus/pkg/v3/log"
+	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v3/util/retry"
 	"github.com/tikv/client-go/v2/txnkv"
 	"github.com/tikv/client-go/v2/txnkv/transaction"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"google.golang.org/protobuf/proto"
 
-	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
+	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
 )
 
 var (

internal/datacoord/segment_info_cache.go

@@ -5,10 +5,10 @@ import (
 
 	"go.uber.org/zap"
 
-	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
-	"github.com/milvus-io/milvus/pkg/v2/log"
-	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
-	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
+	"github.com/milvus-io/milvus-proto/go-api/v3/msgpb"
+	"github.com/milvus-io/milvus/pkg/v3/log"
+	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
+	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
 )
 
 // CachedSegmentsInfo is a concurrent, version-tracked segment metadata store
@@ -149,7 +149,6 @@ func (s *CachedSegmentsInfo) SetSegment(segmentID UniqueID, segment *SegmentInfo
 	return old, existed
 }
 
-
 // DropSegment marks the segment as a tombstone in the cache.
 // The version is retained to reject stale writes.
 func (s *CachedSegmentsInfo) DropSegment(segmentID UniqueID, version int64) {

internal/datacoord/write_through_cache.go

@@ -3,7 +3,7 @@ package datacoord
 import (
 	"sync/atomic"
 
-	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
+	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
 )
 
 // CacheUpdateFunc modifies a cached value in place.

internal/flushcommon/writebuffer/buffer_heap.go

@@ -3,7 +3,7 @@ package writebuffer
 import (
 	"container/heap"
 
-	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
+	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
 )
 
 // bufferHeapItem represents an item in the buffer heap

internal/util/cgo/pool.go

@@ -1,19 +1,17 @@
 package cgo
 
 import (
-	"math"
 	"runtime"
 	"time"
 
 	"github.com/milvus-io/milvus/pkg/v3/metrics"
-	"github.com/milvus-io/milvus/pkg/v3/util/hardware"
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
 var caller *cgoCaller
 
 func initCaller(nodeID string) {
-	chSize := int64(math.Ceil(float64(hardware.GetCPUNum()) * paramtable.Get().QueryNodeCfg.CGOPoolSizeRatio.GetAsFloat()))
+	chSize := paramtable.Get().QueryNodeCfg.MaxReadConcurrency.GetAsInt() * int(paramtable.Get().QueryNodeCfg.CGOPoolSizeRatio.GetAsFloat())
 	if chSize <= 0 {
 		chSize = 1
 	}