1/*
2 Copyright (c) 2016, Facebook, Inc.
3
4 This program is free software; you can redistribute it and/or modify
5 it under the terms of the GNU General Public License as published by
6 the Free Software Foundation; version 2 of the License.
7
8 This program is distributed in the hope that it will be useful,
9 but WITHOUT ANY WARRANTY; without even the implied warranty of
10 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11 GNU General Public License for more details.
12
13 You should have received a copy of the GNU General Public License
14 along with this program; if not, write to the Free Software
15 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
16
17#pragma once
18
19/* MySQL header files */
20#include "../sql/log.h"
21#include "./handler.h" /* handler */
22#include "./my_global.h" /* ulonglong */
23
24/* C++ standard header files */
25#include <queue>
26#include <set>
27#include <vector>
28
29/* RocksDB header files */
30#include "rocksdb/db.h"
31
32/* MyRocks header files */
33#include "./rdb_comparator.h"
34
35namespace myrocks {
36
37/*
38 Length of delimiters used during inplace index creation.
39*/
40#define RDB_MERGE_CHUNK_LEN sizeof(size_t)
41#define RDB_MERGE_REC_DELIMITER sizeof(size_t)
42#define RDB_MERGE_KEY_DELIMITER RDB_MERGE_REC_DELIMITER
43#define RDB_MERGE_VAL_DELIMITER RDB_MERGE_REC_DELIMITER
44
45class Rdb_key_def;
46class Rdb_tbl_def;
47
48class Rdb_index_merge {
49 Rdb_index_merge(const Rdb_index_merge &p) = delete;
50 Rdb_index_merge &operator=(const Rdb_index_merge &p) = delete;
51
52 public:
53 /* Information about temporary files used in external merge sort */
54 struct merge_file_info {
55 File m_fd = -1; /* file descriptor */
56 ulong m_num_sort_buffers = 0; /* number of sort buffers in temp file */
57 };
58
59 /* Buffer for sorting in main memory. */
60 struct merge_buf_info {
61 /* heap memory allocated for main memory sort/merge */
62 std::unique_ptr<uchar[]> m_block;
63 const ulonglong
64 m_block_len; /* amount of data bytes allocated for block above */
65 ulonglong m_curr_offset; /* offset of the record pointer for the block */
66 ulonglong m_disk_start_offset; /* where the chunk starts on disk */
67 ulonglong m_disk_curr_offset; /* current offset on disk */
68 ulonglong m_total_size; /* total # of data bytes in chunk */
69
70 void store_key_value(const rocksdb::Slice &key, const rocksdb::Slice &val)
71 MY_ATTRIBUTE((__nonnull__));
72
73 void store_slice(const rocksdb::Slice &slice) MY_ATTRIBUTE((__nonnull__));
74
75 size_t prepare(File fd, ulonglong f_offset) MY_ATTRIBUTE((__nonnull__));
76
77 int read_next_chunk_from_disk(File fd)
78 MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
79
80 inline bool is_chunk_finished() const {
81 return m_curr_offset + m_disk_curr_offset - m_disk_start_offset ==
82 m_total_size;
83 }
84
85 inline bool has_space(uint64 needed) const {
86 return m_curr_offset + needed <= m_block_len;
87 }
88
89 explicit merge_buf_info(const ulonglong merge_block_size)
90 : m_block(nullptr), m_block_len(merge_block_size), m_curr_offset(0),
91 m_disk_start_offset(0), m_disk_curr_offset(0),
92 m_total_size(merge_block_size) {
93 /* Will throw an exception if it runs out of memory here */
94 m_block = std::unique_ptr<uchar[]>(new uchar[merge_block_size]);
95
96 /* Initialize entire buffer to 0 to avoid valgrind errors */
97 memset(m_block.get(), 0, merge_block_size);
98 }
99 };
100
101 /* Represents an entry in the heap during merge phase of external sort */
102 struct merge_heap_entry {
103 std::shared_ptr<merge_buf_info> m_chunk_info; /* pointer to buffer info */
104 uchar *m_block; /* pointer to heap memory where record is stored */
105 const rocksdb::Comparator *const m_comparator;
106 rocksdb::Slice m_key; /* current key pointed to by block ptr */
107 rocksdb::Slice m_val;
108
109 size_t prepare(File fd, ulonglong f_offset, ulonglong chunk_size)
110 MY_ATTRIBUTE((__nonnull__));
111
112 int read_next_chunk_from_disk(File fd)
113 MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
114
115 int read_rec(rocksdb::Slice *const key, rocksdb::Slice *const val)
116 MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
117
118 int read_slice(rocksdb::Slice *const slice, const uchar **block_ptr)
119 MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
120
121 explicit merge_heap_entry(const rocksdb::Comparator *const comparator)
122 : m_chunk_info(nullptr), m_block(nullptr), m_comparator(comparator) {}
123 };
124
125 struct merge_heap_comparator {
126 bool operator()(const std::shared_ptr<merge_heap_entry> &lhs,
127 const std::shared_ptr<merge_heap_entry> &rhs) {
128 return lhs->m_comparator->Compare(rhs->m_key, lhs->m_key) < 0;
129 }
130 };
131
132 /* Represents a record in unsorted buffer */
133 struct merge_record {
134 uchar *m_block; /* points to offset of key in sort buffer */
135 const rocksdb::Comparator *const m_comparator;
136
137 bool operator<(const merge_record &record) const {
138 return merge_record_compare(this->m_block, record.m_block, m_comparator) <
139 0;
140 }
141
142 merge_record(uchar *const block,
143 const rocksdb::Comparator *const comparator)
144 : m_block(block), m_comparator(comparator) {}
145 };
146
147 private:
148 const char *m_tmpfile_path;
149 const ulonglong m_merge_buf_size;
150 const ulonglong m_merge_combine_read_size;
151 const ulonglong m_merge_tmp_file_removal_delay;
152 rocksdb::ColumnFamilyHandle *m_cf_handle;
153 struct merge_file_info m_merge_file;
154 std::shared_ptr<merge_buf_info> m_rec_buf_unsorted;
155 std::shared_ptr<merge_buf_info> m_output_buf;
156 std::set<merge_record> m_offset_tree;
157 std::priority_queue<std::shared_ptr<merge_heap_entry>,
158 std::vector<std::shared_ptr<merge_heap_entry>>,
159 merge_heap_comparator>
160 m_merge_min_heap;
161
162 static inline void merge_store_uint64(uchar *const dst, uint64 n) {
163 memcpy(dst, &n, sizeof(n));
164 }
165
166 static inline void merge_read_uint64(const uchar **buf_ptr,
167 uint64 *const dst) {
168 DBUG_ASSERT(buf_ptr != nullptr);
169 memcpy(dst, *buf_ptr, sizeof(uint64));
170 *buf_ptr += sizeof(uint64);
171 }
172
173 static inline rocksdb::Slice as_slice(const uchar *block) {
174 uint64 len;
175 merge_read_uint64(&block, &len);
176
177 return rocksdb::Slice(reinterpret_cast<const char *>(block), len);
178 }
179
180 static int merge_record_compare(const uchar *a_block, const uchar *b_block,
181 const rocksdb::Comparator *const comparator)
182 MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
183
184 void merge_read_rec(const uchar *const block, rocksdb::Slice *const key,
185 rocksdb::Slice *const val) MY_ATTRIBUTE((__nonnull__));
186
187 void read_slice(rocksdb::Slice *slice, const uchar *block_ptr)
188 MY_ATTRIBUTE((__nonnull__));
189
190 public:
191 Rdb_index_merge(const char *const tmpfile_path,
192 const ulonglong &merge_buf_size,
193 const ulonglong &merge_combine_read_size,
194 const ulonglong &merge_tmp_file_removal_delay,
195 rocksdb::ColumnFamilyHandle *cf);
196 ~Rdb_index_merge();
197
198 int init() MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
199
200 int merge_file_create() MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
201
202 int add(const rocksdb::Slice &key, const rocksdb::Slice &val)
203 MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
204
205 int merge_buf_write() MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
206
207 int next(rocksdb::Slice *const key, rocksdb::Slice *const val)
208 MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
209
210 int merge_heap_prepare() MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
211
212 void merge_heap_top(rocksdb::Slice *key, rocksdb::Slice *val)
213 MY_ATTRIBUTE((__nonnull__));
214
215 int merge_heap_pop_and_get_next(rocksdb::Slice *const key,
216 rocksdb::Slice *const val)
217 MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
218
219 void merge_reset();
220
221 rocksdb::ColumnFamilyHandle *get_cf() const { return m_cf_handle; }
222};
223
224} // namespace myrocks
225