30002033718

Committed 23 Jul 2026 11:08AM UTC coverage: 52.742% (+0.08%) from 52.66%

Build # 30002033718

Build Type

Pull #99

github

Committed by

web-flow

Commit Message

Merge df0b9588d into 06bc84ec9

Pull Request Pull Request #99: Support CM time_metric (NS/MS/SEC/US) across reader and viz

Coverage Stats

41941 of 102829 branches covered (40.79%)

Branch coverage included in aggregate %.

2221 of 2839 new or added lines in 58 files covered. (78.23%)

137 existing lines in 13 files now uncovered.

37042 of 46924 relevant lines covered (78.94%)

59523.07 hits per line

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

63.81

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

#include <dftracer/utils/core/common/constants.h>
#include <dftracer/utils/core/common/error.h>
#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/filter_policy.h>
#include <rocksdb/slice.h>
#include <rocksdb/table.h>
#include <rocksdb/write_buffer_manager.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 auto options = [] {
        ::rocksdb::WriteOptions wo;
        wo.disableWAL = true;
        return wo;
    }();
    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;
}

const decltype(cf::ALL)& RocksDatabase::default_column_families() {
    return cf::ALL;
}

namespace {
/// One cache for the process. Left unset, every column family gets its own
/// 8 MB default, which no realistic working set fits in.
std::shared_ptr<::rocksdb::Cache>& shared_block_cache() {
    static std::shared_ptr<::rocksdb::Cache> cache =
        ::rocksdb::NewLRUCache(constants::rocksdb::BLOCK_CACHE_BYTES);
    return cache;
}

/// Caps total memtable memory across every column family and every DB in the
/// process; without it each CF reserves its own write buffers.
std::shared_ptr<::rocksdb::WriteBufferManager>& shared_write_buffer_manager() {
    static std::shared_ptr<::rocksdb::WriteBufferManager> mgr =
        std::make_shared<::rocksdb::WriteBufferManager>(
            constants::rocksdb::WRITE_BUFFER_BYTES, shared_block_cache());
    return mgr;
}
}  // namespace

::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();
    options.max_background_jobs = 8;
    options.max_subcompactions = 8;
    options.write_buffer_size = 256 * 1024 * 1024;
    options.max_write_buffer_number = 4;
    options.write_buffer_manager = shared_write_buffer_manager();
    return options;
}

::rocksdb::ColumnFamilyOptions RocksDatabase::default_column_family_options() {
    ::rocksdb::ColumnFamilyOptions options;

    ::rocksdb::BlockBasedTableOptions bbt;
    bbt.block_size = 32 * 1024;
    bbt.format_version = 5;
    bbt.index_block_restart_interval = 16;
    bbt.block_cache = shared_block_cache();
    options.table_factory.reset(::rocksdb::NewBlockBasedTableFactory(bbt));

#ifdef DFTRACER_UTILS_ENABLE_ZSTD
    options.compression = ::rocksdb::kZSTD;
    options.compression_opts.level = constants::rocksdb::ZSTD_COMPRESSION_LEVEL;
    options.compression_opts.max_dict_bytes =
        constants::rocksdb::ZSTD_MAX_DICT_BYTES;
    options.compression_opts.zstd_max_train_bytes =
        constants::rocksdb::ZSTD_MAX_TRAIN_BYTES;
    options.compression_opts.enabled = true;
    options.bottommost_compression = ::rocksdb::kZSTD;
    options.bottommost_compression_opts.level =
        constants::rocksdb::ZSTD_COMPRESSION_LEVEL;
    options.bottommost_compression_opts.max_dict_bytes =
        constants::rocksdb::ZSTD_MAX_DICT_BYTES;
    options.bottommost_compression_opts.zstd_max_train_bytes =
        constants::rocksdb::ZSTD_MAX_TRAIN_BYTES;
    options.bottommost_compression_opts.enabled = true;
#elif defined(DFTRACER_UTILS_ENABLE_LZ4)
    options.compression = ::rocksdb::kLZ4Compression;
    options.bottommost_compression = ::rocksdb::kZlibCompression;
#else
    options.compression = ::rocksdb::kZlibCompression;
    options.bottommost_compression = ::rocksdb::kZlibCompression;
#endif
    return options;
}

::rocksdb::ColumnFamilyOptions
RocksDatabase::point_lookup_column_family_options() {
    auto options = default_column_family_options();

    // Hash keys are uniformly distributed, so every SST's range covers almost
    // every key: without a filter a get searches all of them. Large blocks
    // also mean reading and decompressing 32 KB to return a few dozen bytes.
    ::rocksdb::BlockBasedTableOptions bbt;
    bbt.block_size = constants::rocksdb::POINT_LOOKUP_BLOCK_SIZE;
    bbt.format_version = 5;
    bbt.index_block_restart_interval = 16;
    bbt.block_cache = shared_block_cache();
    bbt.filter_policy.reset(::rocksdb::NewBloomFilterPolicy(
        constants::rocksdb::BLOOM_BITS_PER_KEY, false));
    // Left in the table reader rather than the block cache: a scan streams
    // enough data blocks through the cache to evict the filters it needs.
    bbt.cache_index_and_filter_blocks = false;
    options.table_factory.reset(::rocksdb::NewBlockBasedTableFactory(bbt));
    return options;
}

bool RocksDatabase::is_point_lookup_cf(std::string_view name) noexcept {
    return name == cf::HASH_TABLES || name == cf::NAME_DICTIONARY;
}

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;
        // Read paths do heavy point lookups (e.g. dfanalyzer hash resolution);
        // with a small table cache the SST readers get evicted and every lookup
        // re-opens the file to re-read its filter/index blocks, which thrashes.
        // There is no write-side fd pressure here, so keep every SST open
        // unless an explicit env cap is set.
        if (!Env::get<int>("DFTRACER_UTILS_ROCKSDB_MAX_OPEN_FILES")
                 .has_value()) {
            db_options.max_open_files = -1;
        }
    }
    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 DFTUtilsException(
                ErrorCode::IO, "Failed to list RocksDB column families at '" +
                                   db_path_ + "': " + list_status.ToString());
        }
        column_family_names.reserve(default_column_families().size());
        for (auto name : default_column_families()) {
            column_family_names.emplace_back(name);
        }
    } 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.emplace_back(name);
                }
            }
        }
    }

    std::vector<::rocksdb::ColumnFamilyDescriptor> descriptors;
    descriptors.reserve(column_family_names.size());
    for (const auto& name : column_family_names) {
        auto opts = is_point_lookup_cf(name)
                        ? point_lookup_column_family_options()
                        : cf_options;
        if (cf_options_override_) {
            cf_options_override_(name, opts);
        }
        descriptors.emplace_back(name, opts);
    }

    std::vector<::rocksdb::ColumnFamilyHandle*> handles;
    ::rocksdb::Status status;
    if (open_mode_ == OpenMode::ReadOnly) {
        status = ::rocksdb::DB::OpenForReadOnly(
            db_options, db_path_, descriptors, &handles, &db_, false);
    } else {
        status = ::rocksdb::DB::Open(db_options, db_path_, descriptors,
                                     &handles, &db_);
    }
    if (!status.ok()) {
        cleanup_failed_open(db_, handles);
        throw DFTUtilsException(ErrorCode::IO, "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(cf::DEFAULT)
                                            : std::string(column_family);
    const auto it = column_families_.find(name);
    if (it == column_families_.end() || it->second == nullptr) {
        throw DFTUtilsException(ErrorCode::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::merge(std::string_view key,
                                       std::string_view value,
                                       std::string_view column_family) {
    return db_->Merge(write_options(), column_family_handle(column_family),
                      ::rocksdb::Slice(key.data(), key.size()),
                      ::rocksdb::Slice(value.data(), value.size()));
}

void RocksDatabase::set_cf_options_override(CfOptionsOverride override) {
    cf_options_override_ = std::move(override);
}

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

::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::delete_range(std::string_view begin_key,
                                              std::string_view end_key,
                                              std::string_view column_family) {
    return db_->DeleteRange(
        write_options(), column_family_handle(column_family),
        ::rocksdb::Slice(begin_key.data(), begin_key.size()),
        ::rocksdb::Slice(end_key.data(), end_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)));
}

::rocksdb::Status RocksDatabase::compact(std::string_view column_family) {
    ::rocksdb::CompactRangeOptions opts;
    opts.max_subcompactions = 8;
    return db_->CompactRange(opts, column_family_handle(column_family), nullptr,
                             nullptr);
}

::rocksdb::Status RocksDatabase::ingest_external_files(
    std::string_view column_family,
    const std::vector<std::string>& external_files, bool ingest_behind) {
    if (external_files.empty()) {
        return ::rocksdb::Status::OK();
    }
    ::rocksdb::IngestExternalFileOptions opts;
    // Rename same-FS staged SSTs instead of copying; RocksDB copies on failure.
    opts.move_files = true;
    // Offline bulk build with no live readers, so skip the snapshot/seqno sync.
    opts.snapshot_consistency = false;
    opts.allow_global_seqno = true;
    // Keep the assigned seqno in the manifest instead of rewriting it into each
    // SST (format_version 5 supports this), avoiding a per-SST write + fsync.
    opts.write_global_seqno = false;
    opts.allow_blocking_flush = true;
    opts.ingest_behind = ingest_behind;
    return db_->IngestExternalFile(column_family_handle(column_family),
                                   external_files, opts);
}

::rocksdb::Status RocksDatabase::ingest_external_files_multi(
    const std::vector<
        std::pair<std::string_view, const std::vector<std::string>*>>& per_cf) {
    ::rocksdb::IngestExternalFileOptions opts;
    opts.move_files = true;
    opts.snapshot_consistency = false;
    opts.allow_global_seqno = true;
    opts.write_global_seqno = false;
    opts.allow_blocking_flush = true;

    std::vector<::rocksdb::IngestExternalFileArg> args;
    args.reserve(per_cf.size());
    for (const auto& [cf_name, files] : per_cf) {
        if (!files || files->empty()) continue;
        ::rocksdb::IngestExternalFileArg arg;
        arg.column_family = column_family_handle(cf_name);
        arg.external_files = *files;
        arg.options = opts;
        args.push_back(std::move(arg));
    }
    if (args.empty()) return ::rocksdb::Status::OK();
    return db_->IngestExternalFiles(args);
}

}  // namespace dftracer::utils::rocksdb

1	#include <dftracer/utils/core/common/constants.h>
2	#include <dftracer/utils/core/common/error.h>
3	#include <dftracer/utils/core/common/filesystem.h>
4	#include <dftracer/utils/core/env.h>
5	#include <dftracer/utils/core/rocksdb/database.h>
6	#include <dftracer/utils/core/rocksdb/filesystem.h>
7	#include <rocksdb/filter_policy.h>
8	#include <rocksdb/slice.h>
9	#include <rocksdb/table.h>
10	#include <rocksdb/write_buffer_manager.h>
11
12	#include <algorithm>
13	#include <atomic>
14	#include <cstdlib>
15	#include <stdexcept>
16	#include <utility>
17
18	namespace dftracer::utils::rocksdb {
19
20	namespace {
21
22	std::atomic<bool>& process_exiting_flag() {	12,403✔
23	static std::atomic<bool> flag{false};
24	return flag;	12,403✔
25	}
26
27	const ::rocksdb::ReadOptions& read_options() {	25,972✔
28	static const ::rocksdb::ReadOptions options;	25,972!
29	return options;	25,966✔
30	}
31
32	const ::rocksdb::WriteOptions& write_options() {	4,580✔
33	static const auto options = [] {	2,515!
34	::rocksdb::WriteOptions wo;	224✔
35	wo.disableWAL = true;	224✔
36	return wo;	224✔
37	}();	2,289!
38	return options;	4,580✔
39	}
40
41	void cleanup_failed_open(::rocksdb::DB*& db,	×
42	std::vector<::rocksdb::ColumnFamilyHandle*>& handles) {
43	if (db != nullptr) {	×
44	for (auto* handle : handles) {	×
45	if (handle != nullptr) {	×
46	db->DestroyColumnFamilyHandle(handle);	×
47	}
48	}
49	static_cast<void>(db->Close());	×
50	delete db;	×
51	db = nullptr;	×
52	}
53	handles.clear();	×
54	}	×
55
56	} // namespace
57
58	void mark_process_exiting_for_rocksdb() {	12✔
59	process_exiting_flag().store(true, std::memory_order_relaxed);	12✔
60	}	12✔
61
62	RocksDatabase::RocksDatabase() = default;	31,385✔
63
64	RocksDatabase::RocksDatabase(const std::string& db_path, OpenMode open_mode) {	15✔
65	open(db_path, open_mode);	6!
66	}	9✔
67
68	RocksDatabase::~RocksDatabase() { close(); }	18,756!
69
70	RocksDatabase::RocksDatabase(RocksDatabase&& other) noexcept	×
71	: db_path_(std::move(other.db_path_)),	×
72	open_mode_(other.open_mode_),	×
73	file_system_(std::move(other.file_system_)),	×
74	env_(std::move(other.env_)),	×
75	db_(std::exchange(other.db_, nullptr)),	×
76	column_families_(std::move(other.column_families_)) {}	×
77
78	RocksDatabase& RocksDatabase::operator=(RocksDatabase&& other) noexcept {	×
79	if (this != &other) {	×
80	close();	×
81	db_path_ = std::move(other.db_path_);	×
82	open_mode_ = other.open_mode_;	×
83	file_system_ = std::move(other.file_system_);	×
84	env_ = std::move(other.env_);	×
85	db_ = std::exchange(other.db_, nullptr);	×
86	column_families_ = std::move(other.column_families_);	×
87	}
88	return *this;	×
89	}
90
91	const decltype(cf::ALL)& RocksDatabase::default_column_families() {	3,082✔
92	return cf::ALL;	3,082✔
93	}
94
95	namespace {
96	/// One cache for the process. Left unset, every column family gets its own
97	/// 8 MB default, which no realistic working set fits in.
98	std::shared_ptr<::rocksdb::Cache>& shared_block_cache() {	62,701✔
99	static std::shared_ptr<::rocksdb::Cache> cache =	31,988!
100	::rocksdb::NewLRUCache(constants::rocksdb::BLOCK_CACHE_BYTES);	30,943!
101	return cache;	62,701✔
102	}
103
104	/// Caps total memtable memory across every column family and every DB in the
105	/// process; without it each CF reserves its own write buffers.
106	std::shared_ptr<::rocksdb::WriteBufferManager>& shared_write_buffer_manager() {	12,907✔
107	static std::shared_ptr<::rocksdb::WriteBufferManager> mgr =	6,673✔
108	std::make_shared<::rocksdb::WriteBufferManager>(	115!
109	constants::rocksdb::WRITE_BUFFER_BYTES, shared_block_cache());	6,464!
110	return mgr;	12,907✔
111	}
112	} // namespace
113
114	::rocksdb::Options RocksDatabase::default_options() {	12,907✔
115	::rocksdb::Options options;	12,907✔
116	options.create_if_missing = true;	12,907✔
117	options.create_missing_column_families = true;	12,907✔
118	options.allow_concurrent_memtable_write = true;	12,907✔
119	options.enable_pipelined_write = true;	12,907✔
120	options.max_open_files = Env::rocksdb_max_open_files();	12,907!
121	options.max_background_jobs = 8;	12,907✔
122	options.max_subcompactions = 8;	12,907✔
123	options.write_buffer_size = 256 * 1024 * 1024;	12,907✔
124	options.max_write_buffer_number = 4;	12,907✔
125	options.write_buffer_manager = shared_write_buffer_manager();	12,907!
126	return options;	12,907✔
127	}	6,558!
128
129	::rocksdb::ColumnFamilyOptions RocksDatabase::default_column_family_options() {	37,689✔
130	::rocksdb::ColumnFamilyOptions options;	37,689!
131
132	::rocksdb::BlockBasedTableOptions bbt;	37,689✔
133	bbt.block_size = 32 * 1024;	37,689✔
134	bbt.format_version = 5;	37,689✔
135	bbt.index_block_restart_interval = 16;	37,689✔
136	bbt.block_cache = shared_block_cache();	37,689!
137	options.table_factory.reset(::rocksdb::NewBlockBasedTableFactory(bbt));	37,689!
138
139	#ifdef DFTRACER_UTILS_ENABLE_ZSTD
140	options.compression = ::rocksdb::kZSTD;
141	options.compression_opts.level = constants::rocksdb::ZSTD_COMPRESSION_LEVEL;
142	options.compression_opts.max_dict_bytes =
143	constants::rocksdb::ZSTD_MAX_DICT_BYTES;
144	options.compression_opts.zstd_max_train_bytes =
145	constants::rocksdb::ZSTD_MAX_TRAIN_BYTES;
146	options.compression_opts.enabled = true;
147	options.bottommost_compression = ::rocksdb::kZSTD;
148	options.bottommost_compression_opts.level =
149	constants::rocksdb::ZSTD_COMPRESSION_LEVEL;
150	options.bottommost_compression_opts.max_dict_bytes =
151	constants::rocksdb::ZSTD_MAX_DICT_BYTES;
152	options.bottommost_compression_opts.zstd_max_train_bytes =
153	constants::rocksdb::ZSTD_MAX_TRAIN_BYTES;
154	options.bottommost_compression_opts.enabled = true;
155	#elif defined(DFTRACER_UTILS_ENABLE_LZ4)
156	options.compression = ::rocksdb::kLZ4Compression;
157	options.bottommost_compression = ::rocksdb::kZlibCompression;
158	#else
159	options.compression = ::rocksdb::kZlibCompression;	37,689✔
160	options.bottommost_compression = ::rocksdb::kZlibCompression;	37,689✔
161	#endif
162	return options;	56,220✔
163	}	37,689!
164
165	::rocksdb::ColumnFamilyOptions
166	RocksDatabase::point_lookup_column_family_options() {	24,782✔
167	auto options = default_column_family_options();	24,782!
168
169	// Hash keys are uniformly distributed, so every SST's range covers almost
170	// every key: without a filter a get searches all of them. Large blocks
171	// also mean reading and decompressing 32 KB to return a few dozen bytes.
172	::rocksdb::BlockBasedTableOptions bbt;	24,782✔
173	bbt.block_size = constants::rocksdb::POINT_LOOKUP_BLOCK_SIZE;	24,782✔
174	bbt.format_version = 5;	24,782✔
175	bbt.index_block_restart_interval = 16;	24,782✔
176	bbt.block_cache = shared_block_cache();	24,782!
177	bbt.filter_policy.reset(::rocksdb::NewBloomFilterPolicy(	24,782!
178	constants::rocksdb::BLOOM_BITS_PER_KEY, false));
179	// Left in the table reader rather than the block cache: a scan streams
180	// enough data blocks through the cache to evict the filters it needs.
181	bbt.cache_index_and_filter_blocks = false;	24,782✔
182	options.table_factory.reset(::rocksdb::NewBlockBasedTableFactory(bbt));	24,782!
183	return options;	36,964✔
184	}	24,782!
185
186	bool RocksDatabase::is_point_lookup_cf(std::string_view name) noexcept {	322,147✔
187	return name == cf::HASH_TABLES \|\| name == cf::NAME_DICTIONARY;	322,147✔
188	}
189
190	bool RocksDatabase::open(const std::string& db_path, OpenMode open_mode) {	12,439✔
191	close();	12,439!
192	db_path_ = db_path;	12,439!
193	open_mode_ = open_mode;	12,439✔
194
195	std::error_code ec;	12,439✔
196	if (open_mode_ == OpenMode::ReadWrite) {	12,439✔
197	fs::create_directories(fs::path(db_path_), ec);	1,921!
198	}	949✔
199
200	auto db_options = default_options();	12,439!
201	if (open_mode_ == OpenMode::ReadOnly) {	12,439✔
202	db_options.create_if_missing = false;	10,535✔
203	db_options.create_missing_column_families = false;	10,535✔
204	// Read paths do heavy point lookups (e.g. dfanalyzer hash resolution);
205	// with a small table cache the SST readers get evicted and every lookup
206	// re-opens the file to re-read its filter/index blocks, which thrashes.
207	// There is no write-side fd pressure here, so keep every SST open
208	// unless an explicit env cap is set.
209	if (!Env::get<int>("DFTRACER_UTILS_ROCKSDB_MAX_OPEN_FILES")	15,697!
210	.has_value()) {	10,535✔
211	db_options.max_open_files = -1;	10,534✔
212	}	5,372✔
213	}	5,372✔
214	file_system_ = make_dftracer_file_system();	12,438✔
215	env_ = make_dftracer_env(file_system_);	12,434!
216	db_options.env = env_.get();	12,435✔
217	auto cf_options = default_column_family_options();	12,434✔
218
219	std::vector<std::string> column_family_names;	12,439✔
220	auto list_status = ::rocksdb::DB::ListColumnFamilies(db_options, db_path_,	12,439!
221	&column_family_names);	6,113!
222	if (!list_status.ok()) {	12,439!
223	if (open_mode_ == OpenMode::ReadOnly) {	1,226✔
224	throw DFTUtilsException(	66!
225	ErrorCode::IO, "Failed to list RocksDB column families at '" +	22!
226	db_path_ + "': " + list_status.ToString());	66!
227	}
228	column_family_names.reserve(default_column_families().size());	1,182!
229	for (auto name : default_column_families()) {	31,905✔
230	column_family_names.emplace_back(name);	30,723!
231	}
232	} else {	591✔
233	if (open_mode_ == OpenMode::ReadWrite) {	11,213✔
234	for (const auto& name : default_column_families()) {	19,385✔
235	if (std::find(column_family_names.begin(),	27,999!
236	column_family_names.end(),	9,334✔
237	name) == column_family_names.end()) {	27,999!
238	column_family_names.emplace_back(name);	×
239	}
240	}
241	}	359✔
242	}
243
244	std::vector<::rocksdb::ColumnFamilyDescriptor> descriptors;	12,395✔
245	descriptors.reserve(column_family_names.size());	12,394✔
246	for (const auto& name : column_family_names) {	334,538✔
247	auto opts = is_point_lookup_cf(name)	485,949✔
248	? point_lookup_column_family_options()	12,601✔
249	: cf_options;	309,551!
250	if (cf_options_override_) {	322,152✔
251	cf_options_override_(name, opts);	320,385✔
252	}	162,915✔
253	descriptors.emplace_back(name, opts);	322,145✔
254	}	322,152✔
255
256	std::vector<::rocksdb::ColumnFamilyHandle*> handles;	12,390✔
257	::rocksdb::Status status;	12,391!
258	if (open_mode_ == OpenMode::ReadOnly) {	12,391✔
259	status = ::rocksdb::DB::OpenForReadOnly(	10,491✔
260	db_options, db_path_, descriptors, &handles, &db_, false);	10,491!
261	} else {	5,349✔
262	status = ::rocksdb::DB::Open(db_options, db_path_, descriptors,	3,800!
263	&handles, &db_);	1,900✔
264	}
265	if (!status.ok()) {	12,390!
266	cleanup_failed_open(db_, handles);	×
267	throw DFTUtilsException(ErrorCode::IO, "Failed to open RocksDB at '" +	×
268	db_path_ +	×
269	"': " + status.ToString());	×
270	}
271
272	column_families_.clear();	12,391✔
273	for (std::size_t i = 0; i < descriptors.size(); ++i) {	334,551✔
274	column_families_.emplace(descriptors[i].name, handles[i]);	322,166✔
275	}	163,800✔
276
277	return true;	6,091✔
278	}	12,542✔
279
280	void RocksDatabase::close() {	24,870✔
281	if (db_ == nullptr) {	24,870✔
282	column_families_.clear();	12,479✔
283	return;	12,479✔
284	}
285
286	if (process_exiting_flag().load(std::memory_order_relaxed)) {	12,391!
287	db_ = nullptr;	×
288	column_families_.clear();	×
289	env_.reset();	×
290	file_system_.reset();	×
291	db_path_.clear();	×
292	return;	×
293	}
294
295	for (auto& entry : column_families_) {	334,552✔
296	if (entry.second != nullptr) {	322,161✔
297	db_->DestroyColumnFamilyHandle(entry.second);	322,158!
298	entry.second = nullptr;	322,158✔
299	}	163,797✔
300	}
301	column_families_.clear();	12,391✔
302
303	auto* db = db_;	12,390✔
304	db_ = nullptr;	12,390✔
305	static_cast<void>(db->Close());	12,390✔
306	delete db;	12,391✔
307	env_.reset();	12,391✔
308	file_system_.reset();	12,391✔
309	db_path_.clear();	12,391✔
310	}	12,644✔
311
312	bool RocksDatabase::is_open() const noexcept { return db_ != nullptr; }	36✔
313
314	bool RocksDatabase::is_read_only() const noexcept {	2,319✔
315	return open_mode_ == OpenMode::ReadOnly;	2,319✔
316	}
317
318	const std::string& RocksDatabase::path() const noexcept { return db_path_; }	116✔
319
320	::rocksdb::DB* RocksDatabase::get() const noexcept { return db_; }	×
321
322	::rocksdb::ColumnFamilyHandle* RocksDatabase::column_family_handle(	85,202✔
323	std::string_view column_family) const {
324	const auto name = column_family.empty() ? std::string(cf::DEFAULT)	85,202!
325	: std::string(column_family);	127,766!
326	const auto it = column_families_.find(name);	85,187✔
327	if (it == column_families_.end() \|\| it->second == nullptr) {	85,158!
328	throw DFTUtilsException(ErrorCode::INVALID_ARGUMENT,	×
329	"Unknown RocksDB column family: " + name);	×
330	}
331	return it->second;	127,646✔
332	}	85,140✔
333
334	::rocksdb::Status RocksDatabase::put(std::string_view key,	1,524✔
335	std::string_view value,
336	std::string_view column_family) {
337	return db_->Put(write_options(), column_family_handle(column_family),	2,286✔
338	::rocksdb::Slice(key.data(), key.size()),	1,524✔
339	::rocksdb::Slice(value.data(), value.size()));	3,048!
340	}
341
342	::rocksdb::Status RocksDatabase::get(std::string_view key, std::string* value,	15,245✔
343	std::string_view column_family) const {
344	return db_->Get(read_options(), column_family_handle(column_family),	22,849✔
345	::rocksdb::Slice(key.data(), key.size()), value);	22,888!
346	}
347
348	::rocksdb::Status RocksDatabase::merge(std::string_view key,	×
349	std::string_view value,
350	std::string_view column_family) {
351	return db_->Merge(write_options(), column_family_handle(column_family),	×
352	::rocksdb::Slice(key.data(), key.size()),	×
353	::rocksdb::Slice(value.data(), value.size()));	×
354	}
355
356	void RocksDatabase::set_cf_options_override(CfOptionsOverride override) {	12,366✔
357	cf_options_override_ = std::move(override);	12,366✔
358	}	12,367✔
359
360	::rocksdb::Status RocksDatabase::merge(Batch& batch,	1,924✔
361	std::string_view column_family,
362	std::string_view key,
363	std::string_view value) {
364	return batch.Merge(column_family_handle(column_family),	2,890✔
365	::rocksdb::Slice(key.data(), key.size()),	1,925✔
366	::rocksdb::Slice(value.data(), value.size()));	3,849!
367	}
368
369	::rocksdb::Status RocksDatabase::del(std::string_view key,	×
370	std::string_view column_family) {
371	return db_->Delete(write_options(), column_family_handle(column_family),	×
372	::rocksdb::Slice(key.data(), key.size()));	×
373	}
374
375	::rocksdb::Status RocksDatabase::delete_range(std::string_view begin_key,	×
376	std::string_view end_key,
377	std::string_view column_family) {
378	return db_->DeleteRange(	×
379	write_options(), column_family_handle(column_family),
380	::rocksdb::Slice(begin_key.data(), begin_key.size()),	×
381	::rocksdb::Slice(end_key.data(), end_key.size()));	×
382	}
383
384	::rocksdb::Status RocksDatabase::put(Batch& batch,	55,148✔
385	std::string_view column_family,
386	std::string_view key,
387	std::string_view value) {
388	return batch.Put(column_family_handle(column_family),	82,771✔
389	::rocksdb::Slice(key.data(), key.size()),	55,161✔
390	::rocksdb::Slice(value.data(), value.size()));	110,293!
391	}
392
393	::rocksdb::Status RocksDatabase::del(Batch& batch,	120✔
394	std::string_view column_family,
395	std::string_view key) {
396	return batch.Delete(column_family_handle(column_family),	180✔
397	::rocksdb::Slice(key.data(), key.size()));	180!
398	}
399
400	RocksDatabase::Batch RocksDatabase::begin_batch() const { return Batch(); }	3,066✔
401
402	::rocksdb::Status RocksDatabase::commit_batch(Batch& batch) {	3,060✔
403	return db_->Write(write_options(), &batch);	3,060✔
404	}
405
406	std::unique_ptr<::rocksdb::Iterator> RocksDatabase::new_iterator(	10,729✔
407	std::string_view column_family) const {
408	return std::unique_ptr<::rocksdb::Iterator>(	5,360✔
409	db_->NewIterator(read_options(), column_family_handle(column_family)));	10,729✔
410	}
411
412	::rocksdb::Status RocksDatabase::compact(std::string_view column_family) {	48✔
413	::rocksdb::CompactRangeOptions opts;	48✔
414	opts.max_subcompactions = 8;	48✔
415	return db_->CompactRange(opts, column_family_handle(column_family), nullptr,	48✔
416	nullptr);	48!
417	}
418
UNCOV 419	::rocksdb::Status RocksDatabase::ingest_external_files(	×
420	std::string_view column_family,
421	const std::vector<std::string>& external_files, bool ingest_behind) {
UNCOV 422	if (external_files.empty()) {	×
UNCOV 423	return ::rocksdb::Status::OK();	×
424	}
UNCOV 425	::rocksdb::IngestExternalFileOptions opts;	×
426	// Rename same-FS staged SSTs instead of copying; RocksDB copies on failure.
NEW 427	opts.move_files = true;	×
428	// Offline bulk build with no live readers, so skip the snapshot/seqno sync.
NEW 429	opts.snapshot_consistency = false;	×
UNCOV 430	opts.allow_global_seqno = true;	×
431	// Keep the assigned seqno in the manifest instead of rewriting it into each
432	// SST (format_version 5 supports this), avoiding a per-SST write + fsync.
NEW 433	opts.write_global_seqno = false;	×
UNCOV 434	opts.allow_blocking_flush = true;	×
UNCOV 435	opts.ingest_behind = ingest_behind;	×
UNCOV 436	return db_->IngestExternalFile(column_family_handle(column_family),	×
UNCOV 437	external_files, opts);	×
438	}
439
440	::rocksdb::Status RocksDatabase::ingest_external_files_multi(	30✔
441	const std::vector<
442	std::pair<std::string_view, const std::vector<std::string>*>>& per_cf) {
443	::rocksdb::IngestExternalFileOptions opts;	30✔
444	opts.move_files = true;	30✔
445	opts.snapshot_consistency = false;	30✔
446	opts.allow_global_seqno = true;	30✔
447	opts.write_global_seqno = false;	30✔
448	opts.allow_blocking_flush = true;	30✔
449
450	std::vector<::rocksdb::IngestExternalFileArg> args;	30✔
451	args.reserve(per_cf.size());	30!
452	for (const auto& [cf_name, files] : per_cf) {	466✔
453	if (!files \|\| files->empty()) continue;	436!
454	::rocksdb::IngestExternalFileArg arg;	436✔
455	arg.column_family = column_family_handle(cf_name);	436!
456	arg.external_files = *files;	436!
457	arg.options = opts;	436✔
458	args.push_back(std::move(arg));	436!
459	}	436✔
460	if (args.empty()) return ::rocksdb::Status::OK();	30!
461	return db_->IngestExternalFiles(args);	30!
462	}	30✔
463
464	} // namespace dftracer::utils::rocksdb

llnl / dftracer-utils / 30002033718

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