be/src/storage/segment/vertical_segment_writer.h
Line | Count | Source |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #pragma once |
19 | | |
20 | | #include <gen_cpp/olap_file.pb.h> |
21 | | #include <gen_cpp/segment_v2.pb.h> |
22 | | |
23 | | #include <cstddef> |
24 | | #include <cstdint> |
25 | | #include <map> |
26 | | #include <memory> // unique_ptr |
27 | | #include <string> |
28 | | #include <unordered_set> |
29 | | #include <utility> |
30 | | #include <vector> |
31 | | |
32 | | #include "common/status.h" // Status |
33 | | #include "storage/index/index_file_writer.h" |
34 | | #include "storage/key/row_key_encoder.h" |
35 | | #include "storage/olap_define.h" |
36 | | #include "storage/segment/column_writer.h" |
37 | | #include "storage/segment/segment_index_file_cache_loader.h" |
38 | | #include "storage/tablet/tablet.h" |
39 | | #include "storage/tablet/tablet_schema.h" |
40 | | #include "util/faststring.h" |
41 | | #include "util/slice.h" |
42 | | |
43 | | namespace doris { |
44 | | class Block; |
45 | | class IOlapColumnDataAccessor; |
46 | | class OlapBlockDataConvertor; |
47 | | |
48 | | // TODO(lingbin): Should be a conf that can be dynamically adjusted, or a member in the context |
49 | | const uint32_t MAX_SEGMENT_SIZE = static_cast<uint32_t>(OLAP_MAX_COLUMN_SEGMENT_FILE_SIZE * |
50 | | OLAP_COLUMN_FILE_SEGMENT_SIZE_SCALE); |
51 | | class DataDir; |
52 | | class MemTracker; |
53 | | class ShortKeyIndexBuilder; |
54 | | class PrimaryKeyIndexBuilder; |
55 | | class KeyCoder; |
56 | | struct RowsetWriterContext; |
57 | | |
58 | | namespace io { |
59 | | class FileWriter; |
60 | | class FileSystem; |
61 | | } // namespace io |
62 | | namespace segment_v2 { |
63 | | class IndexFileWriter; |
64 | | class VariantStatsCaculator; |
65 | | |
66 | | class DerivedColumnGenerator; |
67 | | // Matches block_transform.h: at most one derived column (the row-store column) |
68 | | // for each flush, held as a {cid, generator} pair; null generator means none. |
69 | | using DerivedColumn = std::pair<uint32_t, std::shared_ptr<const DerivedColumnGenerator>>; |
70 | | |
71 | | struct VerticalSegmentWriterOptions { |
72 | | uint32_t num_rows_per_block = 1024; |
73 | | // Caps the rows in one segment. Only unit tests set it, to control how many |
74 | | // segments a write produces; production leaves it unlimited. |
75 | | uint32_t max_rows_per_segment = UINT32_MAX; |
76 | | bool enable_unique_key_merge_on_write = false; |
77 | | CompressionTypePB compression_type = UNKNOWN_COMPRESSION; |
78 | | |
79 | | RowsetWriterContext* rowset_ctx = nullptr; |
80 | | DataWriteType write_type = DataWriteType::TYPE_DEFAULT; |
81 | | }; |
82 | | |
83 | | // A segment is written one column group at a time: |
84 | | // |
85 | | // for (column_group : column_groups) { |
86 | | // writer.init(column_group, has_key); |
87 | | // writer.append_block(block, ...); // any number of times |
88 | | // writer.finalize_columns(&index_size); |
89 | | // } |
90 | | // writer.finalize_footer(&file_size); |
91 | | // |
92 | | // Vertical compaction and segcompaction write several groups; init() is |
93 | | // shorthand for one whole-schema group. finalize_columns() writes the |
94 | | // data pages the group still buffers, then its indexes, and closes the group. |
95 | | // |
96 | | // write_block() takes a whole-schema group's rows all at once and writes each |
97 | | // column's pages as soon as that column is done, so it opens the group itself |
98 | | // and needs no init(): |
99 | | // |
100 | | // writer.set_derived_column(derived_column); // only if there is one |
101 | | // writer.write_block(block, 0, num_rows); // opens the group itself |
102 | | // writer.finalize_columns(&index_size); |
103 | | // writer.finalize_footer(&file_size); |
104 | | class VerticalSegmentWriter { |
105 | | public: |
106 | | explicit VerticalSegmentWriter(io::FileWriter* file_writer, uint32_t segment_id, |
107 | | TabletSchemaSPtr tablet_schema, BaseTabletSPtr tablet, |
108 | | DataDir* data_dir, const VerticalSegmentWriterOptions& opts, |
109 | | IndexFileWriter* index_file_writer); |
110 | | ~VerticalSegmentWriter(); |
111 | | |
112 | | VerticalSegmentWriter(const VerticalSegmentWriter&) = delete; |
113 | | const VerticalSegmentWriter& operator=(const VerticalSegmentWriter&) = delete; |
114 | | |
115 | | Status init(); |
116 | | |
117 | | // Opens a column group; the footer keeps every group's column metas. Pass |
118 | | // has_key=true for the group with the key columns, which comes first. A |
119 | | // keyless schema has no such group, so its first value group takes it. |
120 | | Status init(const std::vector<uint32_t>& col_ids, bool has_key); |
121 | | |
122 | | // Feeds the open group. Its column writers buffer the data pages until |
123 | | // finalize_columns(), so more rows can always follow. |
124 | | Status append_block(const Block* block, size_t row_pos, size_t num_rows); |
125 | | |
126 | | // Opens a whole-schema group and writes it in one call, one column at a time, |
127 | | // so only one column's pages are in memory. |
128 | | Status write_block(const Block* block, size_t row_pos, size_t num_rows); |
129 | | |
130 | | // Sets the row-store column for the next write_block, which pumps it from its |
131 | | // generator in batches instead of reading it from the block. append_block |
132 | | // ignores it. |
133 | 181 | void set_derived_column(DerivedColumn derived_column) { |
134 | 181 | _derived_column = std::move(derived_column); |
135 | 181 | } |
136 | | |
137 | 0 | [[nodiscard]] std::string data_dir_path() const { |
138 | 0 | return _data_dir == nullptr ? "" : _data_dir->path(); |
139 | 0 | } |
140 | | |
141 | | // rows fed to the open group; zero again once the group is closed |
142 | 8.54k | [[nodiscard]] uint32_t num_rows_written() const { return _num_rows_written; } |
143 | | |
144 | | // the segment's row count, settled when the group holding the key columns is finalized |
145 | 5.50k | [[nodiscard]] uint32_t row_count() const { return _row_count; } |
146 | 3.79k | [[nodiscard]] uint32_t segment_id() const { return _segment_id; } |
147 | | |
148 | | // Size readings the append_block feed rolls a segment on: how many more |
149 | | // rows fit, what the segment holds right now, and the cluster key MOW |
150 | | // primary keys still waiting for the finalize-time sort. |
151 | | int64_t max_row_to_add(size_t row_avg_size_in_bytes); |
152 | | uint64_t estimate_segment_size(); |
153 | 0 | [[nodiscard]] uint64_t primary_keys_size() const { return _primary_keys_size; } |
154 | | |
155 | | // Closes the open group: its buffered data pages (skipped after write_block), |
156 | | // its index sections, and the key indexes if it holds the key columns. Ends |
157 | | // with clear(), so the next init() can follow. |
158 | | Status finalize_columns(uint64_t* index_size); |
159 | | // Ends the segment, once every group is in. |
160 | | Status finalize_footer(uint64_t* segment_file_size, |
161 | | SegmentIndexFileCacheInfo* index_file_cache_info = nullptr); |
162 | | |
163 | | Slice min_encoded_key(); |
164 | | Slice max_encoded_key(); |
165 | | |
166 | | void clear(); |
167 | | |
168 | 2.38k | Status close_inverted_index(int64_t* inverted_index_file_size) { |
169 | | // no inverted index |
170 | 2.38k | if (_index_file_writer == nullptr) { |
171 | 2.03k | *inverted_index_file_size = 0; |
172 | 2.03k | return Status::OK(); |
173 | 2.03k | } |
174 | 357 | RETURN_IF_ERROR(_index_file_writer->begin_close()); |
175 | 357 | *inverted_index_file_size = _index_file_writer->get_index_file_total_size(); |
176 | 357 | return Status::OK(); |
177 | 357 | } |
178 | | |
179 | | private: |
180 | | friend class TestVerticalSegmentWriter; |
181 | | void _abandon_index_staging(); |
182 | | void _init_column_meta(ColumnMetaPB* meta, uint32_t column_id, const TabletColumn& column, |
183 | | const ColumnWriterOptions& opts); |
184 | | // pos is the column's place in the open group, which is what the column |
185 | | // writers and the convertor are indexed by; cid is its schema column id, |
186 | | // which only the footer meta records. |
187 | | Status _create_column_writer(size_t pos, uint32_t cid, const TabletSchemaSPtr& tablet_schema); |
188 | | // Opens a column group without its column writers: init() then creates |
189 | | // them all at once, write_block() one at a time. |
190 | | Status _open_group(const std::vector<uint32_t>& col_ids, bool has_key); |
191 | | Status _create_writers(const TabletSchemaSPtr& tablet_schema, |
192 | | const std::vector<uint32_t>& col_ids); |
193 | | std::vector<uint32_t> _all_column_ids() const; |
194 | | // bytes the key index builders hold |
195 | | uint64_t _key_index_size(); |
196 | | // cluster key MOW: sorts the collected primary keys, hands them to the index |
197 | | // builder and releases them. The builder decides what reaches the file. |
198 | | Status _sort_primary_keys_into_index(); |
199 | | Status _set_row_count(); |
200 | | // Closes the open group's column writers and writes their data pages, after |
201 | | // checking the data dir has room for them. |
202 | | Status _finalize_columns_data(); |
203 | | Status _write_data(); |
204 | | Status _write_ordinal_index(); |
205 | | Status _write_zone_map(); |
206 | | Status _write_inverted_index(); |
207 | | Status _write_ann_index(); |
208 | | Status _write_bloom_filter_index(); |
209 | | Status _write_short_key_index(); |
210 | | Status _write_primary_key_index(); |
211 | | Status _write_footer(); |
212 | | Status _write_raw_data(const std::vector<Slice>& slices); |
213 | | void _set_min_max_key(const Slice& key); |
214 | | void _set_min_key(const Slice& key); |
215 | | void _set_max_key(const Slice& key); |
216 | | Status _append_generated_column(const DerivedColumnGenerator& generator, const Block& block, |
217 | | size_t row_pos, size_t num_rows, uint32_t cid); |
218 | | // Remembers the accessor if this column is a cluster key. _generate_key_index |
219 | | // needs them in cluster key order, which is not the group's order. |
220 | | void _collect_cluster_key_column( |
221 | | uint32_t cid, IOlapColumnDataAccessor* column, |
222 | | std::map<uint32_t, IOlapColumnDataAccessor*>* cluster_key_columns); |
223 | | Status _generate_key_index( |
224 | | std::vector<IOlapColumnDataAccessor*>& key_columns, IOlapColumnDataAccessor* seq_column, |
225 | | size_t num_rows, |
226 | | const std::map<uint32_t, IOlapColumnDataAccessor*>& cluster_key_columns); |
227 | | Status _generate_primary_key_index( |
228 | | const std::vector<IOlapColumnDataAccessor*>& primary_key_columns, |
229 | | IOlapColumnDataAccessor* seq_column, size_t num_rows, bool need_sort); |
230 | | Status _generate_short_key_index(std::vector<IOlapColumnDataAccessor*>& key_columns, |
231 | | size_t num_rows, const std::vector<size_t>& short_key_pos); |
232 | | Status _check_column_writer_disk_capacity(size_t cid); |
233 | | Status _finalize_column_writer_and_update_meta(size_t cid); |
234 | | |
235 | 362k | bool _is_mow() { |
236 | 362k | return _tablet_schema->keys_type() == UNIQUE_KEYS && _opts.enable_unique_key_merge_on_write; |
237 | 362k | } |
238 | 186k | bool _is_mow_with_cluster_key() { |
239 | 186k | return _is_mow() && !_tablet_schema->cluster_key_uids().empty(); |
240 | 186k | } |
241 | | |
242 | | private: |
243 | | uint32_t _segment_id; |
244 | | TabletSchemaSPtr _tablet_schema; |
245 | | BaseTabletSPtr _tablet; |
246 | | DataDir* _data_dir = nullptr; |
247 | | VerticalSegmentWriterOptions _opts; |
248 | | |
249 | | // Not owned. owned by RowsetWriter or SegmentFlusher |
250 | | io::FileWriter* _file_writer = nullptr; |
251 | | // Not owned. owned by RowsetWriter or SegmentFlusher |
252 | | IndexFileWriter* _index_file_writer = nullptr; |
253 | | |
254 | | SegmentFooterPB _footer; |
255 | | SegmentIndexFileCacheInfo _index_file_cache_info; |
256 | | size_t _num_short_key_columns; |
257 | | |
258 | | std::unique_ptr<ShortKeyIndexBuilder> _short_key_index_builder; |
259 | | std::unique_ptr<PrimaryKeyIndexBuilder> _primary_key_index_builder; |
260 | | std::vector<std::unique_ptr<ColumnWriter>> _column_writers; |
261 | | std::unique_ptr<MemTracker> _mem_tracker; |
262 | | |
263 | | std::unique_ptr<OlapBlockDataConvertor> _olap_data_convertor; |
264 | | // used for building short key index or primary key index during vectorized write. |
265 | | // NOTE: must stay declared after _tablet_schema and _opts, the constructor |
266 | | // init list reads both through _is_mow(). |
267 | | RowKeyEncoder _key_encoder; |
268 | | size_t _short_key_row_pos = 0; |
269 | | |
270 | | // The open column group: the columns it holds, and whether it is the one |
271 | | // carrying the key columns. A whole-schema group (init() or write_block) is |
272 | | // the degenerate case -- every column, has_key=true. |
273 | | std::vector<uint32_t> _column_ids; |
274 | | bool _has_key = true; |
275 | | // _num_rows_written means row count already written in this current column group |
276 | | uint32_t _num_rows_written = 0; |
277 | | |
278 | | // _row_count means total row count of this segment |
279 | | // In vertical compaction row count is recorded when key columns group finish |
280 | | // and _num_rows_written will be updated in value column group |
281 | | uint32_t _row_count = 0; |
282 | | |
283 | | bool _is_first_row = true; |
284 | | faststring _min_key; |
285 | | faststring _max_key; |
286 | | |
287 | | // Cluster key MOW only: primary keys collected here until the group's rows |
288 | | // are all in, then _sort_primary_keys_into_index() hands them over. The |
289 | | // rowset writer rolls a segment on their byte count. |
290 | | std::vector<std::string> _primary_keys; |
291 | | uint64_t _primary_keys_size = 0; |
292 | | // variant statistics calculator for efficient stats collection; only the |
293 | | // compaction write type feeds it |
294 | | std::unique_ptr<VariantStatsCaculator> _variant_stats_calculator; |
295 | | |
296 | | // the derived column the transform chain hands off to write_block's bounded pump |
297 | | DerivedColumn _derived_column; |
298 | | |
299 | | // true once write_block has written the open group's data pages, so |
300 | | // finalize_columns() does not write them again; clear() resets it |
301 | | bool _columns_data_flushed = false; |
302 | | }; |
303 | | |
304 | | } // namespace segment_v2 |
305 | | } // namespace doris |