furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-21 02:05:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: support mmap for vortex local format

Browse Source 
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: aoiasd <zhicheng.yue@zilliz.com>
This commit is contained in:
aoiasd
2026-07-15 17:28:45 +08:00
co-authored by Claude Opus 4.6
parent d688df5445
commit 0704ac563b
7 changed files with 573 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/mmap/SparseVortexFileSystem.cpp
+410 -139
View File
@@ -20,6 +20,7 @@
#include <cerrno>
#include <cstdint>
#include <cstring>
#include <filesystem>
#include <functional>
#include <limits>
#include <memory>
@@ -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<SparseVortexMmapFile> {
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<size_t>(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<uint64_t>(
remaining,
static_cast<uint64_t>(std::numeric_limits<ssize_t>::max()));
const auto result = ::pwrite(fd,
data + written,
static_cast<size_t>(chunk),
static_cast<off_t>(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<uint64_t>(result);
}
return arrow::Status::OK();
}
arrow::Result<int64_t>
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<uint64_t>(
remaining,
static_cast<uint64_t>(std::numeric_limits<ssize_t>::max()));
const auto result = ::pread(fd,
out + read,
static_cast<size_t>(chunk),
static_cast<off_t>(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<uint64_t>(result);
}
return static_cast<int64_t>(read);
}
} // namespace
class SparseVortexRangeFileBase
: public milvus_storage::vortex::VortexRangeFile,
public std::enable_shared_from_this<SparseVortexRangeFileBase> {
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<uint64_t>(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<uint64_t>(std::numeric_limits<off_t>::max()) &&
length <=
static_cast<uint64_t>(std::numeric_limits<off_t>::max())) {
if (::syscall(SYS_fallocate,
fd_,
FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE,
static_cast<off_t>(offset),
static_cast<off_t>(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<uint64_t>(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<size_t>(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<off_t>(offset),
static_cast<off_t>(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<size_t>(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<int64_t>
@@ -302,29 +300,18 @@ class SparseVortexMmapFile final
if (pos >= size_) {
return 0;
}
const auto available = size_ - pos;
const auto read = std::min(static_cast<uint64_t>(nbytes), available);
const auto read = std::min(static_cast<uint64_t>(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<int64_t>(read);
return ReadAtLocked(pos, read, out);
}
arrow::Result<std::shared_ptr<arrow::Buffer>>
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<uint64_t>(nbytes), available);
const auto read = std::min(static_cast<uint64_t>(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<int64_t>
ReadAtLocked(uint64_t position, uint64_t nbytes, void* out) const = 0;
virtual arrow::Result<std::shared_ptr<arrow::Buffer>>
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<size_t>(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<size_t>(data.size()));
PopulateWrittenRangeLocked(offset, static_cast<uint64_t>(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<uint64_t>(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<size_t>(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<size_t>(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<int64_t>
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<int64_t>(nbytes);
}
arrow::Result<std::shared_ptr<arrow::Buffer>>
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<arrow::Buffer>(
new arrow::Buffer(mapping_ + pos, static_cast<int64_t>(read)),
new arrow::Buffer(mapping_ + position,
static_cast<int64_t>(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<uint64_t>(page_size_value)
: static_cast<uint64_t>(4096);
const auto end = std::min<uint64_t>(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<uint64_t>(data.size()));
}
void
ClearRangeLocked(uint64_t offset,
uint64_t length,
int fallocate_error) override {
constexpr uint64_t kZeroBufferSize = 64 * 1024;
const std::vector<uint8_t> zeros(
static_cast<size_t>(std::min(length, kZeroBufferSize)), 0);
uint64_t cleared = 0;
while (cleared < length) {
const auto chunk =
std::min<uint64_t>(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<int64_t>
ReadAtLocked(uint64_t position, uint64_t nbytes, void* out) const override {
return PReadAll(fd_, static_cast<uint8_t*>(out), position, nbytes);
}
arrow::Result<std::shared_ptr<arrow::Buffer>>
ReadAtLocked(uint64_t position, uint64_t nbytes) const override {
ARROW_ASSIGN_OR_RAISE(
auto buffer, arrow::AllocateBuffer(static_cast<int64_t>(nbytes)));
ARROW_ASSIGN_OR_RAISE(
auto bytes_read,
PReadAll(fd_, buffer->mutable_data(), position, nbytes));
std::shared_ptr<arrow::Buffer> 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<milvus_storage::vortex::VortexRangeFile>
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<SparseVortexMmapRangeFile>(
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<SparseVortexMmapRangeFile>(
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<SparseVortexDiskRangeFile>(
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<SparseVortexMmapFile>(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<arrow::fs::FileSystem>
MakeSparseVortexFileSystem(std::string path) {
return std::make_shared<SparseVortexFileSystem>(std::move(path));
return MakeSparseVortexFileSystem(std::move(path),
SparseVortexFileSystemOptions{});
}
std::shared_ptr<arrow::fs::FileSystem>
MakeSparseVortexFileSystem(std::string path,
SparseVortexFileSystemOptions options) {
return std::make_shared<SparseVortexFileSystem>(std::move(path),
std::move(options));
}
} // namespace milvus
internal/core/src/mmap/SparseVortexFileSystem.h
+16
View File
@@ -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<arrow::fs::FileSystem>
MakeSparseVortexFileSystem(std::string path);
std::shared_ptr<arrow::fs::FileSystem>
MakeSparseVortexFileSystem(std::string path,
SparseVortexFileSystemOptions options);
} // namespace milvus
internal/core/src/mmap/VortexColumnGroup.cpp
+46 -2
View File
@@ -16,6 +16,8 @@
#include "mmap/VortexColumnGroup.h"
#include <atomic>
#include <filesystem>
#include <memory>
#include <string_view>
#include <utility>
@@ -37,6 +39,24 @@ namespace milvus {
namespace {
std::atomic<uint64_t> 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<std::string>& 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<VortexColumnFileInfo>& files,
std::shared_ptr<milvus_storage::api::Properties> properties,
const std::vector<std::string>& 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<milvus_storage::vortex::VortexFooterReader>(
state.sparse_fs,
internal/core/src/mmap/VortexColumnGroup.h
+18
View File
@@ -22,6 +22,7 @@
#include <vector>
#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<VortexColumnFileInfo>& files,
std::shared_ptr<milvus_storage::api::Properties> properties,
const std::vector<std::string>& field_names,
CacheWarmupPolicy cache_warmup_policy,
milvus::OpContext* op_ctx,
Options options);
~VortexColumnGroup();
void
internal/core/src/mmap/VortexColumnTest.cpp
+52
View File
@@ -16,6 +16,7 @@
#include "mmap/VortexColumn.h"
#include <array>
#include <cstdint>
#include <cmath>
#include <exception>
@@ -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<int>(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<char, 3> 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<int64_t>(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<int64_t>(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<arrow::Schema>& 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 =
internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp
+30 -12
View File
@@ -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<VortexColumnGroup>(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<std::vector<std::string>>();
needed_columns->reserve(milvus_field_ids.size());
for (const auto& fid : milvus_field_ids) {
internal/core/src/segcore/ChunkedSegmentSealedImpl.h
+1
View File
@@ -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,