24057299873

Committed 07 Apr 2026 12:01AM UTC coverage: 52.076% (+0.8%) from 51.228%

Build # 24057299873

Build Type

push

github

Committed by

rayandrew

Commit Message

feat(rocksdb): migrate SQLite indexing to RocksDB

Replace SQLite-backed indexing and provenance storage with RocksDB-backed stores.

  Key changes:
  - add RocksDB async/database/db-manager/filesystem/key-codec layers
  - migrate index and provenance databases from SQLite to RocksDB
  - update index builder, trace reader, reorganize, view, stats, and comparator paths for
  RocksDB
  - harden transaction atomicity and rollback behavior with TransactionScope
  - add iterator status checking for prefix scans
  - harden gzip/tar indexer cache state and metadata handling
  - capture executor context in RocksDB awaitables
  - clean up failed RocksDB open paths and manager lifecycle behavior
  - vendor CPM 0.42.1 and update CI/build integration
  - refresh docs, Python bindings, and C++/Python test coverage for the new backend

  Validation:
  - full test suite passed
  - Ubuntu 22.04 Docker run passed
  - focused RocksDB/indexer regression tests passed.

Coverage Stats

24097 of 59624 branches covered (40.41%)

Branch coverage included in aggregate %.

2516 of 3144 new or added lines in 75 files covered. (80.03%)

72 existing lines in 15 files now uncovered.

20858 of 26701 relevant lines covered (78.12%)

14113.43 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.07

/src/dftracer/utils/core/rocksdb/database.cpp

#include <dftracer/utils/core/common/filesystem.h>
#include <dftracer/utils/core/env.h>
#include <dftracer/utils/core/rocksdb/database.h>
#include <dftracer/utils/core/rocksdb/filesystem.h>
#include <rocksdb/slice.h>

#include <algorithm>
#include <atomic>
#include <cstdlib>
#include <stdexcept>
#include <utility>

namespace dftracer::utils::rocksdb {

namespace {

std::atomic<bool>& process_exiting_flag() {
    static std::atomic<bool> flag{false};
    return flag;
}

const ::rocksdb::ReadOptions& read_options() {
    static const ::rocksdb::ReadOptions options;
    return options;
}

const ::rocksdb::WriteOptions& write_options() {
    static const ::rocksdb::WriteOptions options;
    return options;
}

void cleanup_failed_open(::rocksdb::DB*& db,
                         std::vector<::rocksdb::ColumnFamilyHandle*>& handles) {
    if (db != nullptr) {
        for (auto* handle : handles) {
            if (handle != nullptr) {
                db->DestroyColumnFamilyHandle(handle);
            }
        }
        static_cast<void>(db->Close());
        delete db;
        db = nullptr;
    }
    handles.clear();
}

}  // namespace

void mark_process_exiting_for_rocksdb() {
    process_exiting_flag().store(true, std::memory_order_relaxed);
}

RocksDatabase::RocksDatabase() = default;

RocksDatabase::RocksDatabase(const std::string& db_path, OpenMode open_mode) {
    open(db_path, open_mode);
}

RocksDatabase::~RocksDatabase() { close(); }

RocksDatabase::RocksDatabase(RocksDatabase&& other) noexcept
    : db_path_(std::move(other.db_path_)),
      open_mode_(other.open_mode_),
      file_system_(std::move(other.file_system_)),
      env_(std::move(other.env_)),
      db_(std::exchange(other.db_, nullptr)),
      column_families_(std::move(other.column_families_)) {}

RocksDatabase& RocksDatabase::operator=(RocksDatabase&& other) noexcept {
    if (this != &other) {
        close();
        db_path_ = std::move(other.db_path_);
        open_mode_ = other.open_mode_;
        file_system_ = std::move(other.file_system_);
        env_ = std::move(other.env_);
        db_ = std::exchange(other.db_, nullptr);
        column_families_ = std::move(other.column_families_);
    }
    return *this;
}

std::vector<std::string> RocksDatabase::default_column_families() {
    return {"default",    "checkpoints", "metadata",   "chunk_bloom",
            "file_bloom", "chunk_stats", "dimensions", "chunk_dim_stats",
            "manifest",   "provenance",  "archives",   "tar_files"};
}

::rocksdb::Options RocksDatabase::default_options() {
    ::rocksdb::Options options;
    options.create_if_missing = true;
    options.create_missing_column_families = true;
    options.allow_concurrent_memtable_write = true;
    options.enable_pipelined_write = true;
    options.max_open_files = Env::rocksdb_max_open_files();
    return options;
}

::rocksdb::ColumnFamilyOptions RocksDatabase::default_column_family_options() {
    ::rocksdb::ColumnFamilyOptions options;
    options.compression = ::rocksdb::kLZ4Compression;
    options.bottommost_compression = ::rocksdb::kZlibCompression;
    return options;
}

bool RocksDatabase::open(const std::string& db_path, OpenMode open_mode) {
    close();
    db_path_ = db_path;
    open_mode_ = open_mode;

    std::error_code ec;
    if (open_mode_ == OpenMode::ReadWrite) {
        fs::create_directories(fs::path(db_path_), ec);
    }

    auto db_options = default_options();
    if (open_mode_ == OpenMode::ReadOnly) {
        db_options.create_if_missing = false;
        db_options.create_missing_column_families = false;
    }
    file_system_ = make_dftracer_file_system();
    env_ = make_dftracer_env(file_system_);
    db_options.env = env_.get();
    auto cf_options = default_column_family_options();

    std::vector<std::string> column_family_names;
    auto list_status = ::rocksdb::DB::ListColumnFamilies(db_options, db_path_,
                                                         &column_family_names);
    if (!list_status.ok()) {
        if (open_mode_ == OpenMode::ReadOnly) {
            throw std::runtime_error(
                "Failed to list RocksDB column families at '" + db_path_ +
                "': " + list_status.ToString());
        }
        column_family_names = default_column_families();
    } else {
        if (open_mode_ == OpenMode::ReadWrite) {
            for (const auto& name : default_column_families()) {
                if (std::find(column_family_names.begin(),
                              column_family_names.end(),
                              name) == column_family_names.end()) {
                    column_family_names.push_back(name);
                }
            }
        }
    }

    std::vector<::rocksdb::ColumnFamilyDescriptor> descriptors;
    descriptors.reserve(column_family_names.size());
    for (const auto& name : column_family_names) {
        descriptors.emplace_back(name, cf_options);
    }

    std::vector<::rocksdb::ColumnFamilyHandle*> handles;
    auto status =
        open_mode_ == OpenMode::ReadOnly
            ? ::rocksdb::DB::OpenForReadOnly(db_options, db_path_, descriptors,
                                             &handles, &db_, false)
            : ::rocksdb::DB::Open(db_options, db_path_, descriptors, &handles,
                                  &db_);
    if (!status.ok()) {
        cleanup_failed_open(db_, handles);
        throw std::runtime_error("Failed to open RocksDB at '" + db_path_ +
                                 "': " + status.ToString());
    }

    column_families_.clear();
    for (std::size_t i = 0; i < descriptors.size(); ++i) {
        column_families_.emplace(descriptors[i].name, handles[i]);
    }

    return true;
}

void RocksDatabase::close() {
    if (db_ == nullptr) {
        column_families_.clear();
        return;
    }

    if (process_exiting_flag().load(std::memory_order_relaxed)) {
        db_ = nullptr;
        column_families_.clear();
        env_.reset();
        file_system_.reset();
        db_path_.clear();
        return;
    }

    for (auto& entry : column_families_) {
        if (entry.second != nullptr) {
            db_->DestroyColumnFamilyHandle(entry.second);
            entry.second = nullptr;
        }
    }
    column_families_.clear();

    auto* db = db_;
    db_ = nullptr;
    static_cast<void>(db->Close());
    delete db;
    env_.reset();
    file_system_.reset();
    db_path_.clear();
}

bool RocksDatabase::is_open() const noexcept { return db_ != nullptr; }

bool RocksDatabase::is_read_only() const noexcept {
    return open_mode_ == OpenMode::ReadOnly;
}

const std::string& RocksDatabase::path() const noexcept { return db_path_; }

::rocksdb::DB* RocksDatabase::get() const noexcept { return db_; }

::rocksdb::ColumnFamilyHandle* RocksDatabase::column_family_handle(
    std::string_view column_family) const {
    const auto name = column_family.empty() ? std::string("default")
                                            : std::string(column_family);
    const auto it = column_families_.find(name);
    if (it == column_families_.end() || it->second == nullptr) {
        throw std::invalid_argument("Unknown RocksDB column family: " + name);
    }
    return it->second;
}

::rocksdb::Status RocksDatabase::put(std::string_view key,
                                     std::string_view value,
                                     std::string_view column_family) {
    return db_->Put(write_options(), column_family_handle(column_family),
                    ::rocksdb::Slice(key.data(), key.size()),
                    ::rocksdb::Slice(value.data(), value.size()));
}

::rocksdb::Status RocksDatabase::get(std::string_view key, std::string* value,
                                     std::string_view column_family) const {
    return db_->Get(read_options(), column_family_handle(column_family),
                    ::rocksdb::Slice(key.data(), key.size()), value);
}

::rocksdb::Status RocksDatabase::del(std::string_view key,
                                     std::string_view column_family) {
    return db_->Delete(write_options(), column_family_handle(column_family),
                       ::rocksdb::Slice(key.data(), key.size()));
}

::rocksdb::Status RocksDatabase::put(Batch& batch,
                                     std::string_view column_family,
                                     std::string_view key,
                                     std::string_view value) {
    return batch.Put(column_family_handle(column_family),
                     ::rocksdb::Slice(key.data(), key.size()),
                     ::rocksdb::Slice(value.data(), value.size()));
}

::rocksdb::Status RocksDatabase::del(Batch& batch,
                                     std::string_view column_family,
                                     std::string_view key) {
    return batch.Delete(column_family_handle(column_family),
                        ::rocksdb::Slice(key.data(), key.size()));
}

RocksDatabase::Batch RocksDatabase::begin_batch() const { return Batch(); }

::rocksdb::Status RocksDatabase::commit_batch(Batch& batch) {
    return db_->Write(write_options(), &batch);
}

std::unique_ptr<::rocksdb::Iterator> RocksDatabase::new_iterator(
    std::string_view column_family) const {
    return std::unique_ptr<::rocksdb::Iterator>(
        db_->NewIterator(read_options(), column_family_handle(column_family)));
}

}  // namespace dftracer::utils::rocksdb

1	#include <dftracer/utils/core/common/filesystem.h>
2	#include <dftracer/utils/core/env.h>
3	#include <dftracer/utils/core/rocksdb/database.h>
4	#include <dftracer/utils/core/rocksdb/filesystem.h>
5	#include <rocksdb/slice.h>
6
7	#include <algorithm>
8	#include <atomic>
9	#include <cstdlib>
10	#include <stdexcept>
11	#include <utility>
12
13	namespace dftracer::utils::rocksdb {
14
15	namespace {
16
17	std::atomic<bool>& process_exiting_flag() {	11,240✔
18	static std::atomic<bool> flag{false};
19	return flag;	11,240✔
20	}
21
22	const ::rocksdb::ReadOptions& read_options() {	15,726✔
23	static const ::rocksdb::ReadOptions options;	15,726!
24	return options;	15,728✔
25	}
26
27	const ::rocksdb::WriteOptions& write_options() {	5,182✔
28	static const ::rocksdb::WriteOptions options;	5,182!
29	return options;	5,188✔
30	}
31
32	void cleanup_failed_open(::rocksdb::DB*& db,	50✔
33	std::vector<::rocksdb::ColumnFamilyHandle*>& handles) {
34	if (db != nullptr) {	50!
NEW 35	for (auto* handle : handles) {	×
NEW 36	if (handle != nullptr) {	×
NEW 37	db->DestroyColumnFamilyHandle(handle);	×
38	}
39	}
NEW 40	static_cast<void>(db->Close());	×
NEW 41	delete db;	×
NEW 42	db = nullptr;	×
43	}
44	handles.clear();	50✔
45	}	50✔
46
47	} // namespace
48
NEW 49	void mark_process_exiting_for_rocksdb() {	×
NEW 50	process_exiting_flag().store(true, std::memory_order_relaxed);	×
NEW 51	}	×
52
NEW 53	RocksDatabase::RocksDatabase() = default;	×
54
55	RocksDatabase::RocksDatabase(const std::string& db_path, OpenMode open_mode) {	17,186✔
56	open(db_path, open_mode);	11,340✔
57	}	11,528✔
58
59	RocksDatabase::~RocksDatabase() { close(); }	17,033!
60
NEW 61	RocksDatabase::RocksDatabase(RocksDatabase&& other) noexcept	×
NEW 62	: db_path_(std::move(other.db_path_)),	×
NEW 63	open_mode_(other.open_mode_),	×
NEW 64	file_system_(std::move(other.file_system_)),	×
NEW 65	env_(std::move(other.env_)),	×
NEW 66	db_(std::exchange(other.db_, nullptr)),	×
NEW 67	column_families_(std::move(other.column_families_)) {}	×
68
NEW 69	RocksDatabase& RocksDatabase::operator=(RocksDatabase&& other) noexcept {	×
NEW 70	if (this != &other) {	×
NEW 71	close();	×
NEW 72	db_path_ = std::move(other.db_path_);	×
NEW 73	open_mode_ = other.open_mode_;	×
NEW 74	file_system_ = std::move(other.file_system_);	×
NEW 75	env_ = std::move(other.env_);	×
NEW 76	db_ = std::exchange(other.db_, nullptr);	×
NEW 77	column_families_ = std::move(other.column_families_);	×
78	}
NEW 79	return *this;	×
80	}
81
82	std::vector<std::string> RocksDatabase::default_column_families() {	1,307✔
83	return {"default", "checkpoints", "metadata", "chunk_bloom",	654!
84	"file_bloom", "chunk_stats", "dimensions", "chunk_dim_stats",	654!
85	"manifest", "provenance", "archives", "tar_files"};	9,155!
86	}
87
88	::rocksdb::Options RocksDatabase::default_options() {	11,340✔
89	::rocksdb::Options options;	11,340✔
90	options.create_if_missing = true;	11,338✔
91	options.create_missing_column_families = true;	11,338✔
92	options.allow_concurrent_memtable_write = true;	11,338✔
93	options.enable_pipelined_write = true;	11,338✔
94	options.max_open_files = Env::rocksdb_max_open_files();	11,338!
95	return options;	11,337✔
96	}	5,846!
97
98	::rocksdb::ColumnFamilyOptions RocksDatabase::default_column_family_options() {	11,337✔
99	::rocksdb::ColumnFamilyOptions options;	11,337✔
100	options.compression = ::rocksdb::kLZ4Compression;	11,340✔
101	options.bottommost_compression = ::rocksdb::kZlibCompression;	11,340✔
102	return options;	11,340✔
103	}	5,846!
104
105	bool RocksDatabase::open(const std::string& db_path, OpenMode open_mode) {	11,340✔
106	close();	11,340!
107	db_path_ = db_path;	11,340!
108	open_mode_ = open_mode;	11,340✔
109
110	std::error_code ec;	11,340✔
111	if (open_mode_ == OpenMode::ReadWrite) {	11,340✔
112	fs::create_directories(fs::path(db_path_), ec);	1,355!
113	}	654✔
114
115	auto db_options = default_options();	11,340!
116	if (open_mode_ == OpenMode::ReadOnly) {	11,336✔
117	db_options.create_if_missing = false;	10,030✔
118	db_options.create_missing_column_families = false;	10,030✔
119	}	5,192✔
120	file_system_ = make_dftracer_file_system();	11,336!
121	env_ = make_dftracer_env(file_system_);	11,340!
122	db_options.env = env_.get();	11,339✔
123	auto cf_options = default_column_family_options();	11,338!
124
125	std::vector<std::string> column_family_names;	11,337✔
126	auto list_status = ::rocksdb::DB::ListColumnFamilies(db_options, db_path_,	11,338!
127	&column_family_names);	5,492!
128	if (!list_status.ok()) {	11,337!
129	if (open_mode_ == OpenMode::ReadOnly) {	1,018✔
130	throw std::runtime_error(	66!
131	"Failed to list RocksDB column families at '" + db_path_ +	66!
132	"': " + list_status.ToString());	110!
133	}
134	column_family_names = default_column_families();	974!
135	} else {	487✔
136	if (open_mode_ == OpenMode::ReadWrite) {	10,319✔
137	for (const auto& name : default_column_families()) {	4,338!
138	if (std::find(column_family_names.begin(),	6,006!
139	column_family_names.end(),	2,001✔
140	name) == column_family_names.end()) {	6,007!
NEW 141	column_family_names.push_back(name);	×
142	}
143	}	167✔
144	}	167✔
145	}
146
147	std::vector<::rocksdb::ColumnFamilyDescriptor> descriptors;	11,294✔
148	descriptors.reserve(column_family_names.size());	11,293!
149	for (const auto& name : column_family_names) {	146,836✔
150	descriptors.emplace_back(name, cf_options);	135,544!
151	}
152
153	std::vector<::rocksdb::ColumnFamilyHandle*> handles;	11,297✔
154	auto status =
155	open_mode_ == OpenMode::ReadOnly	17,117✔
156	? ::rocksdb::DB::OpenForReadOnly(db_options, db_path_, descriptors,	15,157!
157	&handles, &db_, false)	5,170✔
158	: ::rocksdb::DB::Open(db_options, db_path_, descriptors, &handles,	1,960!
159	&db_);	6,123!
160	if (!status.ok()) {	11,288✔
161	cleanup_failed_open(db_, handles);	50✔
162	throw std::runtime_error("Failed to open RocksDB at '" + db_path_ +	125!
163	"': " + status.ToString());	125!
164	}
165
166	column_families_.clear();	11,235✔
167	for (std::size_t i = 0; i < descriptors.size(); ++i) {	146,163✔
168	column_families_.emplace(descriptors[i].name, handles[i]);	134,925✔
169	}	69,577✔
170
171	return true;	5,447✔
172	}	11,601✔
173
174	void RocksDatabase::close() {	22,583✔
175	if (db_ == nullptr) {	22,583✔
176	column_families_.clear();	11,340✔
177	return;	11,340✔
178	}
179
180	if (process_exiting_flag().load(std::memory_order_relaxed)) {	11,243!
NEW 181	db_ = nullptr;	×
NEW 182	column_families_.clear();	×
NEW 183	env_.reset();	×
NEW 184	file_system_.reset();	×
NEW 185	db_path_.clear();	×
NEW 186	return;	×
187	}
188
189	for (auto& entry : column_families_) {	146,173✔
190	if (entry.second != nullptr) {	134,928✔
191	db_->DestroyColumnFamilyHandle(entry.second);	134,921!
192	entry.second = nullptr;	134,921✔
193	}	69,568✔
194	}
195	column_families_.clear();	11,246✔
196
197	auto* db = db_;	11,245✔
198	db_ = nullptr;	11,245✔
199	static_cast<void>(db->Close());	11,245✔
200	delete db;	11,247✔
201	env_.reset();	11,245✔
202	file_system_.reset();	11,247✔
203	db_path_.clear();	11,248✔
204	}	11,647✔
205
206	bool RocksDatabase::is_open() const noexcept { return db_ != nullptr; }	8✔
207
208	bool RocksDatabase::is_read_only() const noexcept {	646✔
209	return open_mode_ == OpenMode::ReadOnly;	646✔
210	}
211
NEW 212	const std::string& RocksDatabase::path() const noexcept { return db_path_; }	×
213
NEW 214	::rocksdb::DB* RocksDatabase::get() const noexcept { return db_; }	×
215
216	::rocksdb::ColumnFamilyHandle* RocksDatabase::column_family_handle(	26,257✔
217	std::string_view column_family) const {
218	const auto name = column_family.empty() ? std::string("default")	26,257!
219	: std::string(column_family);	39,397!
220	const auto it = column_families_.find(name);	26,262✔
221	if (it == column_families_.end() \|\| it->second == nullptr) {	26,258!
NEW 222	throw std::invalid_argument("Unknown RocksDB column family: " + name);	×
223	}
224	return it->second;	39,395✔
225	}	26,268✔
226
227	::rocksdb::Status RocksDatabase::put(std::string_view key,	4,014✔
228	std::string_view value,
229	std::string_view column_family) {
230	return db_->Put(write_options(), column_family_handle(column_family),	6,021✔
231	::rocksdb::Slice(key.data(), key.size()),	4,013✔
232	::rocksdb::Slice(value.data(), value.size()));	8,026!
233	}
234
235	::rocksdb::Status RocksDatabase::get(std::string_view key, std::string* value,	10,667✔
236	std::string_view column_family) const {
237	return db_->Get(read_options(), column_family_handle(column_family),	16,002✔
238	::rocksdb::Slice(key.data(), key.size()), value);	15,966!
239	}
240
241	::rocksdb::Status RocksDatabase::del(std::string_view key,	36✔
242	std::string_view column_family) {
243	return db_->Delete(write_options(), column_family_handle(column_family),	54✔
244	::rocksdb::Slice(key.data(), key.size()));	54!
245	}
246
247	::rocksdb::Status RocksDatabase::put(Batch& batch,	6,475✔
248	std::string_view column_family,
249	std::string_view key,
250	std::string_view value) {
251	return batch.Put(column_family_handle(column_family),	9,714✔
252	::rocksdb::Slice(key.data(), key.size()),	6,478✔
253	::rocksdb::Slice(value.data(), value.size()));	12,952!
254	}
255
256	::rocksdb::Status RocksDatabase::del(Batch& batch,	2✔
257	std::string_view column_family,
258	std::string_view key) {
259	return batch.Delete(column_family_handle(column_family),	3✔
260	::rocksdb::Slice(key.data(), key.size()));	3!
261	}
262
263	RocksDatabase::Batch RocksDatabase::begin_batch() const { return Batch(); }	1,139✔
264
265	::rocksdb::Status RocksDatabase::commit_batch(Batch& batch) {	1,135✔
266	return db_->Write(write_options(), &batch);	1,135✔
267	}
268
269	std::unique_ptr<::rocksdb::Iterator> RocksDatabase::new_iterator(	5,060✔
270	std::string_view column_family) const {
271	return std::unique_ptr<::rocksdb::Iterator>(	2,523✔
272	db_->NewIterator(read_options(), column_family_handle(column_family)));	5,060✔
273	}
274
275	} // namespace dftracer::utils::rocksdb

llnl / dftracer-utils / 24057299873

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous