From 53644b34b78808ca00a0c7640ea28f2f9cd62734 Mon Sep 17 00:00:00 2001 From: sjh9714 <163989462+sjh9714@users.noreply.github.com> Date: Thu, 9 Jul 2026 00:31:58 +0900 Subject: [PATCH 1/2] fix(python): add collection close method --- python/tests/detail/test_collection_open.py | 40 +++++++++++++-------- python/zvec/model/collection.py | 6 ++++ 2 files changed, 31 insertions(+), 15 deletions(-) diff --git a/python/tests/detail/test_collection_open.py b/python/tests/detail/test_collection_open.py index 2d6c38ae4..987f8a30e 100644 --- a/python/tests/detail/test_collection_open.py +++ b/python/tests/detail/test_collection_open.py @@ -108,6 +108,23 @@ def test_collection( class TestCollectionOpen: + def test_close_releases_collection_for_reopen( + self, tmp_path_factory, collection_schema, collection_option + ): + temp_dir = tmp_path_factory.mktemp("zvec") + collection_path = temp_dir / "test_collection" + collection_path_str = str(collection_path) + + created_coll = zvec.create_and_open( + path=collection_path_str, schema=collection_schema, option=collection_option + ) + + created_coll.close() + + opened_coll = zvec.open(path=collection_path_str, option=collection_option) + assert opened_coll.path == collection_path_str + opened_coll.destroy() + def test_open_basic_functionality( self, tmp_path_factory, collection_schema, collection_option ): @@ -818,8 +835,9 @@ def test_open_concurrent_same_path(self, tmp_path_factory): def open_collection_thread(thread_id): try: coll = zvec.open(path=str(collection_path), option=collection_option) + collection_path_result = coll.path with lock: - results.append((thread_id, coll)) + results.append((thread_id, collection_path_result)) # Close the collection if opened successfully if hasattr(coll, "close") and coll is not None: coll.close() @@ -847,23 +865,15 @@ def open_collection_thread(thread_id): ) # Additional verification: check that the successful open has a valid collection - successful_thread_id, successful_collection = results[0] - assert successful_collection is not None, ( + _, successful_collection_path = results[0] + assert successful_collection_path is not None, ( "Successful open should return a valid collection" ) - assert successful_collection.path == str(collection_path), ( - "Collection path mismatch" - ) + assert successful_collection_path == str(collection_path), "Collection path mismatch" - # Clean up the successfully opened collection - if ( - hasattr(successful_collection, "destroy") - and successful_collection is not None - ): - try: - successful_collection.destroy() - except Exception as e: - print(f"Warning: failed to destroy collection: {e}") + # Clean up after the successful worker released its collection handle. + cleanup_coll = zvec.open(path=str(collection_path), option=collection_option) + cleanup_coll.destroy() def test_open_with_corrupted_files(self, tmp_path_factory): # First create a collection diff --git a/python/zvec/model/collection.py b/python/zvec/model/collection.py index b90caae94..04968b0a5 100644 --- a/python/zvec/model/collection.py +++ b/python/zvec/model/collection.py @@ -56,6 +56,12 @@ def __init__(self, obj: _Collection): self._schema = None self._querier = None + def close(self) -> None: + """Release the underlying collection handle.""" + self._obj = None + self._schema = None + self._querier = None + @classmethod def _from_core(cls, core_collection: _Collection) -> Collection: if not core_collection: From 042dfa18668f7434e73c03779857dfed8ca12bef Mon Sep 17 00:00:00 2001 From: sjh9714 <163989462+sjh9714@users.noreply.github.com> Date: Wed, 22 Jul 2026 07:00:14 +0900 Subject: [PATCH 2/2] fix(db): expose explicit native Collection::Close() Close() flushes and releases segments, managers, and the collection file lock even while other references to the handle remain alive. Post-close operations are rejected with "collection is already closed" instead of touching released state, and closing twice is a no-op. Python Collection.close() now calls the native Close() before dropping its references. --- .../detail/test_collection_create_and_open.py | 8 +-- python/tests/detail/test_collection_open.py | 53 ++++++++++++++++++- python/zvec/__init__.pyi | 1 + python/zvec/model/collection.py | 10 +++- src/binding/python/model/python_collection.cc | 13 ++++- src/db/collection.cc | 35 ++++++++++-- src/db/common/typedef.h | 15 ++++++ src/include/zvec/db/collection.h | 2 + 8 files changed, 125 insertions(+), 12 deletions(-) diff --git a/python/tests/detail/test_collection_create_and_open.py b/python/tests/detail/test_collection_create_and_open.py index f746f7803..efe59910a 100644 --- a/python/tests/detail/test_collection_create_and_open.py +++ b/python/tests/detail/test_collection_create_and_open.py @@ -586,7 +586,7 @@ def open_collection_thread(thread_id): path=str(collection_path), option=collection_option ) with lock: - results.append((thread_id, reopened_coll)) + results.append((thread_id, reopened_coll.path)) # Clean up the collection if opened successfully if hasattr(reopened_coll, "close") and reopened_coll is not None: reopened_coll.close() @@ -616,11 +616,11 @@ def open_collection_thread(thread_id): ) # Additional verification: check that the successful open has a valid collection - successful_thread_id, successful_collection = results[0] - assert successful_collection is not None, ( + successful_thread_id, successful_collection_path = results[0] + assert successful_collection_path is not None, ( "Successful open should return a valid collection" ) - assert successful_collection.path == str(collection_path), ( + assert successful_collection_path == str(collection_path), ( "Collection path mismatch" ) diff --git a/python/tests/detail/test_collection_open.py b/python/tests/detail/test_collection_open.py index 987f8a30e..da5e71b66 100644 --- a/python/tests/detail/test_collection_open.py +++ b/python/tests/detail/test_collection_open.py @@ -125,6 +125,55 @@ def test_close_releases_collection_for_reopen( assert opened_coll.path == collection_path_str opened_coll.destroy() + def test_close_releases_collection_with_outstanding_reference( + self, tmp_path_factory, collection_schema, collection_option + ): + temp_dir = tmp_path_factory.mktemp("zvec") + collection_path = temp_dir / "test_collection" + collection_path_str = str(collection_path) + + created_coll = zvec.create_and_open( + path=collection_path_str, schema=collection_schema, option=collection_option + ) + + # Keep an extra reference to the native handle, as a shallow copy or a + # stale reference would. close() must release the file lock anyway. + native_ref = created_coll._obj + + created_coll.close() + + # The path can be reopened even though native_ref is still alive. + opened_coll = zvec.open(path=collection_path_str, option=collection_option) + assert opened_coll.path == collection_path_str + + # Operations through the stale handle fail cleanly instead of touching + # released native state. + with pytest.raises(ValueError, match="already closed"): + native_ref.Stats() + + opened_coll.destroy() + + def test_close_is_idempotent( + self, tmp_path_factory, collection_schema, collection_option + ): + temp_dir = tmp_path_factory.mktemp("zvec") + collection_path = temp_dir / "test_collection" + collection_path_str = str(collection_path) + + created_coll = zvec.create_and_open( + path=collection_path_str, schema=collection_schema, option=collection_option + ) + + native_ref = created_coll._obj + created_coll.close() + created_coll.close() + # Closing the native handle again is also a no-op. + native_ref.Close() + + opened_coll = zvec.open(path=collection_path_str, option=collection_option) + assert opened_coll.path == collection_path_str + opened_coll.destroy() + def test_open_basic_functionality( self, tmp_path_factory, collection_schema, collection_option ): @@ -869,7 +918,9 @@ def open_collection_thread(thread_id): assert successful_collection_path is not None, ( "Successful open should return a valid collection" ) - assert successful_collection_path == str(collection_path), "Collection path mismatch" + assert successful_collection_path == str(collection_path), ( + "Collection path mismatch" + ) # Clean up after the successful worker released its collection handle. cleanup_coll = zvec.open(path=str(collection_path), option=collection_option) diff --git a/python/zvec/__init__.pyi b/python/zvec/__init__.pyi index 3e75f931b..0a755b589 100644 --- a/python/zvec/__init__.pyi +++ b/python/zvec/__init__.pyi @@ -117,6 +117,7 @@ class _Collection: arg2: schema._FieldSchema, arg3: param.AlterColumnOption, ) -> None: ... + def Close(self) -> None: ... def CreateIndex( self, arg0: str, arg1: param.IndexParam, arg2: param.IndexOption ) -> None: ... diff --git a/python/zvec/model/collection.py b/python/zvec/model/collection.py index 04968b0a5..b27fa8ed7 100644 --- a/python/zvec/model/collection.py +++ b/python/zvec/model/collection.py @@ -57,7 +57,15 @@ def __init__(self, obj: _Collection): self._querier = None def close(self) -> None: - """Release the underlying collection handle.""" + """Close the collection and release its native resources. + + Flushes pending writes and releases the collection's file lock so the + path can be reopened or removed, even if other references to this + collection still exist. Closing an already-closed collection is a + no-op. + """ + if self._obj is not None: + self._obj.Close() self._obj = None self._schema = None self._querier = None diff --git a/src/binding/python/model/python_collection.cc b/src/binding/python/model/python_collection.cc index b1311f119..175688109 100644 --- a/src/binding/python/model/python_collection.cc +++ b/src/binding/python/model/python_collection.cc @@ -118,15 +118,24 @@ void ZVecPyCollection::bind_ddl_methods( }); // bind collection ddl methods - col.def("Destroy", + col.def("Close", [](Collection &self) { Status status; { py::gil_scoped_release release; - status = self.Destroy(); + status = self.Close(); } throw_if_error(status); }) + .def("Destroy", + [](Collection &self) { + Status status; + { + py::gil_scoped_release release; + status = self.Destroy(); + } + throw_if_error(status); + }) .def("Flush", [](Collection &self) { Status status; { diff --git a/src/db/collection.cc b/src/db/collection.cc index 14cb2f4bc..7f0fcf9e2 100644 --- a/src/db/collection.cc +++ b/src/db/collection.cc @@ -74,9 +74,9 @@ class CollectionImpl : public Collection { private: Status Open(const CollectionOptions &options); - Status Close(); - public: + Status Close() override; + Status Destroy() override; Status Flush() override; @@ -227,6 +227,8 @@ class CollectionImpl : public Collection { bool destroyed_{false}; + bool closed_{false}; + CollectionSchema::Ptr schema_; CollectionOptions options_; @@ -295,7 +297,7 @@ void CollectionImpl::prepare_schema() { CollectionImpl::CollectionImpl(const std::string &path) : path_(path) {} CollectionImpl::~CollectionImpl() { - if (!destroyed_) { + if (!destroyed_ && !closed_) { Close(); } } @@ -324,11 +326,16 @@ Status CollectionImpl::Open(const CollectionOptions &options) { } Status CollectionImpl::Close() { - // only called in deconstructor std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + // closing twice is a no-op + if (closed_) { + return Status::OK(); + } + closed_ = true; + return close_unsafe(); } @@ -361,6 +368,7 @@ Status CollectionImpl::Destroy() { std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS(closed_, false); auto s = close_unsafe(); CHECK_RETURN_STATUS(s); @@ -377,6 +385,7 @@ Status CollectionImpl::Flush() { std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS(closed_, false); return flush_unsafe(); } @@ -391,6 +400,7 @@ Status CollectionImpl::flush_unsafe() { Result CollectionImpl::Path() const { CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); return path_; } @@ -399,6 +409,7 @@ Result CollectionImpl::Stats() const { std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); auto segments = get_all_segments(); @@ -440,6 +451,7 @@ Result CollectionImpl::Schema() const { std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); return *schema_; } @@ -448,6 +460,7 @@ Result CollectionImpl::Options() const { std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); return options_; } @@ -460,6 +473,7 @@ Status CollectionImpl::CreateIndex(const std::string &column_name, std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS(closed_, false); if (index_params == nullptr) { return Status::InvalidArgument("CreateIndex: index_params is null"); @@ -701,6 +715,7 @@ Status CollectionImpl::DropIndex(const std::string &column_name) { std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS(closed_, false); auto new_schema = std::make_shared(*schema_); auto s = new_schema->drop_index(column_name); @@ -885,6 +900,7 @@ Status CollectionImpl::Optimize(const OptimizeOptions &options) { // allowed CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS(closed_, false); std::vector persist_segments; @@ -1240,6 +1256,7 @@ Status CollectionImpl::AddColumn(const FieldSchema::Ptr &column_schema, std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS(closed_, false); // validate auto s = validate("", column_schema, expression, "", ColumnOp::ADD); @@ -1312,6 +1329,7 @@ Status CollectionImpl::DropColumn(const std::string &column_name) { std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS(closed_, false); // validate auto s = validate(column_name, nullptr, "", "", ColumnOp::DROP); @@ -1386,6 +1404,7 @@ Status CollectionImpl::AlterColumn(const std::string &column_name, std::lock_guard lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS(closed_, false); // validate auto s = @@ -1525,6 +1544,7 @@ Result CollectionImpl::write_impl(std::vector &docs, std::shared_lock lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); for (auto &&doc : docs) { auto s = doc.validate_and_sanitize(schema_, mode == WriteMode::UPDATE); @@ -1618,6 +1638,7 @@ Result CollectionImpl::Delete( std::shared_lock lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); // TODO: The granularity of the write_lock is too coarse. std::lock_guard write_lock(write_mtx_); @@ -1636,6 +1657,7 @@ Status CollectionImpl::DeleteByFilter(const std::string &filter) { std::shared_lock lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS(closed_, false); SearchQuery query; query.filter_ = filter; @@ -1666,6 +1688,7 @@ Result CollectionImpl::Query(const SearchQuery &query) const { std::shared_lock lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); // When field_name_ is set, use get_field to retrieve the schema uniformly. // validate checks that the field type matches the query type @@ -1697,6 +1720,7 @@ Result CollectionImpl::Query(const MultiQuery &query) const { std::shared_lock lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); if (query.queries.size() < 2) { return tl::make_unexpected(Status::InvalidArgument( @@ -1790,6 +1814,7 @@ Result CollectionImpl::GroupByQuery( std::shared_lock lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); auto segments = get_all_segments(); if (segments.empty()) { @@ -1821,6 +1846,7 @@ Result CollectionImpl::Fetch( std::shared_lock lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); auto segments = get_all_segments(); @@ -1854,6 +1880,7 @@ Result CollectionImpl::DebugGetHnswStorageMode( std::shared_lock lock(schema_handle_mtx_); CHECK_DESTROY_RETURN_STATUS_EXPECTED(destroyed_, false); + CHECK_CLOSED_RETURN_STATUS_EXPECTED(closed_, false); // Try all segments (including the writing one). The first segment that has // a fully-built HNSW index wins; if only a building segment exists we still diff --git a/src/db/common/typedef.h b/src/db/common/typedef.h index 79d587aff..5918830af 100644 --- a/src/db/common/typedef.h +++ b/src/db/common/typedef.h @@ -159,6 +159,21 @@ using idx_t = uint64_t; Status::InvalidArgument("collection is already destroyed.")); \ } +#define CHECK_CLOSED_RETURN_STATUS(status, expect) \ + if (status != expect) { \ + LOG_ERROR("Collection[%s] is already closed.", \ + schema_->name().c_str()); \ + return Status::InvalidArgument("collection is already closed."); \ + } + +#define CHECK_CLOSED_RETURN_STATUS_EXPECTED(status, expect) \ + if (status != expect) { \ + LOG_ERROR("Collection[%s] is already closed.", \ + schema_->name().c_str()); \ + return tl::make_unexpected( \ + Status::InvalidArgument("collection is already closed.")); \ + } + #define CHECK_RETURN_STATUS(status) \ if (!status.ok()) { \ return status; \ diff --git a/src/include/zvec/db/collection.h b/src/include/zvec/db/collection.h index 83a289c52..803e17383 100644 --- a/src/include/zvec/db/collection.h +++ b/src/include/zvec/db/collection.h @@ -53,6 +53,8 @@ class Collection { virtual ~Collection(); public: + virtual Status Close() = 0; + virtual Status Destroy() = 0; virtual Status Flush() = 0;