From 0704ac563bfc5e5fcedc1e37824db55b7c6866f1 Mon Sep 17 00:00:00 2001 From: aoiasd Date: Wed, 15 Jul 2026 17:25:45 +0800 Subject: [PATCH] feat: support mmap for vortex local format Co-Authored-By: Claude Opus 4.6 Signed-off-by: aoiasd --- .../core/src/mmap/SparseVortexFileSystem.cpp | 549 +++++++++++++----- .../core/src/mmap/SparseVortexFileSystem.h | 16 + internal/core/src/mmap/VortexColumnGroup.cpp | 48 +- internal/core/src/mmap/VortexColumnGroup.h | 18 + internal/core/src/mmap/VortexColumnTest.cpp | 52 ++ .../src/segcore/ChunkedSegmentSealedImpl.cpp | 42 +- .../src/segcore/ChunkedSegmentSealedImpl.h | 1 + 7 files changed, 573 insertions(+), 153 deletions(-) diff --git a/internal/core/src/mmap/SparseVortexFileSystem.cpp b/internal/core/src/mmap/SparseVortexFileSystem.cpp index 890647d17a..fda80b0a98 100644 --- a/internal/core/src/mmap/SparseVortexFileSystem.cpp +++ b/internal/core/src/mmap/SparseVortexFileSystem.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include @@ -85,39 +86,104 @@ CreateAnonymousSparseFile(std::string_view name) { #endif } -} // namespace - -class SparseVortexMmapFile final - : public milvus_storage::vortex::VortexRangeFile, - public std::enable_shared_from_this { - public: - explicit SparseVortexMmapFile(std::string name) : name_(std::move(name)) { - fd_ = CreateAnonymousSparseFile(name_); - if (fd_ < 0) { - ThrowSystemError(ErrorCode::FileCreateFailed, - "create vortex sparse file"); +int +OpenSparseFile(const std::string& path) { + const auto parent = std::filesystem::path(path).parent_path(); + if (!parent.empty()) { + std::error_code error; + std::filesystem::create_directories(parent, error); + if (error) { + ThrowInfo(ErrorCode::FileCreateFailed, + "create vortex sparse directory {} failed: {}", + parent.string(), + error.message()); } } - ~SparseVortexMmapFile() override { - if (mapping_ != nullptr) { - if (::munmap(mapping_, static_cast(mapped_size_)) != 0) { - const auto error = errno; - LOG_WARN( - "failed to unmap vortex sparse file {} during " - "cleanup: {}", - name_, - std::strerror(error)); + const auto fd = + ::open(path.c_str(), O_CREAT | O_RDWR | O_TRUNC | O_CLOEXEC, 0600); + if (fd < 0) { + ThrowSystemError(ErrorCode::FileCreateFailed, + "open vortex sparse file"); + } + return fd; +} + +arrow::Status +PWriteAll(int fd, const uint8_t* data, uint64_t offset, uint64_t length) { + uint64_t written = 0; + while (written < length) { + const auto remaining = length - written; + const auto chunk = std::min( + remaining, + static_cast(std::numeric_limits::max())); + const auto result = ::pwrite(fd, + data + written, + static_cast(chunk), + static_cast(offset + written)); + if (result < 0) { + if (errno == EINTR) { + continue; } + return arrow::Status::IOError(fmt::format( + "write vortex sparse file failed: {}", std::strerror(errno))); } - if (fd_ >= 0) { - if (::close(fd_) != 0) { - const auto error = errno; + if (result == 0) { + return arrow::Status::IOError( + "write vortex sparse file made no progress"); + } + written += static_cast(result); + } + return arrow::Status::OK(); +} + +arrow::Result +PReadAll(int fd, uint8_t* out, uint64_t offset, uint64_t length) { + uint64_t read = 0; + while (read < length) { + const auto remaining = length - read; + const auto chunk = std::min( + remaining, + static_cast(std::numeric_limits::max())); + const auto result = ::pread(fd, + out + read, + static_cast(chunk), + static_cast(offset + read)); + if (result < 0) { + if (errno == EINTR) { + continue; + } + return arrow::Status::IOError(fmt::format( + "read vortex sparse file failed: {}", std::strerror(errno))); + } + if (result == 0) { + break; + } + read += static_cast(result); + } + return static_cast(read); +} + +} // namespace + +class SparseVortexRangeFileBase + : public milvus_storage::vortex::VortexRangeFile, + public std::enable_shared_from_this { + public: + ~SparseVortexRangeFileBase() override { + if (fd_ >= 0 && ::close(fd_) != 0) { + LOG_WARN("failed to close vortex sparse file {} during cleanup: {}", + name_, + std::strerror(errno)); + } + if (!file_path_.empty()) { + std::error_code error; + std::filesystem::remove(file_path_, error); + if (error) { LOG_WARN( - "failed to close vortex sparse file {} during " - "cleanup: {}", - name_, - std::strerror(error)); + "failed to remove vortex sparse file {} during cleanup: {}", + file_path_, + error.message()); } } } @@ -138,7 +204,7 @@ class SparseVortexMmapFile final "resize vortex sparse file"); } size_ = size; - RemapLocked(); + OnResizeLocked(); } uint64_t @@ -172,13 +238,7 @@ class SparseVortexMmapFile final length, size_)); } - if (mapping_ == nullptr) { - return arrow::Status::IOError( - "VortexRangeFile::WriteAt got empty mapping"); - } - - std::memcpy(mapping_ + offset, data->data(), length); - return arrow::Status::OK(); + return WriteAtLocked(offset, *data); } void @@ -191,96 +251,34 @@ class SparseVortexMmapFile final return; } length = std::min(length, size_ - offset); - if (length == 0 || mapping_ == nullptr) { + if (length == 0) { return; } -#if defined(FALLOC_FL_PUNCH_HOLE) && defined(SYS_fallocate) int fallocate_error = ENOTSUP; +#if defined(FALLOC_FL_PUNCH_HOLE) && defined(SYS_fallocate) if (offset <= static_cast(std::numeric_limits::max()) && length <= static_cast(std::numeric_limits::max())) { - if (::syscall(SYS_fallocate, - fd_, - FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE, - static_cast(offset), - static_cast(length)) == 0) { - return; - } - fallocate_error = errno; - } -#else - constexpr int fallocate_error = ENOTSUP; -#endif - - const auto page_size = ::sysconf(_SC_PAGESIZE); - if (page_size > 0) { - const auto alignment = static_cast(page_size); - const auto end = offset + length; - const auto aligned_begin = AlignUp(offset, alignment); - const auto aligned_end = AlignDown(end, alignment); - if (aligned_begin < aligned_end) { - const auto prefix_end = std::min(aligned_begin, end); - if (offset < prefix_end) { - std::memset(mapping_ + offset, 0, prefix_end - offset); - } - if (aligned_end < end) { - std::memset(mapping_ + aligned_end, 0, end - aligned_end); - } -#ifdef MADV_REMOVE - if (::madvise(mapping_ + aligned_begin, - static_cast(aligned_end - aligned_begin), - MADV_REMOVE) == 0) { - LOG_WARN( - "punching vortex cell [{}, {}) in {} with " - "fallocate failed: {}; released aligned pages " - "with MADV_REMOVE", - offset, - end, - name_, - std::strerror(fallocate_error)); + while (true) { + if (::syscall(SYS_fallocate, + fd_, + FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE, + static_cast(offset), + static_cast(length)) == 0) { return; } - const auto madvise_error = errno; -#else - constexpr int madvise_error = ENOTSUP; -#endif - std::memset( - mapping_ + aligned_begin, 0, aligned_end - aligned_begin); -#ifdef MADV_DONTNEED - if (::madvise(mapping_ + aligned_begin, - static_cast(aligned_end - aligned_begin), - MADV_DONTNEED) != 0) { - LOG_WARN( - "failed to discard zeroed vortex cell pages in " - "{}: {}", - name_, - std::strerror(errno)); + fallocate_error = errno; + if (fallocate_error != EINTR) { + break; } -#endif - LOG_WARN( - "failed to release vortex cell [{}, {}) in {} with " - "fallocate ({}) and MADV_REMOVE ({}); zeroed the " - "range to preserve sparse-file read semantics", - offset, - end, - name_, - std::strerror(fallocate_error), - std::strerror(madvise_error)); - return; } + } else { + fallocate_error = EOVERFLOW; } - - std::memset(mapping_ + offset, 0, length); - LOG_WARN( - "failed to release vortex cell [{}, {}) in {} with " - "fallocate: {}; zeroed the range because it contains no " - "aligned full page", - offset, - offset + length, - name_, - std::strerror(fallocate_error)); +#endif + ClearRangeLocked(offset, length, fallocate_error); } arrow::Result @@ -302,29 +300,18 @@ class SparseVortexMmapFile final if (pos >= size_) { return 0; } - const auto available = size_ - pos; - const auto read = std::min(static_cast(nbytes), available); + const auto read = std::min(static_cast(nbytes), size_ - pos); if (read == 0) { return 0; } - if (mapping_ == nullptr) { - return arrow::Status::IOError( - "VortexRangeFile::ReadAt got empty mapping"); - } - - std::memcpy(out, mapping_ + pos, read); - return static_cast(read); + return ReadAtLocked(pos, read, out); } arrow::Result> ReadAt(int64_t position, int64_t nbytes) const override { - if (nbytes < 0) { + if (position < 0 || nbytes < 0) { return arrow::Status::Invalid( - "VortexRangeFile::ReadAt got negative length"); - } - if (position < 0) { - return arrow::Status::Invalid( - "VortexRangeFile::ReadAt got negative position"); + "VortexRangeFile::ReadAt got negative position or length"); } if (nbytes == 0) { return arrow::Buffer::FromString(std::string()); @@ -335,24 +322,196 @@ class SparseVortexMmapFile final if (pos >= size_) { return arrow::Buffer::FromString(std::string()); } - const auto available = size_ - pos; - const auto read = std::min(static_cast(nbytes), available); + const auto read = std::min(static_cast(nbytes), size_ - pos); if (read == 0) { return arrow::Buffer::FromString(std::string()); } + return ReadAtLocked(pos, read); + } + + protected: + SparseVortexRangeFileBase(std::string name, int fd, std::string file_path) + : name_(std::move(name)), file_path_(std::move(file_path)), fd_(fd) { + } + + virtual void + OnResizeLocked() = 0; + + virtual arrow::Status + WriteAtLocked(uint64_t offset, const arrow::Buffer& data) = 0; + + virtual void + ClearRangeLocked(uint64_t offset, uint64_t length, int fallocate_error) = 0; + + virtual arrow::Result + ReadAtLocked(uint64_t position, uint64_t nbytes, void* out) const = 0; + + virtual arrow::Result> + ReadAtLocked(uint64_t position, uint64_t nbytes) const = 0; + + std::string name_; + std::string file_path_; + int fd_ = -1; + uint64_t size_ = 0; + + private: + mutable std::shared_mutex mutex_; +}; + +class SparseVortexMmapRangeFile final : public SparseVortexRangeFileBase { + public: + SparseVortexMmapRangeFile(std::string name, + int fd, + std::string file_path, + bool mmap_populate) + : SparseVortexRangeFileBase(std::move(name), fd, std::move(file_path)), + mmap_populate_(mmap_populate) { + } + + ~SparseVortexMmapRangeFile() override { + if (mapping_ != nullptr && + ::munmap(mapping_, static_cast(mapped_size_)) != 0) { + LOG_WARN("failed to unmap vortex sparse file {} during cleanup: {}", + name_, + std::strerror(errno)); + } + } + + private: + void + OnResizeLocked() override { + RemapLocked(); + } + + arrow::Status + WriteAtLocked(uint64_t offset, const arrow::Buffer& data) override { + if (mapping_ == nullptr) { + return arrow::Status::IOError( + "VortexRangeFile::WriteAt got empty mapping"); + } + std::memcpy( + mapping_ + offset, data.data(), static_cast(data.size())); + PopulateWrittenRangeLocked(offset, static_cast(data.size())); + return arrow::Status::OK(); + } + + void + ClearRangeLocked(uint64_t offset, + uint64_t length, + int fallocate_error) override { + if (mapping_ == nullptr) { + return; + } + + const auto page_size = ::sysconf(_SC_PAGESIZE); + if (page_size > 0) { + const auto alignment = static_cast(page_size); + const auto end = offset + length; + const auto aligned_begin = AlignUp(offset, alignment); + const auto aligned_end = AlignDown(end, alignment); + if (aligned_begin < aligned_end) { + const auto prefix_end = std::min(aligned_begin, end); + if (offset < prefix_end) { + std::memset(mapping_ + offset, 0, prefix_end - offset); + } + if (aligned_end < end) { + std::memset(mapping_ + aligned_end, 0, end - aligned_end); + } +#ifdef MADV_REMOVE + if (::madvise(mapping_ + aligned_begin, + static_cast(aligned_end - aligned_begin), + MADV_REMOVE) == 0) { + LOG_WARN( + "punching vortex cell [{}, {}) in {} with fallocate " + "failed: {}; released aligned pages with MADV_REMOVE", + offset, + end, + name_, + std::strerror(fallocate_error)); + return; + } + const auto madvise_error = errno; +#else + constexpr int madvise_error = ENOTSUP; +#endif + std::memset( + mapping_ + aligned_begin, 0, aligned_end - aligned_begin); +#ifdef MADV_DONTNEED + if (::madvise(mapping_ + aligned_begin, + static_cast(aligned_end - aligned_begin), + MADV_DONTNEED) != 0) { + LOG_WARN( + "failed to discard zeroed vortex cell pages in {}: {}", + name_, + std::strerror(errno)); + } +#endif + LOG_WARN( + "failed to release vortex cell [{}, {}) in {} with " + "fallocate ({}) and MADV_REMOVE ({}); zeroed the range", + offset, + end, + name_, + std::strerror(fallocate_error), + std::strerror(madvise_error)); + return; + } + } + + std::memset(mapping_ + offset, 0, length); + LOG_WARN( + "failed to release vortex cell [{}, {}) in {} with fallocate: " + "{}; zeroed the range", + offset, + offset + length, + name_, + std::strerror(fallocate_error)); + } + + arrow::Result + ReadAtLocked(uint64_t position, uint64_t nbytes, void* out) const override { + if (mapping_ == nullptr) { + return arrow::Status::IOError( + "VortexRangeFile::ReadAt got empty mapping"); + } + std::memcpy(out, mapping_ + position, nbytes); + return static_cast(nbytes); + } + + arrow::Result> + ReadAtLocked(uint64_t position, uint64_t nbytes) const override { if (mapping_ == nullptr) { return arrow::Status::IOError( "VortexRangeFile::ReadAt got empty mapping"); } auto owner = shared_from_this(); return std::shared_ptr( - new arrow::Buffer(mapping_ + pos, static_cast(read)), + new arrow::Buffer(mapping_ + position, + static_cast(nbytes)), [owner = std::move(owner)](arrow::Buffer* buffer) { delete buffer; }); } - private: + void + PopulateWrittenRangeLocked(uint64_t offset, uint64_t length) const { + if (!mmap_populate_ || mapping_ == nullptr || length == 0) { + return; + } + const auto page_size_value = ::sysconf(_SC_PAGESIZE); + const auto page_size = page_size_value > 0 + ? static_cast(page_size_value) + : static_cast(4096); + const auto end = std::min(offset + length, mapped_size_); + for (auto pos = AlignDown(offset, page_size); pos < end; + pos += page_size) { + const volatile auto value = mapping_[pos]; + (void)value; + } + const volatile auto value = mapping_[end - 1]; + (void)value; + } + void RemapLocked() { if (mapping_ != nullptr) { @@ -384,14 +543,117 @@ class SparseVortexMmapFile final mapped_size_ = size_; } - mutable std::shared_mutex mutex_; - std::string name_; - int fd_ = -1; + bool mmap_populate_ = false; uint8_t* mapping_ = nullptr; uint64_t mapped_size_ = 0; - uint64_t size_ = 0; }; +class SparseVortexDiskRangeFile final : public SparseVortexRangeFileBase { + public: + SparseVortexDiskRangeFile(std::string name, int fd, std::string file_path) + : SparseVortexRangeFileBase(std::move(name), fd, std::move(file_path)) { + } + + private: + void + OnResizeLocked() override { + } + + arrow::Status + WriteAtLocked(uint64_t offset, const arrow::Buffer& data) override { + return PWriteAll( + fd_, data.data(), offset, static_cast(data.size())); + } + + void + ClearRangeLocked(uint64_t offset, + uint64_t length, + int fallocate_error) override { + constexpr uint64_t kZeroBufferSize = 64 * 1024; + const std::vector zeros( + static_cast(std::min(length, kZeroBufferSize)), 0); + uint64_t cleared = 0; + while (cleared < length) { + const auto chunk = + std::min(zeros.size(), length - cleared); + const auto status = + PWriteAll(fd_, zeros.data(), offset + cleared, chunk); + if (!status.ok()) { + LOG_WARN( + "failed to clear vortex disk cell [{}, {}) in {} after " + "fallocate failed ({}): {}", + offset, + offset + length, + name_, + std::strerror(fallocate_error), + status.ToString()); + return; + } + cleared += chunk; + } + LOG_WARN( + "failed to release vortex disk cell [{}, {}) in {} with " + "fallocate: {}; zeroed the range", + offset, + offset + length, + name_, + std::strerror(fallocate_error)); + } + + arrow::Result + ReadAtLocked(uint64_t position, uint64_t nbytes, void* out) const override { + return PReadAll(fd_, static_cast(out), position, nbytes); + } + + arrow::Result> + ReadAtLocked(uint64_t position, uint64_t nbytes) const override { + ARROW_ASSIGN_OR_RAISE( + auto buffer, arrow::AllocateBuffer(static_cast(nbytes))); + ARROW_ASSIGN_OR_RAISE( + auto bytes_read, + PReadAll(fd_, buffer->mutable_data(), position, nbytes)); + std::shared_ptr shared_buffer(std::move(buffer)); + if (bytes_read == shared_buffer->size()) { + return shared_buffer; + } + return arrow::SliceBuffer(shared_buffer, 0, bytes_read); + } +}; + +std::shared_ptr +MakeSparseVortexRangeFile(std::string name, + SparseVortexFileSystemOptions options) { + switch (options.backing) { + case SparseVortexFileBacking::Memory: { + const auto fd = CreateAnonymousSparseFile(name); + if (fd < 0) { + ThrowSystemError(ErrorCode::FileCreateFailed, + "create vortex sparse file"); + } + return std::make_shared( + std::move(name), fd, std::string(), options.mmap_populate); + } + case SparseVortexFileBacking::Mmap: { + auto file_path = + options.file_path.empty() ? name : options.file_path; + const auto fd = OpenSparseFile(file_path); + return std::make_shared( + std::move(name), + fd, + std::move(file_path), + options.mmap_populate); + } + case SparseVortexFileBacking::Disk: { + auto file_path = + options.file_path.empty() ? name : options.file_path; + const auto fd = OpenSparseFile(file_path); + return std::make_shared( + std::move(name), fd, std::move(file_path)); + } + } + ThrowInfo(ErrorCode::UnexpectedError, "unknown vortex sparse file backing"); +} + class SparseVortexInputFile final : public arrow::io::RandomAccessFile { public: explicit SparseVortexInputFile( @@ -470,14 +732,15 @@ class SparseVortexFileSystem final : public arrow::fs::FileSystem, public milvus_storage::vortex::VortexRangeFileProvider { public: - explicit SparseVortexFileSystem(std::string path) - : file_(std::make_shared(path)), + SparseVortexFileSystem(std::string path, + SparseVortexFileSystemOptions options) + : file_(MakeSparseVortexRangeFile(path, std::move(options))), path_(std::move(path)) { } std::string type_name() const override { - return "vortex-sparse-mmap"; + return "vortex-sparse"; } bool @@ -593,7 +856,15 @@ MakeSparseVortexPath(std::string_view source_path) { std::shared_ptr MakeSparseVortexFileSystem(std::string path) { - return std::make_shared(std::move(path)); + return MakeSparseVortexFileSystem(std::move(path), + SparseVortexFileSystemOptions{}); +} + +std::shared_ptr +MakeSparseVortexFileSystem(std::string path, + SparseVortexFileSystemOptions options) { + return std::make_shared(std::move(path), + std::move(options)); } } // namespace milvus diff --git a/internal/core/src/mmap/SparseVortexFileSystem.h b/internal/core/src/mmap/SparseVortexFileSystem.h index 100c08e4fa..92082aeb79 100644 --- a/internal/core/src/mmap/SparseVortexFileSystem.h +++ b/internal/core/src/mmap/SparseVortexFileSystem.h @@ -25,10 +25,26 @@ class FileSystem; namespace milvus { +enum class SparseVortexFileBacking { + Memory, + Mmap, + Disk, +}; + +struct SparseVortexFileSystemOptions { + SparseVortexFileBacking backing = SparseVortexFileBacking::Memory; + std::string file_path; + bool mmap_populate = false; +}; + std::string MakeSparseVortexPath(std::string_view source_path); std::shared_ptr MakeSparseVortexFileSystem(std::string path); +std::shared_ptr +MakeSparseVortexFileSystem(std::string path, + SparseVortexFileSystemOptions options); + } // namespace milvus diff --git a/internal/core/src/mmap/VortexColumnGroup.cpp b/internal/core/src/mmap/VortexColumnGroup.cpp index e96f336c2d..31adb745c9 100644 --- a/internal/core/src/mmap/VortexColumnGroup.cpp +++ b/internal/core/src/mmap/VortexColumnGroup.cpp @@ -16,6 +16,8 @@ #include "mmap/VortexColumnGroup.h" +#include +#include #include #include #include @@ -37,6 +39,24 @@ namespace milvus { namespace { +std::atomic g_vortex_sparse_path_generation{0}; + +std::string +MakeFileBackedSparsePath(const VortexColumnGroup::Options& options, + size_t file_index) { + AssertInfo(!options.mmap_dir_path.empty(), + "vortex file-backed sparse file requires mmap dir path"); + const auto generation = + g_vortex_sparse_path_generation.fetch_add(1, std::memory_order_relaxed); + return (std::filesystem::path(options.mmap_dir_path) / "vortex" / + fmt::format("seg_{}_cg_{}_file_{}_{}.vortex", + options.segment_id, + options.column_group_index, + file_index, + generation)) + .string(); +} + [[noreturn]] void ThrowVortexStatus(const arrow::Status& status, ErrorCode fallback_code, @@ -60,6 +80,21 @@ VortexColumnGroup::VortexColumnGroup( const std::vector& field_names, CacheWarmupPolicy cache_warmup_policy, milvus::OpContext* op_ctx) + : VortexColumnGroup(files, + std::move(properties), + field_names, + cache_warmup_policy, + op_ctx, + Options{}) { +} + +VortexColumnGroup::VortexColumnGroup( + const std::vector& files, + std::shared_ptr properties, + const std::vector& field_names, + CacheWarmupPolicy cache_warmup_policy, + milvus::OpContext* op_ctx, + Options options) : num_fields_(field_names.size()) { AssertInfo(properties != nullptr, "vortex properties is null"); AssertInfo(!files.empty(), "vortex column group has no files"); @@ -69,7 +104,8 @@ VortexColumnGroup::VortexColumnGroup( num_rows_until_chunk_.push_back(0); int64_t row_prefix = 0; - for (const auto& file : files) { + for (size_t file_index = 0; file_index < files.size(); ++file_index) { + const auto& file = files[file_index]; auto fs_result = milvus_storage::FilesystemCache::getInstance().get( *properties, file.path); if (!fs_result.ok()) { @@ -92,7 +128,15 @@ VortexColumnGroup::VortexColumnGroup( state.resolved_path = uri_result.ValueOrDie().key; state.source_fs = fs_result.ValueOrDie(); state.sparse_path = MakeSparseVortexPath(state.resolved_path); - state.sparse_fs = MakeSparseVortexFileSystem(state.sparse_path); + SparseVortexFileSystemOptions sparse_options; + sparse_options.backing = options.sparse_file_backing; + sparse_options.mmap_populate = options.mmap_populate; + if (options.sparse_file_backing != SparseVortexFileBacking::Memory) { + sparse_options.file_path = + MakeFileBackedSparsePath(options, file_index); + } + state.sparse_fs = MakeSparseVortexFileSystem(state.sparse_path, + std::move(sparse_options)); state.footer_reader = std::make_shared( state.sparse_fs, diff --git a/internal/core/src/mmap/VortexColumnGroup.h b/internal/core/src/mmap/VortexColumnGroup.h index 7effbfc562..777aba79a5 100644 --- a/internal/core/src/mmap/VortexColumnGroup.h +++ b/internal/core/src/mmap/VortexColumnGroup.h @@ -22,6 +22,7 @@ #include #include "common/Types.h" +#include "mmap/SparseVortexFileSystem.h" #include "milvus-storage/properties.h" namespace arrow::fs { @@ -55,6 +56,15 @@ class VortexColumnGroup { public: using FileInfo = VortexColumnFileInfo; + struct Options { + SparseVortexFileBacking sparse_file_backing = + SparseVortexFileBacking::Memory; + bool mmap_populate = false; + std::string mmap_dir_path; + int64_t segment_id = 0; + int64_t column_group_index = 0; + }; + struct FileState { std::string path; std::string resolved_path; @@ -78,6 +88,14 @@ class VortexColumnGroup { CacheWarmupPolicy cache_warmup_policy, milvus::OpContext* op_ctx); + VortexColumnGroup( + const std::vector& files, + std::shared_ptr properties, + const std::vector& field_names, + CacheWarmupPolicy cache_warmup_policy, + milvus::OpContext* op_ctx, + Options options); + ~VortexColumnGroup(); void diff --git a/internal/core/src/mmap/VortexColumnTest.cpp b/internal/core/src/mmap/VortexColumnTest.cpp index 1816166433..59892b484c 100644 --- a/internal/core/src/mmap/VortexColumnTest.cpp +++ b/internal/core/src/mmap/VortexColumnTest.cpp @@ -16,6 +16,7 @@ #include "mmap/VortexColumn.h" +#include #include #include #include @@ -31,8 +32,10 @@ #include "common/FieldMeta.h" #include "gtest/gtest.h" #include "milvus-storage/column_groups.h" +#include "milvus-storage/format/vortex/vortex_types.h" #include "milvus-storage/format/vortex/vortex_writer.h" #include "milvus-storage/properties.h" +#include "mmap/SparseVortexFileSystem.h" namespace milvus { namespace { @@ -78,6 +81,50 @@ MakeProperties() { return properties; } +void +CheckSparseVortexFileBacking(SparseVortexFileBacking backing) { + const auto dir = + std::filesystem::temp_directory_path() / + ("milvus_vortex_sparse_fs_test_" + std::to_string(::getpid()) + "_" + + std::to_string(static_cast(backing))); + std::filesystem::create_directories(dir); + const auto file_path = dir / "sparse.vx"; + const std::string logical_path = "sparse-test.vx"; + + SparseVortexFileSystemOptions options; + options.backing = backing; + options.file_path = file_path.string(); + options.mmap_populate = true; + auto fs = MakeSparseVortexFileSystem(logical_path, std::move(options)); + auto range_result = + milvus_storage::vortex::GetVortexRangeFile(fs, logical_path); + ASSERT_TRUE(range_result.ok()) << range_result.status().ToString(); + auto range_file = std::move(range_result).ValueOrDie(); + + range_file->Resize(4096); + ASSERT_TRUE(std::filesystem::exists(file_path)); + auto status = range_file->WriteAt(128, arrow::Buffer::FromString("abc")); + ASSERT_TRUE(status.ok()) << status.ToString(); + + std::array out{}; + auto read_result = range_file->ReadAt(128, out.size(), out.data()); + ASSERT_TRUE(read_result.ok()) << read_result.status().ToString(); + ASSERT_EQ(read_result.ValueOrDie(), static_cast(out.size())); + ASSERT_EQ(std::string(out.data(), out.size()), "abc"); + + range_file->Punch(128, out.size()); + out.fill('\1'); + read_result = range_file->ReadAt(128, out.size(), out.data()); + ASSERT_TRUE(read_result.ok()) << read_result.status().ToString(); + ASSERT_EQ(read_result.ValueOrDie(), static_cast(out.size())); + ASSERT_EQ(std::string(out.data(), out.size()), std::string(out.size(), 0)); + + range_file.reset(); + fs.reset(); + ASSERT_FALSE(std::filesystem::exists(file_path)); + std::filesystem::remove_all(dir); +} + VortexColumn::FileInfo WriteVortexFile(const std::string& path, const std::shared_ptr& schema, @@ -760,6 +807,11 @@ CheckDataScan(VortexColumn& column, DataType type) { } // namespace +TEST(VortexColumnTest, SparseFileSystemFileBackedModes) { + CheckSparseVortexFileBacking(SparseVortexFileBacking::Mmap); + CheckSparseVortexFileBacking(SparseVortexFileBacking::Disk); +} + TEST(VortexColumnTest, ScanAndTake) { auto schema = MakeSchema(); auto properties = diff --git a/internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp b/internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp index d5f1e7b831..95dc0afdfe 100644 --- a/internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp +++ b/internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp @@ -7710,6 +7710,7 @@ ChunkedSegmentSealedImpl::TryLoadVortexColumnGroup( bool eager_load, bool has_warmup_setting, const std::string& aggregated_warmup_policy, + bool use_mmap, milvus::OpContext* op_ctx, bool is_replace, RuntimeResourceState* runtime, @@ -7786,12 +7787,25 @@ ChunkedSegmentSealedImpl::TryLoadVortexColumnGroup( /*is_vector=*/false, /*is_index=*/false, /*in_load_list=*/eager_load); + VortexColumnGroup::Options vortex_options; + if (use_mmap) { + auto& mmap_config = storage::MmapManager::GetInstance().GetMmapConfig(); + vortex_options.sparse_file_backing = SparseVortexFileBacking::Mmap; + vortex_options.mmap_populate = mmap_config.GetMmapPopulate(); + vortex_options.mmap_dir_path = + milvus::storage::LocalChunkManagerSingleton::GetInstance() + .GetChunkManager() + ->GetRootPath(); + vortex_options.segment_id = get_segment_id(); + vortex_options.column_group_index = index; + } auto vortex_column_group = std::make_shared(vortex_files, properties, vortex_field_names, group_cache_warmup_policy, - op_ctx); + op_ctx, + std::move(vortex_options)); const auto vortex_group_memory_size = vortex_column_group->memory_size(); const auto vortex_group_field_count = local_vortex_field_ids.size(); @@ -7816,7 +7830,7 @@ ChunkedSegmentSealedImpl::TryLoadVortexColumnGroup( column, column->NumRows(), field_meta.get_data_type(), - false, + use_mmap, true, segment_load_info, schema_snapshot, @@ -7900,6 +7914,12 @@ ChunkedSegmentSealedImpl::LoadColumnGroup( } } + auto& mmap_config = storage::MmapManager::GetInstance().GetMmapConfig(); + const bool global_use_mmap = is_vector + ? mmap_config.GetVectorFieldEnableMmap() + : mmap_config.GetScalarFieldEnableMmap(); + const bool use_mmap = has_mmap_setting ? mmap_enabled : global_use_mmap; + if (TryLoadVortexColumnGroup(column_group, properties, index, @@ -7910,6 +7930,7 @@ ChunkedSegmentSealedImpl::LoadColumnGroup( eager_load, has_warmup_setting, aggregated_warmup_policy, + use_mmap, op_ctx, is_replace, runtime, @@ -7917,11 +7938,6 @@ ChunkedSegmentSealedImpl::LoadColumnGroup( return; } - auto& mmap_config = storage::MmapManager::GetInstance().GetMmapConfig(); - bool global_use_mmap = is_vector ? mmap_config.GetVectorFieldEnableMmap() - : mmap_config.GetScalarFieldEnableMmap(); - auto use_mmap = has_mmap_setting ? mmap_enabled : global_use_mmap; - // The set of columns this entry projects is exactly the field_ids the // diff handed us. For lazy entries, SegmentLoadInfo::ComputeDiffColumnGroups // emits one entry per field, so each lazy entry produces a single-column @@ -8082,6 +8098,12 @@ ChunkedSegmentSealedImpl::LoadColumnGroup( } } + auto& mmap_config = storage::MmapManager::GetInstance().GetMmapConfig(); + const bool global_use_mmap = is_vector + ? mmap_config.GetVectorFieldEnableMmap() + : mmap_config.GetScalarFieldEnableMmap(); + const bool use_mmap = has_mmap_setting ? mmap_enabled : global_use_mmap; + if (TryLoadVortexColumnGroup(column_group, properties, index, @@ -8092,6 +8114,7 @@ ChunkedSegmentSealedImpl::LoadColumnGroup( eager_load, has_warmup_setting, aggregated_warmup_policy, + use_mmap, op_ctx, is_replace, nullptr, @@ -8099,11 +8122,6 @@ ChunkedSegmentSealedImpl::LoadColumnGroup( return; } - auto& mmap_config = storage::MmapManager::GetInstance().GetMmapConfig(); - bool global_use_mmap = is_vector ? mmap_config.GetVectorFieldEnableMmap() - : mmap_config.GetScalarFieldEnableMmap(); - auto use_mmap = has_mmap_setting ? mmap_enabled : global_use_mmap; - auto needed_columns = std::make_shared>(); needed_columns->reserve(milvus_field_ids.size()); for (const auto& fid : milvus_field_ids) { diff --git a/internal/core/src/segcore/ChunkedSegmentSealedImpl.h b/internal/core/src/segcore/ChunkedSegmentSealedImpl.h index 1f007a1285..543e29a3e8 100644 --- a/internal/core/src/segcore/ChunkedSegmentSealedImpl.h +++ b/internal/core/src/segcore/ChunkedSegmentSealedImpl.h @@ -2002,6 +2002,7 @@ class ChunkedSegmentSealedImpl : public SegmentSealed { bool eager_load, bool has_warmup_setting, const std::string& aggregated_warmup_policy, + bool use_mmap, milvus::OpContext* op_ctx, bool is_replace, RuntimeResourceState* runtime,