From 63340a81fee0cdc4771b8323c3c933625e25e38f Mon Sep 17 00:00:00 2001 From: Binbin_Qian <15843875+binbinqian@user.noreply.gitee.com> Date: Thu, 12 Jun 2025 11:43:10 +0800 Subject: [PATCH] rocksdb interface port---rdb_index_merge.h --- storage/rocksdb/rdb_index_merge.h | 162 +++++++++++++++++++++++++++++- 1 file changed, 161 insertions(+), 1 deletion(-) diff --git a/storage/rocksdb/rdb_index_merge.h b/storage/rocksdb/rdb_index_merge.h index 9bdfc1d60..fa6f7f213 100644 --- a/storage/rocksdb/rdb_index_merge.h +++ b/storage/rocksdb/rdb_index_merge.h @@ -63,4 +63,164 @@ class Rdb_index_merge { ulonglong m_curr_offset; /* offset of the record pointer for the block */ ulonglong m_disk_start_offset; /* where the chunk starts on disk */ ulonglong m_disk_curr_offset; /* current offset on disk */ - ulonglong m_total_size; /* total # of data bytes in chunk */ \ No newline at end of file + ulonglong m_total_size; /* total # of data bytes in chunk */ + + void store_key_value(const rocksdb::Slice &key, const rocksdb::Slice &val) + MY_ATTRIBUTE((__nonnull__)); + + + void store_slice(const rocksdb::Slice &slice) MY_ATTRIBUTE((__nonnull__)); + + size_t prepare(File fd, ulonglong f_offset) MY_ATTRIBUTE((__nonnull__)); + + int read_next_chunk_from_disk(File fd) + MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + inline bool is_chunk_finished() const { + return m_curr_offset + m_disk_curr_offset - m_disk_start_offset == + m_total_size; + } + + inline bool has_space(uint64 needed) const { + return m_curr_offset + needed <= m_block_len; + } + + explicit merge_buf_info(const ulonglong merge_block_size) + : m_block(nullptr), + m_block_len(merge_block_size), + m_curr_offset(0), + m_disk_start_offset(0), + m_disk_curr_offset(0), + m_total_size(merge_block_size) { + /* Will throw an exception if it runs out of memory here */ + m_block = std::unique_ptr(new uchar[merge_block_size]); + + /* Initialize entire buffer to 0 to avoid valgrind errors */ + memset(m_block.get(), 0, merge_block_size); + } + }; + + /* Represents an entry in the heap during merge phase of external sort */ + struct merge_heap_entry { + std::shared_ptr m_chunk_info; /* pointer to buffer info */ + uchar *m_block; /* pointer to heap memory where record is stored */ + const rocksdb::Comparator *const m_comparator; + rocksdb::Slice m_key; /* current key pointed to by block ptr */ + rocksdb::Slice m_val; + + size_t prepare(File fd, ulonglong f_offset, ulonglong chunk_size) + MY_ATTRIBUTE((__nonnull__)); + + int read_next_chunk_from_disk(File fd) + MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + int read_rec(rocksdb::Slice *const key, rocksdb::Slice *const val) + MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + int read_slice(rocksdb::Slice *const slice, const uchar **block_ptr) + MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + explicit merge_heap_entry(const rocksdb::Comparator *const comparator) + : m_chunk_info(nullptr), m_block(nullptr), m_comparator(comparator) {} + }; + + struct merge_heap_comparator { + bool operator()(const std::shared_ptr &lhs, + const std::shared_ptr &rhs) { + return lhs->m_comparator->Compare(rhs->m_key, lhs->m_key) < 0; + } + }; + + /* Represents a record in unsorted buffer */ + struct merge_record { + uchar *m_block; /* points to offset of key in sort buffer */ + const rocksdb::Comparator *const m_comparator; + + bool operator<(const merge_record &record) const { + return merge_record_compare(this->m_block, record.m_block, m_comparator) < + 0; + } + + merge_record(uchar *const block, + const rocksdb::Comparator *const comparator) + : m_block(block), m_comparator(comparator) {} + }; + + private: + const char *m_tmpfile_path; + const ulonglong m_merge_buf_size; + const ulonglong m_merge_combine_read_size; + const ulonglong m_merge_tmp_file_removal_delay; + rocksdb::ColumnFamilyHandle *m_cf_handle; + struct merge_file_info m_merge_file; + std::shared_ptr m_rec_buf_unsorted; + std::shared_ptr m_output_buf; + std::set m_offset_tree; + std::priority_queue, + std::vector>, + merge_heap_comparator> + m_merge_min_heap; + + static inline void merge_store_uint64(uchar *const dst, uint64 n) { + memcpy(dst, &n, sizeof(n)); + } + + static inline void merge_read_uint64(const uchar **buf_ptr, + uint64 *const dst) { + assert(buf_ptr != nullptr); + memcpy(dst, *buf_ptr, sizeof(uint64)); + *buf_ptr += sizeof(uint64); + } + + static inline rocksdb::Slice as_slice(const uchar *block) { + uint64 len; + merge_read_uint64(&block, &len); + + return rocksdb::Slice(reinterpret_cast(block), len); + } + + static int merge_record_compare(const uchar *a_block, const uchar *b_block, + const rocksdb::Comparator *const comparator) + MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + void merge_read_rec(const uchar *const block, rocksdb::Slice *const key, + rocksdb::Slice *const val) MY_ATTRIBUTE((__nonnull__)); + + void read_slice(rocksdb::Slice *slice, const uchar *block_ptr) + MY_ATTRIBUTE((__nonnull__)); + + public: + Rdb_index_merge(const char *const tmpfile_path, + const ulonglong merge_buf_size, + const ulonglong merge_combine_read_size, + const ulonglong merge_tmp_file_removal_delay, + rocksdb::ColumnFamilyHandle *cf); + ~Rdb_index_merge(); + + int init() MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + int merge_file_create() MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + int add(const rocksdb::Slice &key, const rocksdb::Slice &val) + MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + int merge_buf_write() MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + int next(rocksdb::Slice *const key, rocksdb::Slice *const val) + MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + int merge_heap_prepare() MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + void merge_heap_top(rocksdb::Slice *key, rocksdb::Slice *val) + MY_ATTRIBUTE((__nonnull__)); + + int merge_heap_pop_and_get_next(rocksdb::Slice *const key, + rocksdb::Slice *const val) + MY_ATTRIBUTE((__nonnull__, __warn_unused_result__)); + + void merge_reset(); + + rocksdb::ColumnFamilyHandle *get_cf() const { return m_cf_handle; } +}; + +} // namespace myrocks -- Gitee