Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions src/core/interface/index.cc
Original file line number Diff line number Diff line change
Expand Up @@ -557,12 +557,15 @@ int Index::open(const std::string &file_path, StorageOptions storage_options) {
return core::IndexError_Unsupported;
}

// read_options.create_new
int ret = storage_->open(file_path, storage_options.create_new);
const core::IndexStorage::OpenOptions open_options{storage_options.create_new,
storage_options.read_only};
int ret = storage_options.type == StorageOptions::StorageType::kBufferPool
? storage_->open(file_path, open_options)
: storage_->open(file_path, storage_options.create_new);
if (ret != 0) {
LOG_ERROR("Failed to open storage, path: %s, err: %s", file_path.c_str(),
core::IndexError::What(ret));
return core::IndexError_Runtime;
return ret;
}
ret = prepare_streamer_open(storage_options);
if (ret != 0) {
Expand Down
24 changes: 15 additions & 9 deletions src/core/utility/buffer_storage.cc
Original file line number Diff line number Diff line change
Expand Up @@ -1001,7 +1001,15 @@ class BufferStorage : public IndexStorage {

//! Open storage
int open(const std::string &path, bool create_if_missing) override {
if (!ailego::File::IsExist(path) && create_if_missing) {
// Preserve the access mode of callers using the legacy boolean API.
return open_with_options(
path, OpenOptions{create_if_missing, /*read_only=*/!create_if_missing});
}

int open_with_options(const std::string &path,
const OpenOptions &options) override {
const bool writable = !options.read_only;
if (!ailego::File::IsExist(path) && options.create_if_missing) {
size_t last_slash = path.rfind('/');
if (last_slash != std::string::npos) {
ailego::File::MakePath(path.substr(0, last_slash));
Expand All @@ -1017,9 +1025,8 @@ class BufferStorage : public IndexStorage {
try {
// Capture both owners before replacing the published state. In
// particular, a failed reopen must not change existing segments' pool.
// create_if_missing also indicates write intent, like MMapFileStorage.
auto candidate_pool = std::make_shared<ailego::VecBufferPool>(
path, /*writable=*/create_if_missing);
auto candidate_pool =
std::make_shared<ailego::VecBufferPool>(path, writable);
auto candidate_handle =
std::make_shared<ailego::VecBufferPoolHandle>(candidate_pool);
file_name_ = path;
Expand All @@ -1042,16 +1049,16 @@ class BufferStorage : public IndexStorage {
const size_t page_count =
file_size == 0 ? 0 : (file_size - 1) / ailego::kVectorPageSize + 1;
const size_t metadata_bytes =
ailego::VecBufferPool::metadata_bytes_for_page_count(
page_count, /*writable=*/create_if_missing);
ailego::VecBufferPool::metadata_bytes_for_page_count(page_count,
writable);
const size_t available =
ailego::MemoryLimitPool::get_instance().available();
const bool cache_can_fit =
metadata_bytes != std::numeric_limits<size_t>::max() &&
metadata_bytes <= available &&
ailego::kVectorPageSize <= available - metadata_bytes;
ret = cache_can_fit ? buffer_pool_->init() : -1;
if (ret != 0 && create_if_missing) {
if (ret != 0 && writable) {
this->close_index();
return IndexError_NoMemory;
}
Expand All @@ -1063,8 +1070,7 @@ class BufferStorage : public IndexStorage {
path.c_str(), available, metadata_bytes, ailego::kVectorPageSize);
}
LOG_INFO("BufferStorage opened: file=%s, writable=%d, segment_count=%zu",
file_name_.c_str(), static_cast<int>(create_if_missing),
segments_.size());
file_name_.c_str(), static_cast<int>(writable), segments_.size());
return 0;
}

Expand Down
20 changes: 20 additions & 0 deletions src/include/zvec/core/framework/index_storage.h
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,11 @@ class IndexStorage : public IndexModule {
//! Index Storage Pointer
typedef std::shared_ptr<IndexStorage> Pointer;

struct OpenOptions {
bool create_if_missing{false};
bool read_only{true};
};

struct MemoryBlock {
enum MemoryBlockType {
MBT_UNKNOWN = 0,
Expand Down Expand Up @@ -650,6 +655,15 @@ class IndexStorage : public IndexModule {
//! Open storage
virtual int open(const std::string &path, bool create_if_missing) = 0;

//! Open with independent creation and access permissions. Implementations
//! that have not opted in return Unsupported instead of ignoring access.
int open(const std::string &path, const OpenOptions &options) {
if (options.create_if_missing && options.read_only) {
return IndexError_InvalidArgument;
}
return open_with_options(path, options);
}

//! Flush storage
virtual int flush() = 0;

Expand Down Expand Up @@ -709,6 +723,12 @@ class IndexStorage : public IndexModule {
virtual std::string file_path() const {
return "";
}

protected:
//! Separate dispatch keeps legacy open() overrides source-compatible.
virtual int open_with_options(const std::string &, const OpenOptions &) {
return IndexError_Unsupported;
}
};

} // namespace core
Expand Down
84 changes: 79 additions & 5 deletions tests/core/interface/index_interface_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -4868,6 +4868,11 @@ TEST(IndexInterface, IsDirty) {
TEST(IndexInterface, IsDirtyBufferPool) {
constexpr uint32_t kDimension = 16;
const std::string index_name{"test_is_dirty_bp.index"};
auto &pool = zvec::ailego::MemoryLimitPool::get_instance();
const size_t previous_capacity = pool.capacity();
ASSERT_EQ(0, pool.init(64UL * 1024UL * 1024UL));
auto restore = zvec::ailego::ScopeGuard::Make(
[&]() { EXPECT_EQ(0, pool.init(previous_capacity)); });

zvec::test_util::RemoveTestFiles(index_name);

Expand All @@ -4881,15 +4886,16 @@ TEST(IndexInterface, IsDirtyBufferPool) {
.build();
auto index = IndexFactory::CreateAndInitIndex(*param);
ASSERT_NE(nullptr, index);
index->open(index_name, {StorageOptions::StorageType::kMMAP, true});
ASSERT_EQ(
0, index->open(index_name, {StorageOptions::StorageType::kMMAP, true}));
std::vector<float> vec(kDimension, 1.0f);
VectorData vd;
vd.vector = DenseVector{vec.data()};
ASSERT_EQ(0, index->add(vd, 1));
index->close();
ASSERT_EQ(0, index->close());
}

// Reopen with BufferPool storage in writable mode
// Reopen an existing file for writing without permitting file creation.
{
auto param = FlatIndexParamBuilder()
.with_metric_type(MetricType::kInnerProduct)
Expand All @@ -4899,7 +4905,9 @@ TEST(IndexInterface, IsDirtyBufferPool) {
.build();
auto index = IndexFactory::CreateAndInitIndex(*param);
ASSERT_NE(nullptr, index);
index->open(index_name, {StorageOptions::StorageType::kBufferPool, true});
ASSERT_EQ(0,
index->open(index_name, {StorageOptions::StorageType::kBufferPool,
false, false}));

ASSERT_FALSE(index->is_dirty());

Expand All @@ -4912,12 +4920,78 @@ TEST(IndexInterface, IsDirtyBufferPool) {
ASSERT_EQ(0, index->flush());
ASSERT_FALSE(index->is_dirty());

index->close();
ASSERT_EQ(0, index->close());
}

// Both the original and appended vectors must survive another reopen.
{
auto param = FlatIndexParamBuilder()
.with_metric_type(MetricType::kInnerProduct)
.with_data_type(DataType::DT_FP32)
.with_dimension(kDimension)
.with_is_sparse(false)
.build();
auto index = IndexFactory::CreateAndInitIndex(*param);
ASSERT_NE(nullptr, index);
ASSERT_EQ(0,
index->open(index_name, {StorageOptions::StorageType::kBufferPool,
false, true}));
for (uint32_t id : {1U, 2U}) {
VectorDataBuffer fetched;
ASSERT_EQ(0, index->fetch(id, &fetched));
const auto &data =
std::get<DenseVectorBuffer>(fetched.vector_buffer).data;
const std::vector<float> expected(kDimension, static_cast<float>(id));
ASSERT_EQ(expected.size() * sizeof(float), data.size());
EXPECT_EQ(0, std::memcmp(expected.data(), data.data(), data.size()));
}
ASSERT_EQ(0, index->close());
}

zvec::test_util::RemoveTestFiles(index_name);
}

TEST(IndexInterface, BufferPoolOpenPreservesCreationAndMemoryErrors) {
using zvec::core::IndexError_NoMemory;
using zvec::core::IndexError_OpenFile;
const std::string path = "buffer_pool_open_errors.index";
zvec::test_util::RemoveTestFiles(path);
auto remove = zvec::ailego::ScopeGuard::Make(
[&]() { zvec::test_util::RemoveTestFiles(path); });
auto &pool = zvec::ailego::MemoryLimitPool::get_instance();
const size_t previous_capacity = pool.capacity();
auto restore = zvec::ailego::ScopeGuard::Make(
[&]() { EXPECT_EQ(0, pool.init(previous_capacity)); });
ASSERT_EQ(0, pool.init(64UL * 1024UL * 1024UL));
auto param = FlatIndexParamBuilder()
.with_metric_type(MetricType::kL2sq)
.with_data_type(DataType::DT_FP32)
.with_dimension(16)
.with_is_sparse(false)
.build();
{
auto index = IndexFactory::CreateAndInitIndex(*param);
ASSERT_NE(nullptr, index);
EXPECT_EQ(IndexError_OpenFile,
index->open(path, {StorageOptions::StorageType::kBufferPool,
false, false}));
EXPECT_FALSE(zvec::ailego::File::IsExist(path));
}
{
auto index = IndexFactory::CreateAndInitIndex(*param);
ASSERT_NE(nullptr, index);
ASSERT_EQ(
0, index->open(path, {StorageOptions::StorageType::kBufferPool, true}));
ASSERT_EQ(0, index->close());
}
ASSERT_EQ(0, pool.init(zvec::ailego::kVectorPageSize));
auto index = IndexFactory::CreateAndInitIndex(*param);
ASSERT_NE(nullptr, index);
EXPECT_EQ(IndexError_NoMemory,
index->open(path, {StorageOptions::StorageType::kBufferPool, false,
false}));
}

TEST(IndexInterface, BuilderSetsAllBaseFields) {
auto param =
FlatIndexParamBuilder()
Expand Down
124 changes: 121 additions & 3 deletions tests/core/utility/buffer_storage_write_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -49,13 +49,16 @@ class BufferStorageWriteTest : public ::testing::Test {
ailego::File::Delete(file_path_);
}

// Open BufferStorage in writable mode (create_if_missing=true)
// Open BufferStorage in writable mode, creating it when missing.
IndexStorage::Pointer open_writable() {
auto storage = IndexFactory::CreateStorage("BufferStorage");
if (!storage) return nullptr;
ailego::Params params;
storage->init(params);
if (storage->open(file_path_, true) != 0) return nullptr;
if (storage->open(file_path_, IndexStorage::OpenOptions{true, false}) !=
0) {
return nullptr;
}
return storage;
}

Expand All @@ -65,13 +68,128 @@ class BufferStorageWriteTest : public ::testing::Test {
if (!storage) return nullptr;
ailego::Params params;
storage->init(params);
if (storage->open(file_path_, false) != 0) return nullptr;
if (storage->open(file_path_, IndexStorage::OpenOptions{false, true}) !=
0) {
return nullptr;
}
return storage;
}

std::string file_path_;
};

TEST_F(BufferStorageWriteTest, WritableReopenWithoutCreationPersistsChanges) {
{
auto storage = open_writable();
ASSERT_NE(nullptr, storage);
ASSERT_EQ(0, storage->append("payload", 4096));
auto segment = storage->get("payload");
ASSERT_NE(nullptr, segment);
const std::string original = "original";
ASSERT_EQ(original.size(),
segment->write(0, original.data(), original.size()));
ASSERT_EQ(0, storage->close());
}

const std::string expected = "updated after writable reopen";
{
auto storage = IndexFactory::CreateStorage("BufferStorage");
ASSERT_NE(nullptr, storage);
ASSERT_EQ(0, storage->init(ailego::Params{}));
ASSERT_EQ(
0, storage->open(file_path_, IndexStorage::OpenOptions{false, false}));
ASSERT_TRUE(storage->vec_buffer_pool()->writable());
auto segment = storage->get("payload");
ASSERT_NE(nullptr, segment);
ASSERT_EQ(expected.size(),
segment->write(0, expected.data(), expected.size()));
ASSERT_EQ(0, storage->append("appended", 4096));
ASSERT_EQ(0, storage->flush());
ASSERT_EQ(0, storage->close());
}

auto storage = open_read_only();
ASSERT_NE(nullptr, storage);
EXPECT_FALSE(storage->vec_buffer_pool()->writable());
EXPECT_TRUE(storage->has("appended"));
auto segment = storage->get("payload");
ASSERT_NE(nullptr, segment);
std::string actual(expected.size(), '\0');
ASSERT_EQ(actual.size(), segment->fetch(0, actual.data(), actual.size()));
EXPECT_EQ(expected, actual);
ASSERT_EQ(0, storage->close());
}

TEST_F(BufferStorageWriteTest,
WritableOpenWithoutCreationRequiresExistingFile) {
auto storage = IndexFactory::CreateStorage("BufferStorage");
ASSERT_NE(nullptr, storage);
ASSERT_EQ(0, storage->init(ailego::Params{}));
EXPECT_EQ(IndexError_OpenFile,
storage->open(file_path_, IndexStorage::OpenOptions{false, false}));
EXPECT_FALSE(ailego::File::IsExist(file_path_));
EXPECT_EQ(nullptr, storage->vec_buffer_pool());

ASSERT_EQ(0,
storage->open(file_path_, IndexStorage::OpenOptions{true, false}));
EXPECT_TRUE(storage->vec_buffer_pool()->writable());
ASSERT_EQ(0, storage->close());
}

TEST_F(BufferStorageWriteTest, ReadOnlyOpenCannotCreateFile) {
auto storage = IndexFactory::CreateStorage("BufferStorage");
ASSERT_NE(nullptr, storage);
ASSERT_EQ(0, storage->init(ailego::Params{}));
EXPECT_EQ(IndexError_InvalidArgument,
storage->open(file_path_, IndexStorage::OpenOptions{true, true}));
EXPECT_FALSE(ailego::File::IsExist(file_path_));
EXPECT_EQ(nullptr, storage->vec_buffer_pool());
}

TEST_F(BufferStorageWriteTest, LegacyOpenPreservesAccessModes) {
auto storage = IndexFactory::CreateStorage("BufferStorage");
ASSERT_NE(nullptr, storage);
ASSERT_EQ(0, storage->init(ailego::Params{}));
ASSERT_EQ(0, storage->open(file_path_, true));
EXPECT_TRUE(storage->vec_buffer_pool()->writable());
ASSERT_EQ(0, storage->close());
ASSERT_EQ(0, storage->open(file_path_, false));
EXPECT_FALSE(storage->vec_buffer_pool()->writable());
ASSERT_EQ(0, storage->close());
}

TEST_F(BufferStorageWriteTest, ExplicitOpenRequiresImplementationSupport) {
auto storage = IndexFactory::CreateStorage("MMapFileStorage");
ASSERT_NE(nullptr, storage);
ASSERT_EQ(0, storage->init(ailego::Params{}));
EXPECT_EQ(IndexError_Unsupported,
storage->open(file_path_, IndexStorage::OpenOptions{true, false}));
EXPECT_FALSE(ailego::File::IsExist(file_path_));
}

TEST_F(BufferStorageWriteTest, WritableReopenNeedsCacheAndAllowsRetry) {
{
auto storage = open_writable();
ASSERT_NE(nullptr, storage);
ASSERT_EQ(0, storage->close());
}
auto &pool = ailego::MemoryLimitPool::get_instance();
AILEGO_DEFER([&]() { EXPECT_EQ(0, pool.init(64UL * 1024UL * 1024UL)); });
ASSERT_EQ(0, pool.init(ailego::kVectorPageSize));
auto storage = IndexFactory::CreateStorage("BufferStorage");
ASSERT_NE(nullptr, storage);
ASSERT_EQ(0, storage->init(ailego::Params{}));
EXPECT_EQ(IndexError_NoMemory,
storage->open(file_path_, IndexStorage::OpenOptions{false, false}));
EXPECT_EQ(nullptr, storage->vec_buffer_pool());

ASSERT_EQ(0, pool.init(64UL * 1024UL * 1024UL));
ASSERT_EQ(0,
storage->open(file_path_, IndexStorage::OpenOptions{false, false}));
EXPECT_TRUE(storage->vec_buffer_pool()->writable());
ASSERT_EQ(0, storage->close());
}

TEST_F(BufferStorageWriteTest, MissingFileReturnsErrorAndAllowsRetry) {
auto storage = IndexFactory::CreateStorage("BufferStorage");
ASSERT_NE(nullptr, storage);
Expand Down
Loading