From 22defa8027b83060a12b9aadad2155ee5106de65 Mon Sep 17 00:00:00 2001 From: sunby Date: Wed, 3 Jun 2026 09:42:03 +0000 Subject: [PATCH] enhance: extract uncommitted non-log load changes Signed-off-by: sunby --- internal/core/src/exec/operator/MvccNode.cpp | 1 + internal/core/src/exec/operator/MvccNode.h | 13 +++++++++++-- internal/core/src/storage/PrefetchThreadPool.h | 2 +- .../core/thirdparty/milvus-storage/CMakeLists.txt | 2 +- internal/datacoord/optimistic_txn_persist.go | 8 ++++---- internal/datacoord/segment_info_cache.go | 9 ++++----- internal/datacoord/write_through_cache.go | 2 +- internal/flushcommon/writebuffer/buffer_heap.go | 2 +- internal/util/cgo/pool.go | 4 +--- 9 files changed, 25 insertions(+), 18 deletions(-) diff --git a/internal/core/src/exec/operator/MvccNode.cpp b/internal/core/src/exec/operator/MvccNode.cpp index f87f01eb75065..01199a87b2466 100644 --- a/internal/core/src/exec/operator/MvccNode.cpp +++ b/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() diff --git a/internal/core/src/exec/operator/MvccNode.h b/internal/core/src/exec/operator/MvccNode.h index 7a031d75c350e..ff59faf1b8a18 100644 --- a/internal/core/src/exec/operator/MvccNode.h +++ b/internal/core/src/exec/operator/MvccNode.h @@ -16,6 +16,7 @@ #pragma once +#include #include #include @@ -70,7 +71,7 @@ class PhyMvccNode : public Operator { PrefetchAsync(const std::shared_ptr prefetch_pool) override { auto self = std::static_pointer_cast(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,6 +80,13 @@ class PhyMvccNode : public Operator { }); } + void + WaitPrefetch() { + if (prefetch_future_.valid()) { + std::move(prefetch_future_).wait(); + } + } + private: const segcore::SegmentInternalInterface* segment_; milvus::Timestamp query_timestamp_; @@ -86,7 +94,8 @@ class PhyMvccNode : public Operator { bool is_finished_{false}; bool is_source_node_{false}; milvus::Timestamp collection_ttl_timestamp_; + folly::Future prefetch_future_; }; } // namespace exec -} // namespace milvus \ No newline at end of file +} // namespace milvus diff --git a/internal/core/src/storage/PrefetchThreadPool.h b/internal/core/src/storage/PrefetchThreadPool.h index c980cbf083ee4..97ef51583fed4 100644 --- a/internal/core/src/storage/PrefetchThreadPool.h +++ b/internal/core/src/storage/PrefetchThreadPool.h @@ -6,7 +6,7 @@ inline std::shared_ptr GetPrefetchThreadPool() { static std::shared_ptr pool = std::make_shared( - 64, + 24, folly::CPUThreadPoolExecutor::makeDefaultPriorityQueue(1), std::make_shared("MILVUS_PREFETCH_")); return pool; diff --git a/internal/core/thirdparty/milvus-storage/CMakeLists.txt b/internal/core/thirdparty/milvus-storage/CMakeLists.txt index 17d9266b1c5da..6dffe8699bf06 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 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}") diff --git a/internal/datacoord/optimistic_txn_persist.go b/internal/datacoord/optimistic_txn_persist.go index 31823f7acc27c..baedea5fc90c1 100644 --- a/internal/datacoord/optimistic_txn_persist.go +++ b/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 ( diff --git a/internal/datacoord/segment_info_cache.go b/internal/datacoord/segment_info_cache.go index 01ea0c259e855..b24392ca62dbe 100644 --- a/internal/datacoord/segment_info_cache.go +++ b/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) { diff --git a/internal/datacoord/write_through_cache.go b/internal/datacoord/write_through_cache.go index 6812a56cf55e7..d7dce52286081 100644 --- a/internal/datacoord/write_through_cache.go +++ b/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. diff --git a/internal/flushcommon/writebuffer/buffer_heap.go b/internal/flushcommon/writebuffer/buffer_heap.go index 65afd68c1da0c..d95795239396b 100644 --- a/internal/flushcommon/writebuffer/buffer_heap.go +++ b/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 diff --git a/internal/util/cgo/pool.go b/internal/util/cgo/pool.go index 7a7a07ded492c..aae72896e98ee 100644 --- a/internal/util/cgo/pool.go +++ b/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 }