From 957fcae54bd2d5d2a5d0e06c6af3a5d581319ced Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Fri, 24 Jul 2026 11:18:58 +0800 Subject: [PATCH 1/2] do inplace compact in csr dump introduce ChecksumReuseFileDumper trigger compact for edge table with sort_keys parallelsize compact and dump refine --- include/neug/storages/graph/property_graph.h | 1 + src/main/neug_db.cc | 4 +- src/storages/csr/csr_dump_utils.h | 126 ++++++++++ src/storages/csr/csr_parallel_utils.h | 141 +++++++++++ src/storages/csr/immutable_csr.cc | 126 +++++++--- src/storages/csr/mutable_csr.cc | 232 ++++++++++-------- src/storages/graph/edge_table.cc | 14 +- src/storages/graph/property_graph.cc | 17 +- .../loader/abstract_property_graph_loader.cc | 2 +- tests/storage/test_immutable_csr.cc | 199 +++++++-------- tests/storage/test_mutable_csr.cc | 143 +++++++++++ tests/storage/test_open_graph.cc | 63 +++++ tests/storage/test_property_graph.cc | 2 +- 13 files changed, 800 insertions(+), 270 deletions(-) create mode 100644 src/storages/csr/csr_dump_utils.h create mode 100644 src/storages/csr/csr_parallel_utils.h diff --git a/include/neug/storages/graph/property_graph.h b/include/neug/storages/graph/property_graph.h index 90d1c1033..3b28ed523 100644 --- a/include/neug/storages/graph/property_graph.h +++ b/include/neug/storages/graph/property_graph.h @@ -642,6 +642,7 @@ class PropertyGraph { label_t edge_label) const; void compact_schema(); + void compact_internal(bool compact_all_edge_tables); /// Insert / erase an edge table and keep the dirty tracker's edge slots /// in sync. diff --git a/src/main/neug_db.cc b/src/main/neug_db.cc index 0e303bc7a..a3634e990 100644 --- a/src/main/neug_db.cc +++ b/src/main/neug_db.cc @@ -516,9 +516,7 @@ std::shared_ptr NeugDB::consumeLiveGraphAndCommitCheckpoint( CheckpointSession& checkpoint_session) { SnapshotGuard guard(*snapshot_store_); auto* live_graph = guard.get().mutable_graph(); - // Compact rewrites only already-dirty tables (does not mark); dump then - // publishes. ClearAllDirty runs only after a successful Commit. - live_graph->Compact(); + // Compact inside DumpAndClear live_graph->DumpAndClear(checkpoint_session.staging_checkpoint()); auto published_checkpoint = checkpoint_session.Commit(); // Consumed graph is about to be dropped; ClearAllDirty is for the contract diff --git a/src/storages/csr/csr_dump_utils.h b/src/storages/csr/csr_dump_utils.h new file mode 100644 index 000000000..36e4e3146 --- /dev/null +++ b/src/storages/csr/csr_dump_utils.h @@ -0,0 +1,126 @@ +/** Copyright 2020 Alibaba Group Holding Limited. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "neug/storages/checkpoint.h" +#include "neug/storages/container/mmap_container.h" +#include "neug/utils/exception/exception.h" + +namespace neug::csr_dump { + +inline void validate_edge_count(uint64_t expected, size_t actual) { + if (actual == expected) { + return; + } + LOG(WARNING) << "Inconsistent edge count" + << ": expected " << expected << ", actual " << actual; + THROW_STORAGE_EXCEPTION("Inconsistent edge count: expected " + + std::to_string(expected) + ", actual " + + std::to_string(actual)); +} + +class ChecksumReuseFileDumper { + public: + ChecksumReuseFileDumper(Checkpoint& ckp, const IDataContainer* container) + : ckp_(ckp) { + const auto* mapped = dynamic_cast(container); + const auto* mapped_header = + mapped == nullptr ? nullptr : mapped->GetHeader(); + if (mapped != nullptr && !mapped->GetPath().empty() && + mapped_header != nullptr) { + FileHeader zeroed{}; + if (memcmp(mapped_header, &zeroed, sizeof(FileHeader)) != 0) { + reusable_source_ = mapped; + } + } + + MD5_Init(&md5_); + if (reusable_source_ == nullptr) { + open_output(); + } + } + + void AddSegment(const char* data, size_t len) { + if (len == 0) { + return; + } + MD5_Update(&md5_, data, len); + if (reusable_source_ == nullptr) { + WriteSegment(data, len); + } + } + + void operator()(const char* data, size_t len) { AddSegment(data, len); } + + bool BeginRewriteIfChanged() { + MD5_Final(header_.data_md5, &md5_); + if (reusable_source_ != nullptr && + memcmp(reusable_source_->GetHeader()->data_md5, header_.data_md5, + sizeof(header_.data_md5)) != 0) { + open_output(); + } + return reusable_source_ != nullptr && runtime_file_.has_value(); + } + + void WriteSegment(const char* data, size_t len) { + if (len != 0) { + out_.write(data, static_cast(len)); + } + } + + std::string CommitOrReuse() { + if (!runtime_file_.has_value()) { + return ckp_.LinkToSnapshot(reusable_source_->GetPath()); + } + out_.seekp(0); + out_.write(reinterpret_cast(&header_), sizeof(header_)); + out_.flush(); + if (!out_.good()) { + THROW_IO_EXCEPTION("Failed to flush file: " + runtime_file_->path()); + } + out_.close(); + return ckp_.CommitRuntimeFile(std::move(*runtime_file_)); + } + + private: + void open_output() { + runtime_file_.emplace(ckp_.CreateRuntimeFile()); + const auto& path = runtime_file_->path(); + out_.open(path, std::ios::binary); + if (!out_.is_open()) { + THROW_IO_EXCEPTION("Failed to open file for writing: " + path); + } + out_.seekp(sizeof(header_)); + } + + Checkpoint& ckp_; + const MMapContainer* reusable_source_{nullptr}; + FileHeader header_{}; + MD5_CTX md5_; + std::optional runtime_file_; + std::ofstream out_; +}; + +} // namespace neug::csr_dump diff --git a/src/storages/csr/csr_parallel_utils.h b/src/storages/csr/csr_parallel_utils.h new file mode 100644 index 000000000..1e02b5beb --- /dev/null +++ b/src/storages/csr/csr_parallel_utils.h @@ -0,0 +1,141 @@ +/** Copyright 2020 Alibaba Group Holding Limited. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +namespace neug::csr_parallel { + +// Vertex count per scheduling unit. Vertex degrees follow a power-law +// distribution, so a small chunk size with dynamic scheduling keeps +// high-degree vertices from creating stragglers. +constexpr size_t kDefaultChunkSize = 256; + +// Creating a std::thread costs noticeably more than scanning a small CSR. +// Keep small jobs serial and require enough estimated edge-level work for +// every worker before adding it. +constexpr size_t kMinWorkPerWorker = 32 * 1024; +constexpr size_t kMinParallelWork = 2 * kMinWorkPerWorker; + +namespace detail { + +inline int load_degree(const int& degree) { return degree; } + +inline int load_degree(const std::atomic& degree) { + return degree.load(std::memory_order_relaxed); +} + +inline size_t normalize_work(int degree) { + return degree > 0 ? static_cast(degree) : 0; +} + +// std::sort is O(n log n). Use the number of expected comparisons as its +// edge-level work estimate. +inline size_t sort_work(int degree) { + if (degree <= 1) { + return 0; + } + const size_t edge_count = static_cast(degree); + const size_t levels = std::bit_width(edge_count - 1); + const size_t max = std::numeric_limits::max(); + return edge_count > max / levels ? max : edge_count * levels; +} + +template +void parallel_for_degree_ranges(size_t total, const DEGREE_T* degrees, + size_t (*work_for_degree)(int), FUNC&& func, + size_t chunk_size) { + if (total == 0) { + return; + } + if (chunk_size == 0) { + chunk_size = kDefaultChunkSize; + } + + const size_t range_count = 1 + (total - 1) / chunk_size; + const size_t hardware_threads = + std::max(1, std::thread::hardware_concurrency()); + const size_t max_workers = std::min(hardware_threads, range_count); + + const size_t max = std::numeric_limits::max(); + const size_t work_limit = max_workers > max / kMinWorkPerWorker + ? max + : max_workers * kMinWorkPerWorker; + size_t estimated_work = 0; + for (size_t i = 0; i < total && estimated_work < work_limit; ++i) { + const size_t item_work = work_for_degree(load_degree(degrees[i])); + estimated_work += std::min(item_work, work_limit - estimated_work); + } + + size_t workers = 1; + if (estimated_work >= kMinParallelWork) { + workers = std::min(max_workers, estimated_work / kMinWorkPerWorker); + } + if (workers <= 1) { + func(0, total); + return; + } + std::atomic cursor{0}; + auto work = [&]() { + while (true) { + const size_t first = + cursor.fetch_add(chunk_size, std::memory_order_relaxed); + if (first >= total) { + break; + } + const size_t last = + first + chunk_size < total ? first + chunk_size : total; + func(first, last); + } + }; + std::vector threads; + threads.reserve(workers - 1); + for (size_t i = 1; i < workers; ++i) { + threads.emplace_back(work); + } + work(); + for (auto& t : threads) { + t.join(); + } +} + +} // namespace detail + +// Dynamic-chunk parallel normalization over vertex ranges. +template +void parallel_for_normalize_ranges(size_t total, const DEGREE_T* degrees, + FUNC&& func, + size_t chunk_size = kDefaultChunkSize) { + detail::parallel_for_degree_ranges(total, degrees, detail::normalize_work, + func, chunk_size); +} + +// Dynamic-chunk parallel sorting over vertex ranges. Sorting work is estimated +// as degree * ceil(log2(degree)). +template +void parallel_for_sort_ranges(size_t total, const DEGREE_T* degrees, + FUNC&& func, + size_t chunk_size = kDefaultChunkSize) { + detail::parallel_for_degree_ranges(total, degrees, detail::sort_work, func, + chunk_size); +} + +} // namespace neug::csr_parallel diff --git a/src/storages/csr/immutable_csr.cc b/src/storages/csr/immutable_csr.cc index f4813efc1..400ff94fe 100644 --- a/src/storages/csr/immutable_csr.cc +++ b/src/storages/csr/immutable_csr.cc @@ -15,6 +15,8 @@ #include "neug/storages/csr/immutable_csr.h" +#include "csr_dump_utils.h" +#include "csr_parallel_utils.h" #include "neug/storages/checkpoint_manifest.h" #include "neug/storages/module/module_factory.h" @@ -74,6 +76,43 @@ void ImmutableCsr::refresh_prefetch_policy() { prefetch_policy_ = create_csr_prefetch_policy(degree_stats); } +namespace { + +template +size_t normalize_immutable_adjacency_lists(NBR_T* nbr_list, NBR_T** adj_lists, + int* degrees, size_t vnum, + SEGMENT_SINK&& segment_sink) { + NBR_T* write_ptr = nbr_list; + size_t live_edges = 0; + for (size_t i = 0; i < vnum; ++i) { + NBR_T* read_ptr = adj_lists[i]; + NBR_T* segment = write_ptr; + adj_lists[i] = segment; + int new_degree = 0; + if (degrees[i] != 0) { + NBR_T* read_end = read_ptr + degrees[i]; + while (read_ptr != read_end) { + if (read_ptr->neighbor != std::numeric_limits::max()) { + if (write_ptr != read_ptr) { + *write_ptr = *read_ptr; + } + ++write_ptr; + ++new_degree; + } + ++read_ptr; + } + } + degrees[i] = new_degree; + live_edges += static_cast(new_degree); + segment_sink(reinterpret_cast(segment), + static_cast(new_degree) * sizeof(NBR_T)); + } + + return live_edges; +} + +} // namespace + template void ImmutableCsr::Dump(Checkpoint& ckp, CheckpointManifest& meta, const std::string& key) { @@ -81,50 +120,54 @@ void ImmutableCsr::Dump(Checkpoint& ckp, CheckpointManifest& meta, desc.module_type = ModuleTypeName(); desc.set("unsorted_since", std::to_string(unsorted_since_)); desc.set("edge_num", std::to_string(edge_num_.load())); - desc.set_path(ModuleDescriptor::kDegreeListPath, - ckp.Commit(*degree_list_buffer_)); - desc.set_path(ModuleDescriptor::kNbrListPath, ckp.Commit(*nbr_list_buffer_)); + + size_t vnum = size(); + auto** adj_lists = reinterpret_cast(adj_list_buffer_->GetData()); + auto* degrees = reinterpret_cast(degree_list_buffer_->GetData()); + csr_dump::ChecksumReuseFileDumper nbr_file(ckp, nbr_list_buffer_.get()); + size_t live_edges = normalize_immutable_adjacency_lists( + reinterpret_cast(nbr_list_buffer_->GetData()), adj_lists, degrees, + vnum, nbr_file); + csr_dump::validate_edge_count(edge_num_.load(), live_edges); + if (nbr_file.BeginRewriteIfChanged()) { + nbr_file.WriteSegment( + reinterpret_cast(nbr_list_buffer_->GetData()), + live_edges * sizeof(nbr_t)); + } + desc.set_path(ModuleDescriptor::kNbrListPath, nbr_file.CommitOrReuse()); + + csr_dump::ChecksumReuseFileDumper degree_file(ckp, degree_list_buffer_.get()); + auto* degree_data = + reinterpret_cast(degree_list_buffer_->GetData()); + size_t degree_bytes = degree_list_buffer_->GetDataSize(); + degree_file.AddSegment(degree_data, degree_bytes); + if (degree_file.BeginRewriteIfChanged()) { + degree_file.WriteSegment(degree_data, degree_bytes); + } + desc.set_path(ModuleDescriptor::kDegreeListPath, degree_file.CommitOrReuse()); meta.set_module(key, desc); } template void ImmutableCsr::compact() { - // For current adj_list where the dst vertex is invalid, swap it to the end. vid_t vnum = size(); if (vnum <= 0) { return; } - size_t removed = 0; - auto** adj_arr = reinterpret_cast(adj_list_buffer_->GetData()); - auto* deg_arr = reinterpret_cast(degree_list_buffer_->GetData()); - nbr_t* write_ptr = adj_arr[0]; + auto** adj_lists = reinterpret_cast(adj_list_buffer_->GetData()); + auto* degrees = reinterpret_cast(degree_list_buffer_->GetData()); + size_t live_edges = normalize_immutable_adjacency_lists( + reinterpret_cast(nbr_list_buffer_->GetData()), adj_lists, degrees, + vnum, [](const char*, size_t) {}); + csr_dump::validate_edge_count(edge_num_.load(), live_edges); + nbr_list_buffer_->Resize(live_edges * sizeof(nbr_t)); + nbr_t* segment = reinterpret_cast(nbr_list_buffer_->GetData()); for (vid_t i = 0; i < vnum; ++i) { - int deg = deg_arr[i]; - if (deg == 0) { - continue; - } - const nbr_t* read_ptr = adj_arr[i]; - const nbr_t* read_end = read_ptr + deg; - while (read_ptr != read_end) { - if (read_ptr->neighbor != std::numeric_limits::max()) { - if (removed) { - *write_ptr = *read_ptr; - } - ++write_ptr; - } else { - --deg_arr[i]; - ++removed; - } - ++read_ptr; + adj_lists[i] = segment; + if (degrees[i] != 0) { + segment += degrees[i]; } } - nbr_list_buffer_->Resize(nbr_list_buffer_->GetDataSize() - - removed * sizeof(nbr_t)); - nbr_t* ptr = reinterpret_cast(nbr_list_buffer_->GetData()); - for (vid_t i = 0; i < vnum; ++i) { - adj_arr[i] = ptr; - ptr += deg_arr[i]; - } } template @@ -171,11 +214,20 @@ void ImmutableCsr::batch_sort_by_edge_data(timestamp_t ts) { auto** adj_arr = reinterpret_cast(adj_list_buffer_->GetData()); const auto* deg_arr = reinterpret_cast(degree_list_buffer_->GetData()); - for (size_t i = 0; i != vnum; ++i) { - std::sort( - adj_arr[i], adj_arr[i] + deg_arr[i], - [](const nbr_t& lhs, const nbr_t& rhs) { return lhs.data < rhs.data; }); - } + // Each vertex owns a disjoint adjacency list, so sorting can be + // parallelized across vertex ranges without synchronization. + csr_parallel::parallel_for_sort_ranges( + vnum, deg_arr, [adj_arr, deg_arr](size_t first, size_t last) { + for (size_t i = first; i < last; ++i) { + if (deg_arr[i] <= 1) { + continue; + } + std::sort(adj_arr[i], adj_arr[i] + deg_arr[i], + [](const nbr_t& lhs, const nbr_t& rhs) { + return lhs.data < rhs.data; + }); + } + }); unsorted_since_ = ts; } diff --git a/src/storages/csr/mutable_csr.cc b/src/storages/csr/mutable_csr.cc index 5152fcf95..c6951d649 100644 --- a/src/storages/csr/mutable_csr.cc +++ b/src/storages/csr/mutable_csr.cc @@ -15,6 +15,8 @@ #include "neug/storages/csr/mutable_csr.h" +#include "csr_dump_utils.h" +#include "csr_parallel_utils.h" #include "neug/storages/checkpoint_manifest.h" #include "neug/storages/module/module_factory.h" @@ -95,27 +97,78 @@ void MutableCsr::refresh_prefetch_policy() { prefetch_policy_ = create_csr_prefetch_policy(degree_stats); } +namespace { + +// In-place removal of invalidated edges within a single vertex's adjacency +// list; live entries keep their relative order and their timestamps are +// reset. Returns the new degree. +template +int normalize_mutable_adjacency_list(NBR_T* data, int degree) { + NBR_T* read_ptr = data; + NBR_T* read_end = data + degree; + NBR_T* write_ptr = data; + int removed = 0; + while (read_ptr != read_end) { + if (read_ptr->timestamp.load(std::memory_order_relaxed) != + INVALID_TIMESTAMP) { + if (removed != 0) { + *write_ptr = *read_ptr; + } + write_ptr->timestamp.store(0, std::memory_order_relaxed); + ++write_ptr; + } else { + ++removed; + } + ++read_ptr; + } + return degree - removed; +} + +// Parallel in-place normalization of every vertex's adjacency list. Each +// vertex owns a disjoint capacity buffer, so per-vertex work is independent; +// only the live-edge count is reduced across threads. Returns the total +// number of live edges. template -bool is_nbr_list_unmodified(MD5_CTX& ctx, FileHeader& header, - const IDataContainer* nbr_container, - const NBR_T* const* adj_lists, const int* cap_arr, - size_t vnum) { - MD5_Init(&ctx); +size_t parallel_normalize_mutable_adjacency_lists(NBR_T** adj_lists, + std::atomic* degrees, + size_t vnum) { + std::atomic live_edges{0}; + csr_parallel::parallel_for_normalize_ranges( + vnum, degrees, + [adj_lists, degrees, &live_edges](size_t first, size_t last) { + size_t local_live = 0; + for (size_t i = first; i < last; ++i) { + int degree = degrees[i].load(std::memory_order_relaxed); + NBR_T* data = adj_lists[i]; + if (data == nullptr) { + continue; + } + int new_degree = normalize_mutable_adjacency_list(data, degree); + degrees[i].store(new_degree, std::memory_order_relaxed); + local_live += static_cast(new_degree); + } + if (local_live != 0) { + live_edges.fetch_add(local_live, std::memory_order_relaxed); + } + }); + return live_edges.load(std::memory_order_relaxed); +} + +template +size_t normalize_single_adjacency_list(NBR_T* data, size_t vnum) { + size_t live_edges = 0; for (size_t i = 0; i < vnum; ++i) { - const char* data = reinterpret_cast(adj_lists[i]); - size_t len = cap_arr[i] * sizeof(NBR_T); - MD5_Update(&ctx, data, len); - } - MD5_Final(header.data_md5, &ctx); - auto casted = dynamic_cast(nbr_container); - if (casted && !casted->GetPath().empty() && casted->GetHeader()) { - return memcmp(casted->GetHeader()->data_md5, header.data_md5, - sizeof(header.data_md5)) == 0; - } else { - return false; + if (data[i].timestamp.load(std::memory_order_relaxed) != + INVALID_TIMESTAMP) { + data[i].timestamp.store(0, std::memory_order_relaxed); + ++live_edges; + } } + return live_edges; } +} // namespace + template void MutableCsr::Dump(Checkpoint& ckp, CheckpointManifest& meta, const std::string& key) { @@ -126,44 +179,31 @@ void MutableCsr::Dump(Checkpoint& ckp, CheckpointManifest& meta, size_t vnum = vertex_capacity(); + // nbr_list_ identifies the reusable snapshot; adj_lists hold the live data. + auto** adj_lists = reinterpret_cast(adj_list_buffer_->GetData()); + auto* degrees = reinterpret_cast*>(degree_list_->GetData()); + const int* capacities = reinterpret_cast(cap_list_->GetData()); + csr_dump::ChecksumReuseFileDumper nbr_file(ckp, nbr_list_.get()); + // Normalize in parallel first; the checksum/rewrite pass below must stay + // serial because MD5 chaining requires the segments in vertex order. + size_t live_edges = + parallel_normalize_mutable_adjacency_lists(adj_lists, degrees, vnum); + csr_dump::validate_edge_count(edge_num_.load(), live_edges); + for (size_t i = 0; i < vnum; ++i) { + nbr_file.AddSegment(reinterpret_cast(adj_lists[i]), + static_cast(capacities[i]) * sizeof(nbr_t)); + } + if (nbr_file.BeginRewriteIfChanged()) { + for (size_t i = 0; i < vnum; ++i) { + nbr_file.WriteSegment(reinterpret_cast(adj_lists[i]), + static_cast(capacities[i]) * sizeof(nbr_t)); + } + } + descriptor.set_path(ModuleDescriptor::kNbrListPath, nbr_file.CommitOrReuse()); // Each internal buffer's path is stored as a named entry in the // descriptor's typed paths_ map. descriptor.set_path(ModuleDescriptor::kDegreeListPath, ckp.Commit(*degree_list_)); - - const nbr_t* const* adj_lists = - reinterpret_cast(adj_list_buffer_->GetData()); - const int* cap_arr = reinterpret_cast(cap_list_->GetData()); - - MD5_CTX ctx; - FileHeader header{}; - if (is_nbr_list_unmodified(ctx, header, nbr_list_.get(), adj_lists, cap_arr, - vnum)) { - // If the neighbor list is unmodified, we can reuse the existing file. - descriptor.set_path(ModuleDescriptor::kNbrListPath, - ckp.LinkToSnapshot(nbr_list_->GetPath())); - } else { - std::string nbr_path_committed; - - auto runtime_file = ckp.CreateRuntimeFile(); - const auto& nbr_path = runtime_file.path(); - std::ofstream nbr_out(nbr_path, std::ios::binary); - if (!nbr_out.is_open()) { - THROW_IO_EXCEPTION("Failed to open file for writing: " + nbr_path); - } - nbr_out.write(reinterpret_cast(&header), sizeof(header)); - for (size_t i = 0; i < vnum; ++i) { - const char* data = reinterpret_cast(adj_lists[i]); - size_t len = cap_arr[i] * sizeof(nbr_t); - nbr_out.write(data, len); - } - nbr_out.flush(); - nbr_out.close(); - nbr_path_committed = ckp.CommitRuntimeFile(std::move(runtime_file)); - - descriptor.set_path(ModuleDescriptor::kNbrListPath, nbr_path_committed); - } - descriptor.set_path(ModuleDescriptor::kCapacityListPath, ckp.Commit(*cap_list_)); meta.set_module(key, descriptor); @@ -171,44 +211,11 @@ void MutableCsr::Dump(Checkpoint& ckp, CheckpointManifest& meta, template void MutableCsr::compact() { - // Remove deleted edges and reset timestamps on surviving edges. - size_t vnum = vertex_capacity(); - auto** buf_arr = reinterpret_cast(adj_list_buffer_->GetData()); - auto* sz_arr = reinterpret_cast*>(degree_list_->GetData()); - size_t total_edge_num = 0; - for (size_t i = 0; i != vnum; ++i) { - int sz = sz_arr[i].load(std::memory_order_relaxed); - nbr_t* read_ptr = buf_arr[i]; - if (read_ptr == nullptr) { - continue; - } - nbr_t* read_end = read_ptr + sz; - nbr_t* write_ptr = read_ptr; - int removed = 0; - while (read_ptr != read_end) { - if (read_ptr->timestamp != INVALID_TIMESTAMP) { - if (removed) { - *write_ptr = *read_ptr; - } - write_ptr->timestamp.store(0, std::memory_order_relaxed); - ++write_ptr; - } else { - ++removed; - } - ++read_ptr; - } - sz_arr[i].store(sz - removed, std::memory_order_relaxed); - total_edge_num += (sz - removed); - } - if (total_edge_num != edge_num_.load()) { - LOG(WARNING) << "Inconsistent edge count after compaction" - << ": expected " << edge_num_.load() << ", actual " - << total_edge_num; - THROW_STORAGE_EXCEPTION( - "Inconsistent edge count after compaction: expected " + - std::to_string(edge_num_.load()) + ", actual " + - std::to_string(total_edge_num)); - } + auto** adj_lists = reinterpret_cast(adj_list_buffer_->GetData()); + auto* degrees = reinterpret_cast*>(degree_list_->GetData()); + size_t live_edges = parallel_normalize_mutable_adjacency_lists( + adj_lists, degrees, vertex_capacity()); + csr_dump::validate_edge_count(edge_num_.load(), live_edges); } template @@ -261,16 +268,25 @@ void MutableCsr::batch_sort_by_edge_data(timestamp_t ts) { size_t vnum = vertex_capacity(); auto** buf_arr = reinterpret_cast(adj_list_buffer_->GetData()); auto* sz_arr = reinterpret_cast*>(degree_list_->GetData()); - for (size_t i = 0; i != vnum; ++i) { - nbr_t* begin = buf_arr[i]; - if (begin == nullptr) { - continue; - } - int deg = sz_arr[i].load(std::memory_order_relaxed); - std::sort(begin, begin + deg, [](const nbr_t& lhs, const nbr_t& rhs) { - return lhs.data < rhs.data; - }); - } + // Each vertex owns a disjoint adjacency list, so sorting can be + // parallelized across vertex ranges without synchronization. + csr_parallel::parallel_for_sort_ranges( + vnum, sz_arr, [buf_arr, sz_arr](size_t first, size_t last) { + for (size_t i = first; i < last; ++i) { + nbr_t* begin = buf_arr[i]; + if (begin == nullptr) { + continue; + } + int deg = sz_arr[i].load(std::memory_order_relaxed); + if (deg <= 1) { + continue; + } + std::sort(begin, begin + deg, + [](const nbr_t& lhs, const nbr_t& rhs) { + return lhs.data < rhs.data; + }); + } + }); } unsorted_since_ = ts; } @@ -551,8 +567,20 @@ void SingleMutableCsr::Dump(Checkpoint& ckp, CheckpointManifest& meta, const std::string& key) { ModuleDescriptor descriptor; descriptor.module_type = ModuleTypeName(); - descriptor.set_path(ModuleDescriptor::kNbrListPath, ckp.Commit(*nbr_list_)); descriptor.set("edge_num", std::to_string(edge_num_.load())); + + nbr_t* data = reinterpret_cast(nbr_list_->GetData()); + size_t vnum = vertex_capacity(); + csr_dump::ChecksumReuseFileDumper nbr_file(ckp, nbr_list_.get()); + size_t live_edges = normalize_single_adjacency_list(data, vnum); + csr_dump::validate_edge_count(edge_num_.load(), live_edges); + nbr_file.AddSegment(reinterpret_cast(data), + vnum * sizeof(nbr_t)); + if (nbr_file.BeginRewriteIfChanged()) { + nbr_file.WriteSegment(reinterpret_cast(data), + vnum * sizeof(nbr_t)); + } + descriptor.set_path(ModuleDescriptor::kNbrListPath, nbr_file.CommitOrReuse()); meta.set_module(key, descriptor); } @@ -563,11 +591,7 @@ void SingleMutableCsr::compact() { } nbr_t* data = reinterpret_cast(nbr_list_->GetData()); size_t vnum = vertex_capacity(); - for (size_t i = 0; i != vnum; ++i) { - if (data[i].timestamp != INVALID_TIMESTAMP) { - data[i].timestamp.store(0, std::memory_order_relaxed); - } - } + normalize_single_adjacency_list(data, vnum); } template diff --git a/src/storages/graph/edge_table.cc b/src/storages/graph/edge_table.cc index a4bc15081..052208471 100644 --- a/src/storages/graph/edge_table.cc +++ b/src/storages/graph/edge_table.cc @@ -526,7 +526,12 @@ void EdgeTable::Close() { } void EdgeTable::SortByEdgeData(timestamp_t ts) { - // TODO + if (!meta_->is_bundled()) { + THROW_INVALID_ARGUMENT_EXCEPTION( + "sort key is not supported for unbundled edge table currently"); + } + out_csr_->batch_sort_by_edge_data(ts); + in_csr_->batch_sort_by_edge_data(ts); } void EdgeTable::BatchDeleteVertices(const std::set& src_set, @@ -875,12 +880,7 @@ void EdgeTable::Compact(const std::optional& sort_key_for_nbr) { out_csr_->compact(); in_csr_->compact(); if (sort_key_for_nbr.has_value()) { - if (!meta_->is_bundled()) { - THROW_INVALID_ARGUMENT_EXCEPTION( - "sort key is not supported for unbundled edge table currently"); - } - out_csr_->batch_sort_by_edge_data(1); - in_csr_->batch_sort_by_edge_data(1); + SortByEdgeData(1); } } diff --git a/src/storages/graph/property_graph.cc b/src/storages/graph/property_graph.cc index b66ca1e4e..7eaf4b33b 100644 --- a/src/storages/graph/property_graph.cc +++ b/src/storages/graph/property_graph.cc @@ -887,13 +887,17 @@ void PropertyGraph::compact_schema() { v_mutex_.resize(new_schema.vertex_label_frontier()); } -void PropertyGraph::Compact() { +void PropertyGraph::Compact() { compact_internal(true); } + +void PropertyGraph::compact_internal(bool compact_all_edge_tables) { /** * The compaction process includes two parts: * 1. Schema: remove the deleted properties and labels from * schema. * 2. Data: for each vertex and edge table, remove the deleted * data and compact the storage. + * Checkpoint compacts and sorts only edge tables with a sort key; other CSRs + * are normalized by Dump(). * * Assume concurrency is controlled by the caller. */ @@ -932,16 +936,23 @@ void PropertyGraph::Compact() { } const auto& sort_key_for_nbr = schema_.get_sort_key_for_nbr(src_label_i, dst_label_i, e_label_i); - edge_tables_.at(index).Compact(sort_key_for_nbr); + if (compact_all_edge_tables || sort_key_for_nbr.has_value()) { + edge_tables_.at(index).Compact(sort_key_for_nbr); + } } } } - LOG(INFO) << "Compaction completed."; + LOG(INFO) << (compact_all_edge_tables ? "Compaction" + : "Checkpoint preparation") + << " completed."; } void PropertyGraph::DumpAndClear(std::shared_ptr ckp) { LOG(INFO) << "Creating checkpoint at " << ckp->path(); + // Compact sort-key edge tables; Dump() normalizes the remaining CSRs. + compact_internal(false); + CheckpointManifest meta; ModuleBroker store; diff --git a/src/storages/loader/abstract_property_graph_loader.cc b/src/storages/loader/abstract_property_graph_loader.cc index f7c21bb75..43a098189 100644 --- a/src/storages/loader/abstract_property_graph_loader.cc +++ b/src/storages/loader/abstract_property_graph_loader.cc @@ -185,7 +185,7 @@ result AbstractPropertyGraphLoader::LoadFragment() { try { loadVertices(); loadEdges(); - graph_.Compact(); + // Compact graph inside DumpAndClear graph_.DumpAndClear(staging_checkpoint_->checkpoint()); staging_checkpoint_->Commit(); diff --git a/tests/storage/test_immutable_csr.cc b/tests/storage/test_immutable_csr.cc index 537424a6a..f0af65b35 100644 --- a/tests/storage/test_immutable_csr.cc +++ b/tests/storage/test_immutable_csr.cc @@ -18,6 +18,7 @@ #include #include #include "neug/storages/checkpoint_manager.h" +#include "neug/storages/container/file_header.h" #include "neug/storages/csr/csr_view_utils.h" #include "neug/storages/csr/immutable_csr.h" #include "unittest/utils.h" @@ -87,126 +88,26 @@ class IMMutableCsrTest : public ::testing::Test { } bool check_edge_data_ordered(CsrView& generic_view) { - for (vid_t v = 0; v < 500; v++) { - if constexpr (std::is_same_v) { - NbrList nbr_list = generic_view.get_edges(0); - int32_t cur_value = - *static_cast(nbr_list.begin().get_data_ptr()); - for (auto nbr = ++nbr_list.begin(); nbr != nbr_list.end(); ++nbr) { - int32_t next_value = *static_cast(nbr.get_data_ptr()); - if (next_value < cur_value) { - return false; - } else { - cur_value = next_value; - } - } - } else if constexpr (std::is_same_v) { - NbrList nbr_list = generic_view.get_edges(0); - int64_t cur_value = - *static_cast(nbr_list.begin().get_data_ptr()); - for (auto nbr = ++nbr_list.begin(); nbr != nbr_list.end(); ++nbr) { - int64_t next_value = *static_cast(nbr.get_data_ptr()); - if (next_value < cur_value) { - return false; - } else { - cur_value = next_value; - } - } - } else if constexpr (std::is_same_v) { - NbrList nbr_list = generic_view.get_edges(0); - uint32_t cur_value = - *static_cast(nbr_list.begin().get_data_ptr()); - for (auto nbr = ++nbr_list.begin(); nbr != nbr_list.end(); ++nbr) { - uint32_t next_value = - *static_cast(nbr.get_data_ptr()); - if (next_value < cur_value) { - return false; - } else { - cur_value = next_value; - } - } - } else if constexpr (std::is_same_v) { - NbrList nbr_list = generic_view.get_edges(0); - uint64_t cur_value = - *static_cast(nbr_list.begin().get_data_ptr()); - for (auto nbr = ++nbr_list.begin(); nbr != nbr_list.end(); ++nbr) { - uint64_t next_value = - *static_cast(nbr.get_data_ptr()); - if (next_value < cur_value) { - return false; - } else { - cur_value = next_value; - } - } - } else if constexpr (std::is_same_v) { - NbrList nbr_list = generic_view.get_edges(0); - float cur_value = - *static_cast(nbr_list.begin().get_data_ptr()); - for (auto nbr = ++nbr_list.begin(); nbr != nbr_list.end(); ++nbr) { - float next_value = *static_cast(nbr.get_data_ptr()); - if (next_value < cur_value) { - return false; - } else { - cur_value = next_value; - } - } - } else if constexpr (std::is_same_v) { - NbrList nbr_list = generic_view.get_edges(0); - double cur_value = - *static_cast(nbr_list.begin().get_data_ptr()); - for (auto nbr = ++nbr_list.begin(); nbr != nbr_list.end(); ++nbr) { - double next_value = *static_cast(nbr.get_data_ptr()); - if (next_value < cur_value) { - return false; - } else { - cur_value = next_value; - } - } - } else if constexpr (std::is_same_v) { - NbrList nbr_list = generic_view.get_edges(0); - Date cur_value = - *static_cast(nbr_list.begin().get_data_ptr()); - for (auto nbr = ++nbr_list.begin(); nbr != nbr_list.end(); ++nbr) { - Date next_value = *static_cast(nbr.get_data_ptr()); - if (next_value < cur_value) { - return false; - } else { - cur_value = next_value; - } - } - } else if constexpr (std::is_same_v) { - NbrList nbr_list = generic_view.get_edges(0); - DateTime cur_value = - *static_cast(nbr_list.begin().get_data_ptr()); - for (auto nbr = ++nbr_list.begin(); nbr != nbr_list.end(); ++nbr) { - DateTime next_value = - *static_cast(nbr.get_data_ptr()); - if (next_value < cur_value) { - return false; - } else { - cur_value = next_value; - } + if constexpr (std::is_same_v) { + return true; + } else { + for (vid_t v = 0; v < 500; ++v) { + auto nbr_list = generic_view.get_edges(v); + auto nbr = nbr_list.begin(); + if (nbr == nbr_list.end()) { + continue; } - } else if constexpr (std::is_same_v) { - NbrList nbr_list = generic_view.get_edges(0); - Interval cur_value = - *static_cast(nbr_list.begin().get_data_ptr()); - for (auto nbr = ++nbr_list.begin(); nbr != nbr_list.end(); ++nbr) { - Interval next_value = - *static_cast(nbr.get_data_ptr()); - if (next_value < cur_value) { + EDATA_T current = *static_cast(nbr.get_data_ptr()); + for (++nbr; nbr != nbr_list.end(); ++nbr) { + EDATA_T next = *static_cast(nbr.get_data_ptr()); + if (next < current) { return false; - } else { - cur_value = next_value; } + current = next; } - } else if constexpr (std::is_same_v) { - continue; - } else { - return false; } + return true; } - return true; } neug::CheckpointManager& Workspace() { return ws; } @@ -307,6 +208,76 @@ TYPED_TEST(IMMutableCsrTest, TestDumpAndOpen) { EXPECT_EQ(hugepage_single_immutable_csr.edge_num(), 500); } +TYPED_TEST(IMMutableCsrTest, TestDumpCompactsDeletedEdges) { + ImmutableCsr csr; + auto ckp = make_checkpoint(this->Workspace()); + csr.Open(*ckp, ModuleDescriptor(), MemoryLevel::kInMemory); + csr.resize(2); + csr.batch_put_edges({0, 0, 1}, {10, 11, 12}, std::vector(3), 0); + csr.delete_edge(0, 0, 0); + + auto desc = dump_module_descriptor(csr, *ckp, "compacted"); + EXPECT_EQ(std::filesystem::file_size( + desc.get_path(ModuleDescriptor::kNbrListPath).value()), + sizeof(FileHeader) + 2 * sizeof(ImmutableNbr)); + ImmutableCsr reopened; + reopened.Open(*ckp, desc, MemoryLevel::kInMemory); + + auto view = reopened.get_generic_view(MAX_TIMESTAMP); + auto edges = view.get_edges(0); + auto it = edges.begin(); + ASSERT_NE(it, edges.end()); + EXPECT_EQ(it.get_vertex(), 11); + EXPECT_EQ(++it, edges.end()); + EXPECT_EQ(view.get_edges(1).begin().get_vertex(), 12); + EXPECT_EQ(reopened.edge_num(), 2); +} + +TYPED_TEST(IMMutableCsrTest, TestCleanDumpReusesFiles) { + ImmutableCsr csr; + auto ckp = this->load_csr_data(csr); + auto original = dump_module_descriptor(csr, *ckp, "original"); + + ImmutableCsr reopened; + reopened.Open(*ckp, original, MemoryLevel::kInMemory); + auto next_ckp = make_checkpoint(this->Workspace()); + auto reused = dump_module_descriptor(reopened, *next_ckp, "reused"); + + EXPECT_TRUE(std::filesystem::equivalent( + original.get_path(ModuleDescriptor::kNbrListPath).value(), + reused.get_path(ModuleDescriptor::kNbrListPath).value())); + EXPECT_TRUE(std::filesystem::equivalent( + original.get_path(ModuleDescriptor::kDegreeListPath).value(), + reused.get_path(ModuleDescriptor::kDegreeListPath).value())); +} + +TYPED_TEST(IMMutableCsrTest, TestDirtyReopenDump) { + ImmutableCsr csr; + auto ckp = make_checkpoint(this->Workspace()); + csr.Open(*ckp, ModuleDescriptor(), MemoryLevel::kInMemory); + csr.resize(2); + csr.batch_put_edges({0, 0, 1}, {10, 11, 12}, std::vector(3), 0); + auto original = dump_module_descriptor(csr, *ckp, "original"); + + for (auto level : {MemoryLevel::kInMemory, MemoryLevel::kHugePagePreferred, + MemoryLevel::kSyncToFile}) { + ImmutableCsr dirty; + dirty.Open(*ckp, original, level); + dirty.delete_edge(0, 0, 0); + auto next_ckp = make_checkpoint(this->Workspace()); + auto compacted = dump_module_descriptor(dirty, *next_ckp, "compacted"); + + ImmutableCsr reopened; + reopened.Open(*next_ckp, compacted, MemoryLevel::kInMemory); + auto edges = reopened.get_generic_view(MAX_TIMESTAMP).get_edges(0); + auto it = edges.begin(); + ASSERT_NE(it, edges.end()); + EXPECT_EQ(it.get_vertex(), 11); + EXPECT_EQ(++it, edges.end()); + EXPECT_EQ(reopened.edge_num(), 2); + } +} + TYPED_TEST(IMMutableCsrTest, TestResize) { ImmutableCsr immutable_csr; this->load_csr_data(immutable_csr); diff --git a/tests/storage/test_mutable_csr.cc b/tests/storage/test_mutable_csr.cc index 8ab438da3..db33b06f6 100644 --- a/tests/storage/test_mutable_csr.cc +++ b/tests/storage/test_mutable_csr.cc @@ -1024,6 +1024,149 @@ TEST_F(MutableCsrDumpDirtyTest, VariousMutationsSetDirty) { expect_slow_after("batch_delete_edges", [](CsrT& c) { c.batch_delete_edges({0}, {3}); }); } + +TEST_F(MutableCsrDumpDirtyTest, DumpNormalizesTimestampsAndRemovesTombstones) { + CsrT csr; + ModuleDescriptor original_desc; + auto ckp = prepare(csr, original_desc); + + csr.put_edge(0, 2, 999, 7, *alloc_); + csr.delete_edge(0, 0, 8); + ASSERT_EQ(csr.edge_num(), src_.size()); + + auto normalized_desc = dump_module_descriptor(csr, *ckp, "normalized"); + CsrT reopened; + reopened.Open(*ckp, normalized_desc, MemoryLevel::kInMemory); + + size_t actual_edge_num = 0; + std::vector src_zero_neighbors; + auto view = reopened.get_generic_view(MAX_TIMESTAMP); + for (vid_t src = 0; src < VNUM; ++src) { + auto edges = view.get_edges(src); + for (auto it = edges.begin(); it != edges.end(); ++it) { + EXPECT_EQ(it.get_timestamp(), 0); + ++actual_edge_num; + if (src == 0) { + src_zero_neighbors.push_back(it.get_vertex()); + } + } + } + EXPECT_EQ(actual_edge_num, src_.size()); + EXPECT_EQ(reopened.edge_num(), src_.size()); + EXPECT_EQ(src_zero_neighbors, (std::vector{4, 2})); +} + +TEST_F(MutableCsrDumpDirtyTest, DumpCompactsMiddleAndTailDeletions) { + CsrT csr; + auto ckp = make_checkpoint(checkpoint_mgr_); + csr.Open(*ckp, ModuleDescriptor(), MemoryLevel::kInMemory); + csr.resize(1); + csr.batch_put_edges({0, 0, 0, 0}, {10, 11, 12, 13}, {1, 2, 3, 4}); + csr.delete_edge(0, 1, 1); + csr.delete_edge(0, 3, 2); + ASSERT_EQ(csr.edge_num(), 2); + + auto desc = dump_module_descriptor(csr, *ckp, "compacted"); + CsrT reopened; + reopened.Open(*ckp, desc, MemoryLevel::kInMemory); + + std::vector neighbors; + auto edges = reopened.get_generic_view(MAX_TIMESTAMP).get_edges(0); + for (auto it = edges.begin(); it != edges.end(); ++it) { + EXPECT_EQ(it.get_timestamp(), 0); + neighbors.push_back(it.get_vertex()); + } + EXPECT_EQ(neighbors, (std::vector{10, 12})); + EXPECT_EQ(reopened.edge_num(), 2); +} + +class SingleMutableCsrDumpTest : public ::testing::Test { + protected: + using CsrT = SingleMutableCsr; + static constexpr vid_t VNUM = 4; + + void SetUp() override { + test_dir_ = make_unique_test_dir("single_mutable_csr_dump_test"); + checkpoint_mgr_.Open(test_dir_.string()); + alloc_ = std::make_unique(MemoryLevel::kInMemory, ""); + } + + void TearDown() override { + checkpoint_mgr_.Close(); + if (std::filesystem::exists(test_dir_)) { + std::filesystem::remove_all(test_dir_); + } + } + + std::shared_ptr prepare(CsrT& csr, ModuleDescriptor& desc) { + CsrT original; + auto ckp = make_checkpoint(checkpoint_mgr_); + original.Open(*ckp, ModuleDescriptor(), MemoryLevel::kInMemory); + original.resize(VNUM); + original.batch_put_edges({0, 1, 3}, {10, 11, 13}, {100, 110, 130}); + desc = dump_module_descriptor(original, *ckp, "original"); + csr.Open(*ckp, desc, MemoryLevel::kInMemory); + return ckp; + } + + static ino_t inode_of(const std::string& path) { + struct stat st {}; + if (stat(path.c_str(), &st) != 0) { + throw std::runtime_error("stat() failed for path: " + path + " — " + + std::strerror(errno)); + } + return st.st_ino; + } + + static std::string nbr_path(const ModuleDescriptor& desc) { + return desc.get_path(ModuleDescriptor::kNbrListPath).value(); + } + + std::filesystem::path test_dir_; + CheckpointManager checkpoint_mgr_; + std::unique_ptr alloc_; +}; + +TEST_F(SingleMutableCsrDumpTest, CleanDumpReusesNeighborFile) { + CsrT csr; + ModuleDescriptor original_desc; + prepare(csr, original_desc); + auto original_inode = inode_of(nbr_path(original_desc)); + + auto next_ckp = make_checkpoint(checkpoint_mgr_); + auto reused_desc = dump_module_descriptor(csr, *next_ckp, "reused"); + EXPECT_EQ(original_inode, inode_of(nbr_path(reused_desc))); +} + +TEST_F(SingleMutableCsrDumpTest, + DumpNormalizesTimestampsAndPreservesEmptySlots) { + CsrT csr; + ModuleDescriptor original_desc; + auto ckp = prepare(csr, original_desc); + + csr.put_edge(2, 12, 120, 7, *alloc_); + csr.delete_edge(1, 0, 8); + ASSERT_EQ(csr.edge_num(), 3); + + auto normalized_desc = dump_module_descriptor(csr, *ckp, "normalized"); + EXPECT_NE(inode_of(nbr_path(original_desc)), + inode_of(nbr_path(normalized_desc))); + + CsrT reopened; + reopened.Open(*ckp, normalized_desc, MemoryLevel::kInMemory); + auto view = reopened.get_generic_view(MAX_TIMESTAMP); + std::vector visible_sources; + for (vid_t src = 0; src < VNUM; ++src) { + auto edges = view.get_edges(src); + for (auto it = edges.begin(); it != edges.end(); ++it) { + EXPECT_EQ(it.get_timestamp(), 0); + visible_sources.push_back(src); + } + } + EXPECT_EQ(visible_sources, (std::vector{0, 2, 3})); + EXPECT_EQ(reopened.edge_num(), 3); +} + // Concurrent read-write test: verifies that lock-free readers using // get_edges() / foreach_nbr_lt() see consistent (degree, buffer) snapshots // even when a concurrent writer triggers CSR buffer reallocation via put_edge. diff --git a/tests/storage/test_open_graph.cc b/tests/storage/test_open_graph.cc index ea8b653cc..c86458a84 100644 --- a/tests/storage/test_open_graph.cc +++ b/tests/storage/test_open_graph.cc @@ -261,6 +261,69 @@ TEST(DatabaseTest, TestPersist) { } } +TEST(DatabaseTest, TestCopyInitialLoadCheckpointReopen) { + auto test_dir = unique_test_dir("test_copy_initial_load_checkpoint_reopen"); + auto db_dir = (test_dir / "db").string(); + auto csv_dir = (test_dir / "csv").string(); + if (std::filesystem::exists(test_dir)) { + std::filesystem::remove_all(test_dir); + } + std::filesystem::create_directories(csv_dir); + auto person_csv_path = + (std::filesystem::path(csv_dir) / "person.csv").string(); + auto knows_csv_path = + (std::filesystem::path(csv_dir) / "person_knows_person.csv").string(); + { + std::ofstream person_csv(person_csv_path); + ASSERT_TRUE(person_csv.good()); + person_csv << "id|name|age\n"; + person_csv << "1|Alice|30\n"; + person_csv << "2|Bob|31\n"; + person_csv << "3|Cora|32\n"; + person_csv << "4|Dan|33\n"; + } + { + std::ofstream knows_csv(knows_csv_path); + ASSERT_TRUE(knows_csv.good()); + knows_csv << "from|to|weight\n"; + knows_csv << "1|2|0.5\n"; + knows_csv << "2|3|1.5\n"; + } + + { + neug::NeugDB db; + db.Open(db_dir, 1, neug::DBMode::READ_WRITE, "gopt", false); + auto conn = db.Connect(); + EXPECT_TRUE(conn->Query( + "CREATE NODE TABLE person(id INT64, name STRING, age INT64, " + "PRIMARY KEY(id));")); + EXPECT_TRUE(conn->Query( + "CREATE REL TABLE knows(FROM person TO person, weight DOUBLE);")); + EXPECT_TRUE(conn->Query("COPY person from \"" + person_csv_path + "\";")); + EXPECT_TRUE(conn->Query("COPY knows from \"" + knows_csv_path + + "\" (from=\"person\", to=\"person\");")); + EXPECT_TRUE(conn->Query("CHECKPOINT;")); + conn->Close(); + db.Close(); + } + + { + neug::NeugDB db; + db.Open(db_dir, 1, neug::DBMode::READ_ONLY); + auto conn = db.Connect(); + auto vertex_count = conn->Query("MATCH (n: person) return COUNT(n);"); + EXPECT_TRUE(vertex_count); + neug::test::AssertInt64Column(vertex_count.value().response(), 0, {4}); + auto edge_count = conn->Query( + "MATCH (a: person)-[r: knows]->(b: person) return COUNT(r);"); + EXPECT_TRUE(edge_count); + neug::test::AssertInt64Column(edge_count.value().response(), 0, {2}); + conn->Close(); + db.Close(); + } + std::filesystem::remove_all(test_dir); +} + TEST(DatabaseTest, TestCompaction) { std::string db_dir = "/tmp/test_compaction"; { diff --git a/tests/storage/test_property_graph.cc b/tests/storage/test_property_graph.cc index cd087e08a..0c0396e96 100644 --- a/tests/storage/test_property_graph.cc +++ b/tests/storage/test_property_graph.cc @@ -144,4 +144,4 @@ TEST_F(PropertyGraphTest, TestOpenAndBulkInsert) { } } -} // namespace neug \ No newline at end of file +} // namespace neug From a1015c9fce83a4b294eb55349cd023d854513913 Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Wed, 29 Jul 2026 17:47:05 +0800 Subject: [PATCH 2/2] resolve comments --- include/neug/transaction/update_transaction.h | 4 +- src/storages/csr/csr_parallel_utils.h | 9 ++- src/storages/csr/immutable_csr.cc | 12 ++++ src/storages/graph/property_graph.cc | 7 +- src/transaction/update_transaction.cc | 8 +++ tests/storage/test_immutable_csr.cc | 58 ++++++++------- tests/storage/test_mutable_csr.cc | 42 ++--------- tests/storage/test_property_graph.cc | 70 +++++++++++++++++++ tests/transaction/test_update_transaction.cc | 64 +++++++++++++++++ 9 files changed, 211 insertions(+), 63 deletions(-) diff --git a/include/neug/transaction/update_transaction.h b/include/neug/transaction/update_transaction.h index d4a234427..17e22fd72 100644 --- a/include/neug/transaction/update_transaction.h +++ b/include/neug/transaction/update_transaction.h @@ -177,7 +177,8 @@ class StorageTPUpdateInterface : public StorageUpdateInterface { mut_view_(txn.view_), alloc_(txn.alloc_), ckp_(txn.ckp_), - wal_(txn.wal_builder_) {} + wal_(txn.wal_builder_), + vm_(txn.vm_) {} ~StorageTPUpdateInterface() = default; void CreateCheckpoint() override; @@ -273,6 +274,7 @@ class StorageTPUpdateInterface : public StorageUpdateInterface { Allocator& alloc_; std::shared_ptr& ckp_; WalBuilder& wal_; + IVersionManager& vm_; }; } // namespace neug diff --git a/src/storages/csr/csr_parallel_utils.h b/src/storages/csr/csr_parallel_utils.h index 1e02b5beb..9a3ca9ac1 100644 --- a/src/storages/csr/csr_parallel_utils.h +++ b/src/storages/csr/csr_parallel_utils.h @@ -19,6 +19,7 @@ #include #include #include +#include #include #include @@ -109,7 +110,13 @@ void parallel_for_degree_ranges(size_t total, const DEGREE_T* degrees, std::vector threads; threads.reserve(workers - 1); for (size_t i = 1; i < workers; ++i) { - threads.emplace_back(work); + try { + threads.emplace_back(work); + } catch (const std::system_error&) { + // Handle thread creation failure before joinable threads can unwind. + // The workers already created plus the caller finish the remaining work. + break; + } } work(); for (auto& t : threads) { diff --git a/src/storages/csr/immutable_csr.cc b/src/storages/csr/immutable_csr.cc index 400ff94fe..bda03a37e 100644 --- a/src/storages/csr/immutable_csr.cc +++ b/src/storages/csr/immutable_csr.cc @@ -136,6 +136,18 @@ void ImmutableCsr::Dump(Checkpoint& ckp, CheckpointManifest& meta, } desc.set_path(ModuleDescriptor::kNbrListPath, nbr_file.CommitOrReuse()); + // In-place normalization shrank the live data to the buffer head; restore + // the GetDataSize() == sum(degrees) * sizeof(nbr_t) invariant, matching + // compact(), so the object stays usable (e.g. batch_put_edges) after Dump. + // Resize must run after CommitOrReuse because it clears the container's + // mapped path used for reuse decisions. + nbr_list_buffer_->Resize(live_edges * sizeof(nbr_t)); + auto* segment = reinterpret_cast(nbr_list_buffer_->GetData()); + for (size_t i = 0; i < vnum; ++i) { + adj_lists[i] = segment; + segment += degrees[i]; + } + csr_dump::ChecksumReuseFileDumper degree_file(ckp, degree_list_buffer_.get()); auto* degree_data = reinterpret_cast(degree_list_buffer_->GetData()); diff --git a/src/storages/graph/property_graph.cc b/src/storages/graph/property_graph.cc index 7eaf4b33b..4f1be020b 100644 --- a/src/storages/graph/property_graph.cc +++ b/src/storages/graph/property_graph.cc @@ -950,7 +950,12 @@ void PropertyGraph::compact_internal(bool compact_all_edge_tables) { void PropertyGraph::DumpAndClear(std::shared_ptr ckp) { LOG(INFO) << "Creating checkpoint at " << ckp->path(); - // Compact sort-key edge tables; Dump() normalizes the remaining CSRs. + // Compact sort-key edge tables; Dump() normalizes the remaining CSRs in + // place. CALLER CONTRACT: no concurrent readers may access this graph's + // CSR buffers during DumpAndClear — TP enforces it via + // begin_update_commit + drain_readers (StorageTPUpdateInterface:: + // CreateCheckpoint), AP via TimestampLease::makeUpdateExclusive, and + // NeugDB admin paths run with no live connections. compact_internal(false); CheckpointManifest meta; diff --git a/src/transaction/update_transaction.cc b/src/transaction/update_transaction.cc index a43f0e315..4e479042f 100644 --- a/src/transaction/update_transaction.cc +++ b/src/transaction/update_transaction.cc @@ -1452,6 +1452,14 @@ void StorageTPUpdateInterface::CreateCheckpoint() { } auto ckp = cow_graph_->checkpoint_ptr(); auto memory_level = cow_graph_->memory_level(); + // Dump() normalizes CSR buffers in place, and cow_graph_ shares those + // buffers with the published snapshot that concurrent readers may still + // hold (MutableCsr::Detach does not deep-copy nbr_list_). Block new + // readers and drain in-flight ones before touching shared data, as the + // AP path already does via TimestampLease::makeUpdateExclusive(). + // begin_update_commit is idempotent: Commit() re-enters the same state. + vm_.begin_update_commit(read_ts_); + vm_.drain_readers(); cow_graph_->DumpAndClear(ckp); cow_graph_->Open(ckp, memory_level); mut_view_.Rebuild(*cow_graph_); diff --git a/tests/storage/test_immutable_csr.cc b/tests/storage/test_immutable_csr.cc index f0af65b35..9a17b52df 100644 --- a/tests/storage/test_immutable_csr.cc +++ b/tests/storage/test_immutable_csr.cc @@ -208,31 +208,6 @@ TYPED_TEST(IMMutableCsrTest, TestDumpAndOpen) { EXPECT_EQ(hugepage_single_immutable_csr.edge_num(), 500); } -TYPED_TEST(IMMutableCsrTest, TestDumpCompactsDeletedEdges) { - ImmutableCsr csr; - auto ckp = make_checkpoint(this->Workspace()); - csr.Open(*ckp, ModuleDescriptor(), MemoryLevel::kInMemory); - csr.resize(2); - csr.batch_put_edges({0, 0, 1}, {10, 11, 12}, std::vector(3), 0); - csr.delete_edge(0, 0, 0); - - auto desc = dump_module_descriptor(csr, *ckp, "compacted"); - EXPECT_EQ(std::filesystem::file_size( - desc.get_path(ModuleDescriptor::kNbrListPath).value()), - sizeof(FileHeader) + 2 * sizeof(ImmutableNbr)); - ImmutableCsr reopened; - reopened.Open(*ckp, desc, MemoryLevel::kInMemory); - - auto view = reopened.get_generic_view(MAX_TIMESTAMP); - auto edges = view.get_edges(0); - auto it = edges.begin(); - ASSERT_NE(it, edges.end()); - EXPECT_EQ(it.get_vertex(), 11); - EXPECT_EQ(++it, edges.end()); - EXPECT_EQ(view.get_edges(1).begin().get_vertex(), 12); - EXPECT_EQ(reopened.edge_num(), 2); -} - TYPED_TEST(IMMutableCsrTest, TestCleanDumpReusesFiles) { ImmutableCsr csr; auto ckp = this->load_csr_data(csr); @@ -266,6 +241,9 @@ TYPED_TEST(IMMutableCsrTest, TestDirtyReopenDump) { dirty.delete_edge(0, 0, 0); auto next_ckp = make_checkpoint(this->Workspace()); auto compacted = dump_module_descriptor(dirty, *next_ckp, "compacted"); + EXPECT_EQ(std::filesystem::file_size( + compacted.get_path(ModuleDescriptor::kNbrListPath).value()), + sizeof(FileHeader) + 2 * sizeof(ImmutableNbr)); ImmutableCsr reopened; reopened.Open(*next_ckp, compacted, MemoryLevel::kInMemory); @@ -278,6 +256,36 @@ TYPED_TEST(IMMutableCsrTest, TestDirtyReopenDump) { } } +TYPED_TEST(IMMutableCsrTest, TestDumpThenContinueWriting) { + ImmutableCsr csr; + auto ckp = make_checkpoint(this->Workspace()); + csr.Open(*ckp, ModuleDescriptor(), MemoryLevel::kInMemory); + csr.resize(2); + csr.batch_put_edges({0, 0, 1}, {10, 11, 12}, std::vector(3), 0); + // Tombstone the first edge so Dump() compacts the buffer in place and + // shrinks the live region. + csr.delete_edge(0, 0, 0); + dump_module_descriptor(csr, *ckp, "csr"); + + // Dump() must restore the GetDataSize() == sum(degrees) invariant; + // further writes have to append after the surviving edges instead of + // using the stale pre-compaction size and clobbering live data. + csr.batch_put_edges({0, 1}, {13, 14}, std::vector(2), 1); + EXPECT_EQ(csr.edge_num(), 4); + + auto view = csr.get_generic_view(MAX_TIMESTAMP); + auto collect = [&view](vid_t v) { + std::vector nbrs; + auto edges = view.get_edges(v); + for (auto it = edges.begin(); it != edges.end(); ++it) { + nbrs.push_back(it.get_vertex()); + } + return nbrs; + }; + EXPECT_EQ(collect(0), (std::vector{11, 13})); + EXPECT_EQ(collect(1), (std::vector{12, 14})); +} + TYPED_TEST(IMMutableCsrTest, TestResize) { ImmutableCsr immutable_csr; this->load_csr_data(immutable_csr); diff --git a/tests/storage/test_mutable_csr.cc b/tests/storage/test_mutable_csr.cc index db33b06f6..d010dfe01 100644 --- a/tests/storage/test_mutable_csr.cc +++ b/tests/storage/test_mutable_csr.cc @@ -16,7 +16,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -1026,47 +1028,17 @@ TEST_F(MutableCsrDumpDirtyTest, VariousMutationsSetDirty) { } TEST_F(MutableCsrDumpDirtyTest, DumpNormalizesTimestampsAndRemovesTombstones) { - CsrT csr; - ModuleDescriptor original_desc; - auto ckp = prepare(csr, original_desc); - - csr.put_edge(0, 2, 999, 7, *alloc_); - csr.delete_edge(0, 0, 8); - ASSERT_EQ(csr.edge_num(), src_.size()); - - auto normalized_desc = dump_module_descriptor(csr, *ckp, "normalized"); - CsrT reopened; - reopened.Open(*ckp, normalized_desc, MemoryLevel::kInMemory); - - size_t actual_edge_num = 0; - std::vector src_zero_neighbors; - auto view = reopened.get_generic_view(MAX_TIMESTAMP); - for (vid_t src = 0; src < VNUM; ++src) { - auto edges = view.get_edges(src); - for (auto it = edges.begin(); it != edges.end(); ++it) { - EXPECT_EQ(it.get_timestamp(), 0); - ++actual_edge_num; - if (src == 0) { - src_zero_neighbors.push_back(it.get_vertex()); - } - } - } - EXPECT_EQ(actual_edge_num, src_.size()); - EXPECT_EQ(reopened.edge_num(), src_.size()); - EXPECT_EQ(src_zero_neighbors, (std::vector{4, 2})); -} - -TEST_F(MutableCsrDumpDirtyTest, DumpCompactsMiddleAndTailDeletions) { CsrT csr; auto ckp = make_checkpoint(checkpoint_mgr_); csr.Open(*ckp, ModuleDescriptor(), MemoryLevel::kInMemory); csr.resize(1); csr.batch_put_edges({0, 0, 0, 0}, {10, 11, 12, 13}, {1, 2, 3, 4}); + csr.put_edge(0, 14, 5, 7, *alloc_); csr.delete_edge(0, 1, 1); csr.delete_edge(0, 3, 2); - ASSERT_EQ(csr.edge_num(), 2); + ASSERT_EQ(csr.edge_num(), 3); - auto desc = dump_module_descriptor(csr, *ckp, "compacted"); + auto desc = dump_module_descriptor(csr, *ckp, "normalized"); CsrT reopened; reopened.Open(*ckp, desc, MemoryLevel::kInMemory); @@ -1076,8 +1048,8 @@ TEST_F(MutableCsrDumpDirtyTest, DumpCompactsMiddleAndTailDeletions) { EXPECT_EQ(it.get_timestamp(), 0); neighbors.push_back(it.get_vertex()); } - EXPECT_EQ(neighbors, (std::vector{10, 12})); - EXPECT_EQ(reopened.edge_num(), 2); + EXPECT_EQ(neighbors, (std::vector{10, 12, 14})); + EXPECT_EQ(reopened.edge_num(), 3); } class SingleMutableCsrDumpTest : public ::testing::Test { diff --git a/tests/storage/test_property_graph.cc b/tests/storage/test_property_graph.cc index 0c0396e96..6530d3fec 100644 --- a/tests/storage/test_property_graph.cc +++ b/tests/storage/test_property_graph.cc @@ -15,6 +15,10 @@ #include +#include +#include +#include + #include "neug/common/types/value.h" #include "neug/storages/checkpoint_manager.h" #include "neug/storages/graph/property_graph.h" @@ -144,4 +148,70 @@ TEST_F(PropertyGraphTest, TestOpenAndBulkInsert) { } } +TEST_F(PropertyGraphTest, CheckpointSortsEdgesByConfiguredKey) { + CreateVertexTypeParamBuilder vertex_builder; + ASSERT_TRUE(graph_ + ->CreateVertexType(vertex_builder.VertexLabel("person") + .AddProperty("id", Value::INT64(0)) + .AddPrimaryKeyName("id") + .Build()) + .ok()); + CreateEdgeTypeParamBuilder edge_builder; + ASSERT_TRUE(graph_ + ->CreateEdgeType( + edge_builder.SrcLabel("person") + .DstLabel("person") + .EdgeLabel("knows") + .AddProperty("weight", Value::INT64(0)) + .SortKeyForNbr(std::optional{"weight"}) + .Build()) + .ok()); + + const label_t person = graph_->schema().get_vertex_label_id("person"); + const label_t knows = graph_->schema().get_edge_label_id("knows"); + std::vector vertices; + for (int64_t id = 0; id != 4; ++id) { + vid_t vertex; + ASSERT_TRUE( + graph_->AddVertex(person, Value::INT64(id), {}, vertex, 0).ok()); + vertices.push_back(vertex); + } + + Allocator allocator(MemoryLevel::kInMemory, ""); + for (const auto& [dst, weight] : std::vector>{ + {vertices[1], 30}, {vertices[2], 10}, {vertices[3], 20}}) { + int32_t offset; + const void* property = nullptr; + ASSERT_TRUE(graph_ + ->AddEdge(person, vertices[0], person, dst, knows, + {Value::INT64(weight)}, 7, allocator, offset, + property) + .ok()); + } + graph_->MarkVertexTableDirty(person); + graph_->MarkEdgeTableDirty(person, person, knows); + + auto checkpoint = make_checkpoint(checkpoint_mgr_); + graph_->DumpAndClear(checkpoint); + + std::vector neighbors; + std::vector weights; + PropertyGraph reopened; + reopened.Open(checkpoint, MemoryLevel::kInMemory); + auto view = reopened.get_edge_table(person, person, knows) + .get_outgoing_view(MAX_TIMESTAMP); + ASSERT_EQ(view.type(), CsrViewType::kMultipleMutable); + auto typed_view = + view.get_typed_view(); + EXPECT_EQ(typed_view.unsorted_since, 1); + auto edges = view.get_edges(vertices[0]); + for (auto it = edges.begin(); it != edges.end(); ++it) { + neighbors.push_back(it.get_vertex()); + weights.push_back(*static_cast(it.get_data_ptr())); + } + EXPECT_EQ(neighbors, + (std::vector{vertices[2], vertices[3], vertices[1]})); + EXPECT_EQ(weights, (std::vector{10, 20, 30})); +} + } // namespace neug diff --git a/tests/transaction/test_update_transaction.cc b/tests/transaction/test_update_transaction.cc index e5c6f1356..212e343e1 100644 --- a/tests/transaction/test_update_transaction.cc +++ b/tests/transaction/test_update_transaction.cc @@ -21,7 +21,10 @@ #include "neug/storages/graph/graph_interface.h" #include "neug/transaction/update_transaction.h" +#include +#include #include +#include #include "column_assertions.h" #include "glog/logging.h" @@ -2711,6 +2714,67 @@ TEST_F(UpdateTransactionTest, TestCheckpoint) { } } +// CreateCheckpoint() normalizes CSR buffers in place while cow_graph_ +// shares them with the published snapshot, so it must block new readers and +// drain in-flight ones before dumping. +TEST_F(UpdateTransactionTest, TestCheckpointDrainsReaders) { + neug::NeugDB db; + neug::NeugDBConfig config(db_dir); + config.memory_level = neug::MemoryLevel::kInMemory; + db.Open(config); + auto svc = std::make_shared(db); + + // Dirty the graph so CreateCheckpoint() takes the dump path instead of + // the no-modification early return. + { + auto slot = svc->AcquireExecutionSlot(); + auto txn = slot->GetUpdateTransaction(); + neug::StorageTPUpdateInterface gui(txn); + auto person_label = txn.schema().get_vertex_label_id("person"); + neug::vid_t vid; + EXPECT_TRUE(gui.AddVertex( + person_label, neug::Value::INT64(3), + {neug::Value::STRING(std::string("Eve")), neug::Value::INT64(28)}, + vid)); + EXPECT_TRUE(txn.Commit()); + } + + std::atomic reader_holding{false}; + std::atomic checkpoint_returned{false}; + + std::thread reader([&]() { + auto slot = svc->AcquireExecutionSlot(); + auto txn = slot->GetReadTransaction(); + reader_holding.store(true, std::memory_order_release); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + }); // The read transaction dtor releases the read timestamp. + + while (!reader_holding.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + + std::thread checkpoint_thread([&]() { + auto slot = svc->AcquireExecutionSlot(); + auto txn = slot->GetUpdateTransaction(); + neug::StorageTPUpdateInterface gui(txn); + gui.CreateCheckpoint(); + checkpoint_returned.store(true, std::memory_order_release); + EXPECT_TRUE(txn.Commit()); + }); + + // While the reader holds its snapshot, the checkpoint must stay blocked + // in drain_readers(); it may only complete after the reader finishes. + std::this_thread::sleep_for(std::chrono::milliseconds(150)); + EXPECT_FALSE(checkpoint_returned.load(std::memory_order_acquire)); + + reader.join(); + checkpoint_thread.join(); + EXPECT_TRUE(checkpoint_returned.load(std::memory_order_acquire)); + + svc.reset(); + db.Close(); +} + TEST_F(UpdateTransactionTest, TestUnsupportedInterface) { neug::NeugDB db; neug::NeugDBConfig config(db_dir);