be/src/storage/segment/rle_page.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 "common/cast_set.h" |
21 | | #include "storage/field_type.h" // for FieldType |
22 | | #include "storage/segment/options.h" // for PageBuilderOptions/PageDecoderOptions |
23 | | #include "storage/segment/page_builder.h" // for PageBuilder |
24 | | #include "storage/segment/page_decoder.h" // for PageDecoder |
25 | | #include "util/coding.h" // for encode_fixed32_le/decode_fixed32_le |
26 | | #include "util/rle_encoding.h" // for RleEncoder/RleDecoder |
27 | | #include "util/slice.h" // for OwnedSlice |
28 | | |
29 | | namespace doris { |
30 | | namespace segment_v2 { |
31 | | |
32 | | enum { RLE_PAGE_HEADER_SIZE = 4 }; |
33 | | |
34 | | // RLE builder for generic integer and bool types. What is missing is some way |
35 | | // to enforce that this can only be instantiated for INT and BOOL types. |
36 | | // |
37 | | // The page format is as follows: |
38 | | // |
39 | | // 1. Header: (4 bytes total) |
40 | | // |
41 | | // <num_elements> [32-bit] |
42 | | // The number of elements encoded in the page. |
43 | | // |
44 | | // NOTE: all on-disk ints are encoded little-endian |
45 | | // |
46 | | // 2. Element data |
47 | | // |
48 | | // The header is followed by the rle-encoded element data. |
49 | | // |
50 | | // This Rle encoding algorithm is only effective for repeated INT type and bool type, |
51 | | // It is not good for sequence number or random number. BitshufflePage is recommended |
52 | | // for these case. |
53 | | // |
54 | | // TODO(hkp): optimize rle algorithm |
55 | | template <FieldType Type> |
56 | | class RlePageBuilder : public PageBuilderHelper<RlePageBuilder<Type> > { |
57 | | public: |
58 | | using Self = RlePageBuilder<Type>; |
59 | | friend class PageBuilderHelper<Self>; |
60 | | |
61 | 13.3k | Status init() override { |
62 | 13.3k | switch (Type) { |
63 | 13.3k | case FieldType::OLAP_FIELD_TYPE_BOOL: { |
64 | 13.3k | _bit_width = 1; |
65 | 13.3k | break; |
66 | 0 | } |
67 | 0 | default: { |
68 | 0 | _bit_width = SIZE_OF_TYPE * 8; |
69 | 0 | break; |
70 | 0 | } |
71 | 13.3k | } |
72 | 13.3k | _rle_encoder = new RleEncoder<CppType>(&_buf, _bit_width); |
73 | 13.3k | return reset(); |
74 | 13.3k | } |
75 | | |
76 | 13.3k | ~RlePageBuilder() { delete _rle_encoder; } |
77 | | |
78 | 186k | bool is_page_full() override { return _rle_encoder->len() >= _options.data_page_size; } |
79 | | |
80 | 186k | Status add(const uint8_t* vals, size_t* count) override { |
81 | 186k | DCHECK(!_finished); |
82 | 186k | auto new_vals = reinterpret_cast<const CppType*>(vals); |
83 | 4.34M | for (int i = 0; i < *count; ++i) { |
84 | | // note: vals is not guaranteed to be aligned for now, thus memcpy here |
85 | 4.16M | CppType value; |
86 | 4.16M | memcpy(&value, &new_vals[i], SIZE_OF_TYPE); |
87 | 4.16M | _rle_encoder->Put(value); |
88 | 4.16M | } |
89 | | |
90 | 186k | _count += *count; |
91 | 186k | _raw_data_size += *count * SIZE_OF_TYPE; |
92 | 186k | return Status::OK(); |
93 | 186k | } |
94 | | |
95 | 12.2k | Status finish(OwnedSlice* slice) override { |
96 | 12.2k | DCHECK(!_finished); |
97 | 12.2k | _finished = true; |
98 | | // here should Flush first and then encode the count header |
99 | | // or it will lead to a bug if the header is less than 8 byte and the data is small |
100 | 12.2k | _rle_encoder->Flush(); |
101 | 12.2k | encode_fixed32_le(&_buf[0], cast_set<uint32_t>(_count)); |
102 | 12.2k | *slice = _buf.build(); |
103 | 12.2k | return Status::OK(); |
104 | 12.2k | } |
105 | | |
106 | 25.6k | Status reset() override { |
107 | 25.6k | RETURN_IF_CATCH_EXCEPTION({ |
108 | 25.6k | _count = 0; |
109 | 25.6k | _finished = false; |
110 | 25.6k | _raw_data_size = 0; |
111 | 25.6k | _rle_encoder->Clear(); |
112 | 25.6k | _rle_encoder->Reserve(RLE_PAGE_HEADER_SIZE, 0); |
113 | 25.6k | }); |
114 | 25.6k | return Status::OK(); |
115 | 25.6k | } |
116 | | |
117 | 0 | size_t count() const override { return _count; } |
118 | | |
119 | 532 | uint64_t size() const override { return _rle_encoder->len(); } |
120 | | |
121 | 12.3k | uint64_t get_raw_data_size() const override { return _raw_data_size; } |
122 | | |
123 | | private: |
124 | | RlePageBuilder(const PageBuilderOptions& options) |
125 | 13.3k | : _options(options), |
126 | 13.3k | _count(0), |
127 | 13.3k | _finished(false), |
128 | 13.3k | _bit_width(0), |
129 | 13.3k | _rle_encoder(nullptr) {} |
130 | | |
131 | | typedef typename TypeTraits<Type>::CppType CppType; |
132 | | enum { SIZE_OF_TYPE = TypeTraits<Type>::size }; |
133 | | |
134 | | PageBuilderOptions _options; |
135 | | size_t _count; |
136 | | bool _finished; |
137 | | int _bit_width; |
138 | | RleEncoder<CppType>* _rle_encoder = nullptr; |
139 | | faststring _buf; |
140 | | uint64_t _raw_data_size = 0; |
141 | | }; |
142 | | |
143 | | template <FieldType Type> |
144 | | class RlePageDecoder : public PageDecoder { |
145 | | public: |
146 | | RlePageDecoder(Slice slice, const PageDecoderOptions& options) |
147 | 26.3k | : _data(slice), |
148 | 26.3k | _options(options), |
149 | 26.3k | _parsed(false), |
150 | 26.3k | _num_elements(0), |
151 | 26.3k | _cur_index(0), |
152 | 26.3k | _bit_width(0) {} |
153 | | |
154 | 26.3k | Status init() override { |
155 | 26.3k | CHECK(!_parsed); |
156 | | |
157 | 26.3k | if (_data.size < RLE_PAGE_HEADER_SIZE) { |
158 | 0 | return Status::Corruption("not enough bytes for header in RleBitMapBlockDecoder"); |
159 | 0 | } |
160 | 26.3k | _num_elements = decode_fixed32_le((const uint8_t*)&_data[0]); |
161 | | |
162 | 26.3k | _parsed = true; |
163 | | |
164 | 26.3k | switch (Type) { |
165 | 26.3k | case FieldType::OLAP_FIELD_TYPE_BOOL: { |
166 | 26.3k | _bit_width = 1; |
167 | 26.3k | break; |
168 | 0 | } |
169 | 0 | default: { |
170 | 0 | _bit_width = SIZE_OF_TYPE * 8; |
171 | 0 | break; |
172 | 0 | } |
173 | 26.3k | } |
174 | | |
175 | 26.3k | _rle_decoder = |
176 | 26.3k | RleDecoder<CppType>((uint8_t*)_data.data + RLE_PAGE_HEADER_SIZE, |
177 | 26.3k | cast_set<int>(_data.size - RLE_PAGE_HEADER_SIZE), _bit_width); |
178 | | |
179 | 26.3k | RETURN_IF_ERROR(seek_to_position_in_page(0)); |
180 | 26.3k | return Status::OK(); |
181 | 26.3k | } |
182 | | |
183 | 258k | Status seek_to_position_in_page(size_t pos) override { |
184 | 18.4E | DCHECK(_parsed) << "Must call init()"; |
185 | 258k | DCHECK_LE(pos, _num_elements) |
186 | 0 | << "Tried to seek to " << pos << " which is > number of elements (" << _num_elements |
187 | 0 | << ") in the block!"; |
188 | | // If the block is empty (e.g. the column is filled with nulls), there is no data to seek. |
189 | 258k | if (_num_elements == 0) [[unlikely]] { |
190 | 1.99k | if (pos != 0) { |
191 | 0 | return Status::Error<ErrorCode::INTERNAL_ERROR, false>( |
192 | 0 | "seek pos {} is larger than total elements {}", pos, _num_elements); |
193 | 1.99k | } else { |
194 | 1.99k | return Status::OK(); |
195 | 1.99k | } |
196 | 1.99k | } |
197 | 256k | if (_cur_index == pos) { |
198 | | // No need to seek. |
199 | 25.1k | return Status::OK(); |
200 | 231k | } else if (_cur_index < pos) { |
201 | 229k | size_t nskip = pos - _cur_index; |
202 | 229k | _rle_decoder.Skip(nskip); |
203 | 229k | } else { |
204 | 1.68k | _rle_decoder = RleDecoder<CppType>((uint8_t*)_data.data + RLE_PAGE_HEADER_SIZE, |
205 | 1.68k | cast_set<int>(_data.size - RLE_PAGE_HEADER_SIZE), |
206 | 1.68k | _bit_width); |
207 | 1.68k | _rle_decoder.Skip(pos); |
208 | 1.68k | } |
209 | 231k | _cur_index = pos; |
210 | 231k | return Status::OK(); |
211 | 256k | } |
212 | | |
213 | 30.8k | Status next_batch(size_t* n, MutableColumnPtr& dst) override { |
214 | 30.8k | DCHECK(_parsed); |
215 | 30.8k | if (*n == 0 || _cur_index >= _num_elements) [[unlikely]] { |
216 | 0 | *n = 0; |
217 | 0 | return Status::OK(); |
218 | 0 | } |
219 | | |
220 | 30.8k | size_t to_fetch = std::min(*n, static_cast<size_t>(_num_elements - _cur_index)); |
221 | 30.8k | size_t remaining = to_fetch; |
222 | 30.8k | bool result = false; |
223 | 30.8k | CppType value; |
224 | 6.66M | while (remaining > 0) { |
225 | 6.63M | result = _rle_decoder.Get(&value); |
226 | 6.63M | DCHECK(result); |
227 | 6.63M | dst->insert_data((char*)(&value), SIZE_OF_TYPE); |
228 | 6.63M | remaining--; |
229 | 6.63M | } |
230 | | |
231 | 30.8k | _cur_index += to_fetch; |
232 | 30.8k | *n = to_fetch; |
233 | 30.8k | return Status::OK(); |
234 | 30.8k | } |
235 | | |
236 | | Status read_by_rowids(const rowid_t* rowids, ordinal_t page_first_ordinal, size_t* n, |
237 | 7.00k | MutableColumnPtr& dst) override { |
238 | 7.00k | DCHECK(_parsed); |
239 | 7.00k | if (*n == 0 || _cur_index >= _num_elements) [[unlikely]] { |
240 | 0 | *n = 0; |
241 | 0 | return Status::OK(); |
242 | 0 | } |
243 | | |
244 | 7.00k | auto total = *n; |
245 | 7.00k | bool result = false; |
246 | 7.00k | size_t read_count = 0; |
247 | 7.00k | CppType value; |
248 | 798k | for (size_t i = 0; i < total; ++i) { |
249 | 791k | ordinal_t ord = rowids[i] - page_first_ordinal; |
250 | 791k | if (UNLIKELY(ord >= _num_elements)) { |
251 | 0 | *n = read_count; |
252 | 0 | return Status::OK(); |
253 | 0 | } |
254 | | |
255 | 791k | _rle_decoder.Skip(ord - _cur_index); |
256 | 791k | _cur_index = ord; |
257 | | |
258 | 791k | result = _rle_decoder.Get(&value); |
259 | 791k | _cur_index++; |
260 | 791k | DCHECK(result); |
261 | 791k | dst->insert_data((char*)(&value), SIZE_OF_TYPE); |
262 | 791k | read_count++; |
263 | 791k | } |
264 | 7.00k | *n = read_count; |
265 | 7.00k | return Status::OK(); |
266 | 7.00k | } |
267 | | |
268 | 0 | size_t count() const override { return _num_elements; } |
269 | | |
270 | 219k | size_t current_index() const override { return _cur_index; } |
271 | | |
272 | | private: |
273 | | typedef typename TypeTraits<Type>::CppType CppType; |
274 | | enum { SIZE_OF_TYPE = TypeTraits<Type>::size }; |
275 | | |
276 | | Slice _data; |
277 | | PageDecoderOptions _options; |
278 | | bool _parsed; |
279 | | uint32_t _num_elements; |
280 | | size_t _cur_index; |
281 | | int _bit_width; |
282 | | RleDecoder<CppType> _rle_decoder; |
283 | | }; |
284 | | |
285 | | } // namespace segment_v2 |
286 | | } // namespace doris |