be/src/storage/segment/column_writer.cpp
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 | | #include "storage/segment/column_writer.h" |
19 | | |
20 | | #include <gen_cpp/segment_v2.pb.h> |
21 | | |
22 | | #include <algorithm> |
23 | | #include <cstring> |
24 | | #include <filesystem> |
25 | | #include <memory> |
26 | | |
27 | | #include "common/config.h" |
28 | | #include "common/logging.h" |
29 | | #include "core/data_type/data_type_agg_state.h" |
30 | | #include "core/data_type/data_type_factory.hpp" |
31 | | #include "core/types.h" |
32 | | #include "io/fs/file_writer.h" |
33 | | #include "storage/index/bloom_filter/bloom_filter_index_writer.h" |
34 | | #include "storage/index/inverted/inverted_index_writer.h" |
35 | | #include "storage/index/ordinal_page_index.h" |
36 | | #include "storage/index/zone_map/zone_map_index.h" |
37 | | #include "storage/olap_common.h" |
38 | | #include "storage/segment/encoding_info.h" |
39 | | #include "storage/segment/options.h" |
40 | | #include "storage/segment/page_builder.h" |
41 | | #include "storage/segment/page_io.h" |
42 | | #include "storage/segment/page_pointer.h" |
43 | | #include "storage/segment/variant/variant_column_writer_impl.h" |
44 | | #include "storage/tablet/tablet_schema.h" |
45 | | #include "storage/types.h" |
46 | | #include "util/block_compression.h" |
47 | | #include "util/debug_points.h" |
48 | | #include "util/faststring.h" |
49 | | #include "util/rle_encoding.h" |
50 | | #include "util/simd/bits.h" |
51 | | |
52 | | namespace doris::segment_v2 { |
53 | | |
54 | | class NullBitmapBuilder { |
55 | | public: |
56 | 461k | NullBitmapBuilder() : _has_null(false), _bitmap_buf(512), _rle_encoder(&_bitmap_buf, 1) {} |
57 | | |
58 | | explicit NullBitmapBuilder(size_t reserve_bits) |
59 | | : _has_null(false), |
60 | | _bitmap_buf(BitmapSize(reserve_bits)), |
61 | 0 | _rle_encoder(&_bitmap_buf, 1) {} |
62 | | |
63 | 2.34M | void reserve_for_write(size_t num_rows, size_t non_null_count) { |
64 | 2.34M | if (num_rows == 0) { |
65 | 0 | return; |
66 | 0 | } |
67 | 2.34M | if (non_null_count == 0 || (non_null_count == num_rows && !_has_null)) { |
68 | 378k | if (_bitmap_buf.capacity() < kSmallReserveBytes) { |
69 | 0 | _bitmap_buf.reserve(kSmallReserveBytes); |
70 | 0 | } |
71 | 378k | return; |
72 | 378k | } |
73 | 1.97M | size_t raw_bytes = BitmapSize(num_rows); |
74 | 1.97M | size_t run_est = std::min(num_rows, non_null_count * 2 + 1); |
75 | 1.97M | size_t run_bytes_est = run_est * kBytesPerRun + kReserveSlackBytes; |
76 | 1.97M | size_t raw_overhead = raw_bytes / 63 + 1; |
77 | 1.97M | size_t raw_est = raw_bytes + raw_overhead + kReserveSlackBytes; |
78 | 1.97M | size_t reserve_bytes = std::min(raw_est, run_bytes_est); |
79 | 1.97M | if (_bitmap_buf.capacity() < reserve_bytes) { |
80 | 9.51k | const size_t cap = _bitmap_buf.capacity(); |
81 | 9.51k | const size_t grow = cap + cap / 2; |
82 | 9.51k | const size_t new_cap = std::max(reserve_bytes, grow); |
83 | 9.51k | _bitmap_buf.reserve(new_cap); |
84 | 9.51k | } |
85 | 1.97M | } |
86 | | |
87 | 5.98M | void add_run(bool value, size_t run) { |
88 | 5.98M | _has_null |= value; |
89 | 5.98M | _rle_encoder.Put(value, run); |
90 | 5.98M | } |
91 | | |
92 | | // Returns whether the building nullmap contains nullptr |
93 | 460k | bool has_null() const { return _has_null; } |
94 | | |
95 | 128k | Status finish(OwnedSlice* slice) { |
96 | 128k | _rle_encoder.Flush(); |
97 | 128k | RETURN_IF_CATCH_EXCEPTION({ *slice = _bitmap_buf.build(); }); |
98 | 128k | return Status::OK(); |
99 | 128k | } |
100 | | |
101 | 460k | void reset() { |
102 | 460k | _has_null = false; |
103 | 460k | _rle_encoder.Clear(); |
104 | 460k | } |
105 | | |
106 | 28.5k | uint64_t size() { return _bitmap_buf.size(); } |
107 | | |
108 | | private: |
109 | | static constexpr size_t kSmallReserveBytes = 64; |
110 | | static constexpr size_t kReserveSlackBytes = 16; |
111 | | static constexpr size_t kBytesPerRun = 6; |
112 | | |
113 | | bool _has_null; |
114 | | faststring _bitmap_buf; |
115 | | RleEncoder<bool> _rle_encoder; |
116 | | }; |
117 | | |
118 | | inline ScalarColumnWriter* get_null_writer(const ColumnWriterOptions& opts, |
119 | 66.1k | io::FileWriter* file_writer, uint32_t id) { |
120 | 66.1k | if (!opts.meta->is_nullable()) { |
121 | 29.5k | return nullptr; |
122 | 29.5k | } |
123 | | |
124 | 36.5k | FieldType null_type = FieldType::OLAP_FIELD_TYPE_TINYINT; |
125 | 36.5k | ColumnWriterOptions null_options; |
126 | 36.5k | null_options.meta = opts.meta->add_children_columns(); |
127 | 36.5k | null_options.meta->set_column_id(id); |
128 | 36.5k | null_options.meta->set_unique_id(id); |
129 | 36.5k | null_options.meta->set_type(int(null_type)); |
130 | 36.5k | null_options.meta->set_is_nullable(false); |
131 | 36.5k | null_options.meta->set_length( |
132 | 36.5k | cast_set<int32_t>(field_type_size(FieldType::OLAP_FIELD_TYPE_TINYINT))); |
133 | 36.5k | null_options.meta->set_compression(opts.meta->compression()); |
134 | | |
135 | 36.5k | null_options.need_zone_map = false; |
136 | 36.5k | null_options.need_bloom_filter = false; |
137 | 36.5k | null_options.storage_format = opts.storage_format; |
138 | | |
139 | 36.5k | auto null_column_ptr = std::make_shared<TabletColumn>( |
140 | 36.5k | FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, null_type, false, |
141 | 36.5k | null_options.meta->unique_id(), null_options.meta->length()); |
142 | 36.5k | null_column_ptr->set_name("nullable"); |
143 | 36.5k | null_column_ptr->set_index_length(-1); // no short key index |
144 | 36.5k | null_options.meta->set_encoding( |
145 | 36.5k | EncodingInfo::resolve_default_encoding(opts.storage_format, *null_column_ptr)); |
146 | 36.5k | return new ScalarColumnWriter(null_options, std::move(null_column_ptr), file_writer); |
147 | 66.1k | } |
148 | | |
149 | | ColumnWriter::ColumnWriter(TabletColumnPtr column, bool is_nullable, ColumnMetaPB* meta) |
150 | 939k | : _column(std::move(column)), _is_nullable(is_nullable), _column_meta(meta) { |
151 | 939k | _data_type = DataTypeFactory::instance().create_data_type(*_column_meta); |
152 | 939k | } |
153 | | Status ColumnWriter::create_struct_writer(const ColumnWriterOptions& opts, |
154 | | const TabletColumn* column, io::FileWriter* file_writer, |
155 | 2.12k | std::unique_ptr<ColumnWriter>* writer) { |
156 | | // not support empty struct |
157 | 2.12k | DCHECK(column->get_subtype_count() >= 1); |
158 | 2.12k | std::vector<std::unique_ptr<ColumnWriter>> sub_column_writers; |
159 | 2.12k | sub_column_writers.reserve(column->get_subtype_count()); |
160 | 11.3k | for (uint32_t i = 0; i < column->get_subtype_count(); i++) { |
161 | 9.26k | const TabletColumn& sub_column = column->get_sub_column(i); |
162 | 9.26k | RETURN_IF_ERROR(sub_column.check_valid()); |
163 | | |
164 | | // create sub writer |
165 | 9.26k | ColumnWriterOptions column_options; |
166 | 9.26k | column_options.meta = opts.meta->mutable_children_columns(i); |
167 | 9.26k | column_options.need_zone_map = false; |
168 | 9.26k | column_options.need_bloom_filter = sub_column.is_bf_column(); |
169 | 9.26k | column_options.storage_format = opts.storage_format; |
170 | 9.26k | std::unique_ptr<ColumnWriter> sub_column_writer; |
171 | 9.26k | RETURN_IF_ERROR( |
172 | 9.26k | ColumnWriter::create(column_options, &sub_column, file_writer, &sub_column_writer)); |
173 | 9.26k | sub_column_writers.push_back(std::move(sub_column_writer)); |
174 | 9.26k | } |
175 | | |
176 | 2.12k | ScalarColumnWriter* null_writer = |
177 | 2.12k | get_null_writer(opts, file_writer, column->get_subtype_count() + 1); |
178 | | |
179 | 2.12k | *writer = std::unique_ptr<ColumnWriter>(new StructColumnWriter( |
180 | 2.12k | opts, std::make_shared<TabletColumn>(*column), null_writer, sub_column_writers)); |
181 | 2.12k | return Status::OK(); |
182 | 2.12k | } |
183 | | |
184 | | Status ColumnWriter::create_array_writer(const ColumnWriterOptions& opts, |
185 | | const TabletColumn* column, io::FileWriter* file_writer, |
186 | 42.8k | std::unique_ptr<ColumnWriter>* writer) { |
187 | 42.8k | DCHECK(column->get_subtype_count() == 1); |
188 | 42.8k | const TabletColumn& item_column = column->get_sub_column(0); |
189 | 42.8k | RETURN_IF_ERROR(item_column.check_valid()); |
190 | | |
191 | | // create item writer |
192 | 42.8k | ColumnWriterOptions item_options; |
193 | 42.8k | item_options.meta = opts.meta->mutable_children_columns(0); |
194 | 42.8k | item_options.need_zone_map = false; |
195 | 42.8k | item_options.need_bloom_filter = item_column.is_bf_column(); |
196 | 42.8k | item_options.storage_format = opts.storage_format; |
197 | 42.8k | std::unique_ptr<ColumnWriter> item_writer; |
198 | 42.8k | RETURN_IF_ERROR(ColumnWriter::create(item_options, &item_column, file_writer, &item_writer)); |
199 | | |
200 | | // create length writer |
201 | 42.8k | FieldType length_type = FieldType::OLAP_FIELD_TYPE_UNSIGNED_BIGINT; |
202 | | |
203 | 42.8k | ColumnWriterOptions length_options; |
204 | 42.8k | length_options.meta = opts.meta->add_children_columns(); |
205 | 42.8k | length_options.meta->set_column_id(2); |
206 | 42.8k | length_options.meta->set_unique_id(2); |
207 | 42.8k | length_options.meta->set_type(int(length_type)); |
208 | 42.8k | length_options.meta->set_is_nullable(false); |
209 | 42.8k | length_options.meta->set_length( |
210 | 42.8k | cast_set<int32_t>(field_type_size(FieldType::OLAP_FIELD_TYPE_UNSIGNED_BIGINT))); |
211 | 42.8k | length_options.meta->set_compression(opts.meta->compression()); |
212 | | |
213 | 42.8k | length_options.need_zone_map = false; |
214 | 42.8k | length_options.need_bloom_filter = false; |
215 | 42.8k | length_options.storage_format = opts.storage_format; |
216 | | |
217 | 42.8k | auto length_column_ptr = std::make_shared<TabletColumn>( |
218 | 42.8k | FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, length_type, |
219 | 42.8k | length_options.meta->is_nullable(), length_options.meta->unique_id(), |
220 | 42.8k | length_options.meta->length()); |
221 | 42.8k | length_column_ptr->set_name("length"); |
222 | 42.8k | length_column_ptr->set_index_length(-1); // no short key index |
223 | 42.8k | length_options.meta->set_encoding( |
224 | 42.8k | EncodingInfo::resolve_default_encoding(opts.storage_format, *length_column_ptr)); |
225 | 42.8k | auto* length_writer = |
226 | 42.8k | new OffsetColumnWriter(length_options, std::move(length_column_ptr), file_writer); |
227 | | |
228 | 42.8k | ScalarColumnWriter* null_writer = get_null_writer(opts, file_writer, 3); |
229 | | |
230 | 42.8k | *writer = std::unique_ptr<ColumnWriter>( |
231 | 42.8k | new ArrayColumnWriter(opts, std::make_shared<TabletColumn>(*column), length_writer, |
232 | 42.8k | null_writer, std::move(item_writer))); |
233 | 42.8k | return Status::OK(); |
234 | 42.8k | } |
235 | | |
236 | | Status ColumnWriter::create_map_writer(const ColumnWriterOptions& opts, const TabletColumn* column, |
237 | | io::FileWriter* file_writer, |
238 | 21.1k | std::unique_ptr<ColumnWriter>* writer) { |
239 | 21.1k | DCHECK(column->get_subtype_count() == 2); |
240 | 21.1k | if (column->get_subtype_count() < 2) { |
241 | 0 | return Status::InternalError( |
242 | 0 | "If you upgraded from version 1.2.*, please DROP the MAP columns and then " |
243 | 0 | "ADD the MAP columns back."); |
244 | 0 | } |
245 | | // create key & value writer |
246 | 21.1k | std::vector<std::unique_ptr<ColumnWriter>> inner_writer_list; |
247 | 63.5k | for (int i = 0; i < 2; ++i) { |
248 | 42.3k | const TabletColumn& item_column = column->get_sub_column(i); |
249 | 42.3k | RETURN_IF_ERROR(item_column.check_valid()); |
250 | | |
251 | | // create item writer |
252 | 42.3k | ColumnWriterOptions item_options; |
253 | 42.3k | item_options.meta = opts.meta->mutable_children_columns(i); |
254 | 42.3k | item_options.need_zone_map = false; |
255 | 42.3k | item_options.need_bloom_filter = item_column.is_bf_column(); |
256 | 42.3k | item_options.storage_format = opts.storage_format; |
257 | 42.3k | std::unique_ptr<ColumnWriter> item_writer; |
258 | 42.3k | RETURN_IF_ERROR( |
259 | 42.3k | ColumnWriter::create(item_options, &item_column, file_writer, &item_writer)); |
260 | 42.3k | inner_writer_list.push_back(std::move(item_writer)); |
261 | 42.3k | } |
262 | | |
263 | | // create offset writer |
264 | 21.1k | FieldType length_type = FieldType::OLAP_FIELD_TYPE_UNSIGNED_BIGINT; |
265 | | |
266 | | // Be Cautious: column unique id is used for column reader creation |
267 | 21.1k | ColumnWriterOptions length_options; |
268 | 21.1k | length_options.meta = opts.meta->add_children_columns(); |
269 | 21.1k | length_options.meta->set_column_id(column->get_subtype_count() + 1); |
270 | 21.1k | length_options.meta->set_unique_id(column->get_subtype_count() + 1); |
271 | 21.1k | length_options.meta->set_type(int(length_type)); |
272 | 21.1k | length_options.meta->set_is_nullable(false); |
273 | 21.1k | length_options.meta->set_length( |
274 | 21.1k | cast_set<int32_t>(field_type_size(FieldType::OLAP_FIELD_TYPE_UNSIGNED_BIGINT))); |
275 | 21.1k | length_options.meta->set_compression(opts.meta->compression()); |
276 | | |
277 | 21.1k | length_options.need_zone_map = false; |
278 | 21.1k | length_options.need_bloom_filter = false; |
279 | 21.1k | length_options.storage_format = opts.storage_format; |
280 | | |
281 | 21.1k | auto length_column_ptr = std::make_shared<TabletColumn>( |
282 | 21.1k | FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, length_type, |
283 | 21.1k | length_options.meta->is_nullable(), length_options.meta->unique_id(), |
284 | 21.1k | length_options.meta->length()); |
285 | 21.1k | length_column_ptr->set_name("length"); |
286 | 21.1k | length_column_ptr->set_index_length(-1); // no short key index |
287 | 21.1k | length_options.meta->set_encoding( |
288 | 21.1k | EncodingInfo::resolve_default_encoding(opts.storage_format, *length_column_ptr)); |
289 | 21.1k | auto* length_writer = |
290 | 21.1k | new OffsetColumnWriter(length_options, std::move(length_column_ptr), file_writer); |
291 | | |
292 | 21.1k | ScalarColumnWriter* null_writer = |
293 | 21.1k | get_null_writer(opts, file_writer, column->get_subtype_count() + 2); |
294 | | |
295 | 21.1k | *writer = std::unique_ptr<ColumnWriter>( |
296 | 21.1k | new MapColumnWriter(opts, std::make_shared<TabletColumn>(*column), null_writer, |
297 | 21.1k | length_writer, inner_writer_list)); |
298 | | |
299 | 21.1k | return Status::OK(); |
300 | 21.1k | } |
301 | | |
302 | | Status ColumnWriter::create_agg_state_writer(const ColumnWriterOptions& opts, |
303 | | const TabletColumn* column, |
304 | | io::FileWriter* file_writer, |
305 | 1.26k | std::unique_ptr<ColumnWriter>* writer) { |
306 | 1.26k | auto data_type = DataTypeFactory::instance().create_data_type(*column); |
307 | 1.26k | const auto* agg_state_type = assert_cast<const DataTypeAggState*>(data_type.get()); |
308 | 1.26k | auto type = agg_state_type->get_serialized_type()->get_primitive_type(); |
309 | 1.26k | if (type == PrimitiveType::TYPE_STRING || type == PrimitiveType::INVALID_TYPE || |
310 | 1.26k | type == PrimitiveType::TYPE_FIXED_LENGTH_OBJECT || type == PrimitiveType::TYPE_BITMAP) { |
311 | 1.19k | *writer = std::unique_ptr<ColumnWriter>( |
312 | 1.19k | new ScalarColumnWriter(opts, std::make_shared<TabletColumn>(*column), file_writer)); |
313 | 1.19k | } else if (type == PrimitiveType::TYPE_ARRAY) { |
314 | 22 | RETURN_IF_ERROR(create_array_writer(opts, column, file_writer, writer)); |
315 | 54 | } else if (type == PrimitiveType::TYPE_MAP) { |
316 | 54 | RETURN_IF_ERROR(create_map_writer(opts, column, file_writer, writer)); |
317 | 18.4E | } else { |
318 | 18.4E | throw Exception(ErrorCode::INTERNAL_ERROR, |
319 | 18.4E | "OLAP_FIELD_TYPE_AGG_STATE meet unsupported type: {}", |
320 | 18.4E | agg_state_type->get_name()); |
321 | 18.4E | } |
322 | 1.27k | return Status::OK(); |
323 | 1.26k | } |
324 | | |
325 | | Status ColumnWriter::create_variant_writer(const ColumnWriterOptions& opts, |
326 | | const TabletColumn* column, io::FileWriter* file_writer, |
327 | 6.19k | std::unique_ptr<ColumnWriter>* writer) { |
328 | | // Variant extracted columns have two kinds of physical writers: |
329 | | // - Doc-value snapshot column (`...__DORIS_VARIANT_DOC_VALUE__...`): use `VariantDocCompactWriter` |
330 | | // to store the doc snapshot in a compact binary form. |
331 | | // - Regular extracted subcolumns: use `VariantSubcolumnWriter`. |
332 | | // The root VARIANT column itself uses `VariantColumnWriter`. |
333 | 6.19k | if (column->is_extracted_column()) { |
334 | 291 | if (column->name().find(DOC_VALUE_COLUMN_PATH) != std::string::npos) { |
335 | 282 | *writer = std::make_unique<VariantDocCompactWriter>( |
336 | 282 | opts, std::make_shared<TabletColumn>(*column)); |
337 | 282 | return Status::OK(); |
338 | 282 | } |
339 | 9 | VLOG_DEBUG << "gen subwriter for " << column->path_info_ptr()->get_path(); |
340 | 9 | *writer = std::make_unique<VariantSubcolumnWriter>(opts, |
341 | 9 | std::make_shared<TabletColumn>(*column)); |
342 | 9 | return Status::OK(); |
343 | 291 | } |
344 | 5.90k | *writer = std::make_unique<VariantColumnWriter>(opts, std::make_shared<TabletColumn>(*column)); |
345 | 5.90k | return Status::OK(); |
346 | 6.19k | } |
347 | | |
348 | | //Todo(Amory): here should according nullable and offset and need sub to simply this function |
349 | | Status ColumnWriter::create(const ColumnWriterOptions& opts, const TabletColumn* column, |
350 | 820k | io::FileWriter* file_writer, std::unique_ptr<ColumnWriter>* writer) { |
351 | 820k | auto column_ptr = std::make_shared<TabletColumn>(*column); |
352 | 820k | if (is_scalar_type(column->type())) { |
353 | 758k | *writer = std::unique_ptr<ColumnWriter>( |
354 | 758k | new ScalarColumnWriter(opts, std::move(column_ptr), file_writer)); |
355 | 758k | return Status::OK(); |
356 | 758k | } else { |
357 | 61.7k | switch (column->type()) { |
358 | 1.27k | case FieldType::OLAP_FIELD_TYPE_AGG_STATE: { |
359 | 1.27k | RETURN_IF_ERROR(create_agg_state_writer(opts, column, file_writer, writer)); |
360 | 1.27k | return Status::OK(); |
361 | 1.27k | } |
362 | 2.12k | case FieldType::OLAP_FIELD_TYPE_STRUCT: { |
363 | 2.12k | RETURN_IF_ERROR(create_struct_writer(opts, column, file_writer, writer)); |
364 | 2.12k | return Status::OK(); |
365 | 2.12k | } |
366 | 42.8k | case FieldType::OLAP_FIELD_TYPE_ARRAY: { |
367 | 42.8k | RETURN_IF_ERROR(create_array_writer(opts, column, file_writer, writer)); |
368 | 42.8k | return Status::OK(); |
369 | 42.8k | } |
370 | 9.86k | case FieldType::OLAP_FIELD_TYPE_MAP: { |
371 | 9.86k | RETURN_IF_ERROR(create_map_writer(opts, column, file_writer, writer)); |
372 | 9.86k | return Status::OK(); |
373 | 9.86k | } |
374 | 6.21k | case FieldType::OLAP_FIELD_TYPE_VARIANT: { |
375 | | // Process columns with sparse column |
376 | 6.21k | RETURN_IF_ERROR(create_variant_writer(opts, column, file_writer, writer)); |
377 | 6.21k | return Status::OK(); |
378 | 6.21k | } |
379 | 0 | default: |
380 | 0 | return Status::NotSupported("unsupported type for ColumnWriter: {}", |
381 | 0 | std::to_string(int(column_ptr->type()))); |
382 | 61.7k | } |
383 | 61.7k | } |
384 | 820k | } |
385 | | |
386 | | Status ColumnWriter::append_nullable(const uint8_t* is_null_bits, const void* data, |
387 | 651k | size_t num_rows) { |
388 | 651k | const auto* ptr = (const uint8_t*)data; |
389 | 651k | BitmapIterator null_iter(is_null_bits, num_rows); |
390 | 651k | bool is_null = false; |
391 | 651k | size_t this_run = 0; |
392 | 1.30M | while ((this_run = null_iter.Next(&is_null)) > 0) { |
393 | 651k | if (is_null) { |
394 | 0 | RETURN_IF_ERROR(append_nulls(this_run)); |
395 | 651k | } else { |
396 | 651k | RETURN_IF_ERROR(append_data(&ptr, this_run)); |
397 | 651k | } |
398 | 651k | } |
399 | 651k | return Status::OK(); |
400 | 651k | } |
401 | | |
402 | | Status ColumnWriter::append_nullable(const uint8_t* null_map, const uint8_t** ptr, |
403 | 0 | size_t num_rows) { |
404 | | // Fast path: use SIMD to detect all-NULL or all-non-NULL columns |
405 | 0 | if (config::enable_rle_batch_put_optimization) { |
406 | 0 | size_t non_null_count = |
407 | 0 | simd::count_zero_num(reinterpret_cast<const int8_t*>(null_map), num_rows); |
408 | |
|
409 | 0 | if (non_null_count == 0) { |
410 | | // All NULL: skip run-length iteration, directly append all nulls |
411 | 0 | RETURN_IF_ERROR(append_nulls(num_rows)); |
412 | 0 | *ptr += cell_size() * num_rows; |
413 | 0 | return Status::OK(); |
414 | 0 | } |
415 | | |
416 | 0 | if (non_null_count == num_rows) { |
417 | | // All non-NULL: skip run-length iteration, directly append all data |
418 | 0 | return append_data(ptr, num_rows); |
419 | 0 | } |
420 | 0 | } |
421 | | |
422 | | // Mixed case or sparse optimization disabled: use run-length processing |
423 | 0 | size_t offset = 0; |
424 | 0 | auto next_run_step = [&]() { |
425 | 0 | size_t step = 1; |
426 | 0 | for (auto i = offset + 1; i < num_rows; ++i) { |
427 | 0 | if (null_map[offset] == null_map[i]) { |
428 | 0 | step++; |
429 | 0 | } else { |
430 | 0 | break; |
431 | 0 | } |
432 | 0 | } |
433 | 0 | return step; |
434 | 0 | }; |
435 | |
|
436 | 0 | do { |
437 | 0 | auto step = next_run_step(); |
438 | 0 | if (null_map[offset]) { |
439 | 0 | RETURN_IF_ERROR(append_nulls(step)); |
440 | 0 | *ptr += cell_size() * step; |
441 | 0 | } else { |
442 | | // TODO: |
443 | | // 1. `*ptr += cell_size() * step;` should do in this function, not append_data; |
444 | | // 2. support array vectorized load and ptr offset add |
445 | 0 | RETURN_IF_ERROR(append_data(ptr, step)); |
446 | 0 | } |
447 | 0 | offset += step; |
448 | 0 | } while (offset < num_rows); |
449 | | |
450 | 0 | return Status::OK(); |
451 | 0 | } |
452 | | |
453 | 2.73M | Status ColumnWriter::append(const uint8_t* nullmap, const void* data, size_t num_rows) { |
454 | 2.73M | assert(data && num_rows > 0); |
455 | 2.73M | const auto* ptr = (const uint8_t*)data; |
456 | 2.73M | if (nullmap) { |
457 | 2.38M | return append_nullable(nullmap, &ptr, num_rows); |
458 | 2.38M | } else { |
459 | 349k | return append_data(&ptr, num_rows); |
460 | 349k | } |
461 | 2.73M | } |
462 | | |
463 | | /////////////////////////////////////////////////////////////////////////////////// |
464 | | |
465 | | ScalarColumnWriter::ScalarColumnWriter(const ColumnWriterOptions& opts, TabletColumnPtr column, |
466 | | io::FileWriter* file_writer) |
467 | 866k | : ColumnWriter(std::move(column), opts.meta->is_nullable(), opts.meta), |
468 | 866k | _opts(opts), |
469 | 866k | _file_writer(file_writer), |
470 | 866k | _data_size(0) { |
471 | | // these opts.meta fields should be set by client |
472 | 866k | DCHECK(opts.meta->has_column_id()); |
473 | 866k | DCHECK(opts.meta->has_unique_id()); |
474 | 866k | DCHECK(opts.meta->has_type()); |
475 | 866k | DCHECK(opts.meta->has_length()); |
476 | 866k | DCHECK(opts.meta->has_encoding()); |
477 | 866k | DCHECK(opts.meta->has_compression()); |
478 | 866k | DCHECK(opts.meta->has_is_nullable()); |
479 | 866k | DCHECK(file_writer != nullptr); |
480 | 866k | _inverted_index_builders.resize(_opts.inverted_indexes.size()); |
481 | 866k | } |
482 | | |
483 | 868k | ScalarColumnWriter::~ScalarColumnWriter() { |
484 | | // delete all pages |
485 | 868k | _pages.clear(); |
486 | 868k | } |
487 | | |
488 | 865k | Status ScalarColumnWriter::init() { |
489 | 865k | RETURN_IF_ERROR(get_block_compression_codec(_opts.meta->compression(), &_compress_codec)); |
490 | | |
491 | 865k | PageBuilder* page_builder = nullptr; |
492 | | |
493 | | // Caller must set a concrete (non-DEFAULT) encoding on the meta before init. |
494 | 865k | if (_opts.meta->encoding() == DEFAULT_ENCODING) { |
495 | 0 | return Status::InternalError( |
496 | 0 | "ColumnMetaPB encoding is DEFAULT_ENCODING for column_id={}, type={}; caller must " |
497 | 0 | "resolve to a concrete encoding before ScalarColumnWriter::init", |
498 | 0 | _opts.meta->column_id(), get_column()->type()); |
499 | 0 | } |
500 | 865k | RETURN_IF_ERROR( |
501 | 865k | EncodingInfo::get(get_column()->type(), _opts.meta->encoding(), &_encoding_info)); |
502 | | // create page builder |
503 | 865k | PageBuilderOptions opts; |
504 | 865k | opts.data_page_size = _opts.data_page_size; |
505 | 865k | opts.dict_page_size = _opts.dict_page_size; |
506 | | // V3 segments store the dictionary word page (and the dict-overflow fallback plain page) |
507 | | // with the V3 binary plain layout; pre-V3 segments keep V1. |
508 | 865k | opts.dict_binary_plain_encoding = |
509 | 865k | (_opts.storage_format == TabletStorageFormatPB::TABLET_STORAGE_FORMAT_V3) |
510 | 865k | ? PLAIN_ENCODING_V3 |
511 | 865k | : PLAIN_ENCODING; |
512 | 865k | RETURN_IF_ERROR(_encoding_info->create_page_builder(opts, &page_builder)); |
513 | 865k | if (page_builder == nullptr) { |
514 | 0 | return Status::NotSupported("Failed to create page builder for type {} and encoding {}", |
515 | 0 | get_column()->type(), _opts.meta->encoding()); |
516 | 0 | } |
517 | 18.4E | VLOG_DEBUG << fmt::format( |
518 | 18.4E | "[verbose] scalar column writer init, column_id={}, type={}, encoding={}, " |
519 | 18.4E | "is_nullable={}", |
520 | 18.4E | _opts.meta->column_id(), get_column()->type(), |
521 | 18.4E | EncodingTypePB_Name(_opts.meta->encoding()), _opts.meta->is_nullable()); |
522 | 865k | _page_builder.reset(page_builder); |
523 | | // create ordinal builder |
524 | 865k | _ordinal_index_builder = std::make_unique<OrdinalIndexWriter>(); |
525 | | // create null bitmap builder |
526 | 865k | if (is_nullable()) { |
527 | 461k | _null_bitmap_builder = std::make_unique<NullBitmapBuilder>(); |
528 | 461k | } |
529 | 865k | if (_opts.need_zone_map) { |
530 | 617k | RETURN_IF_ERROR( |
531 | 617k | ZoneMapIndexWriter::create(_data_type, get_column(), _zone_map_index_builder)); |
532 | 617k | } |
533 | | |
534 | 865k | if (_opts.need_inverted_index) { |
535 | 38.1k | do { |
536 | 76.5k | for (size_t i = 0; i < _opts.inverted_indexes.size(); i++) { |
537 | 38.3k | DBUG_EXECUTE_IF("column_writer.init", { |
538 | 38.3k | class InvertedIndexColumnWriterEmpty final : public IndexColumnWriter { |
539 | 38.3k | public: |
540 | 38.3k | Status init() override { return Status::OK(); } |
541 | 38.3k | Status add_values(const std::string name, const void* values, |
542 | 38.3k | size_t count) override { |
543 | 38.3k | return Status::OK(); |
544 | 38.3k | } |
545 | 38.3k | Status add_array_values(size_t field_size, const void* value_ptr, |
546 | 38.3k | const uint8_t* null_map, const uint8_t* offsets_ptr, |
547 | 38.3k | size_t count) override { |
548 | 38.3k | return Status::OK(); |
549 | 38.3k | } |
550 | 38.3k | Status add_nulls(uint32_t count) override { return Status::OK(); } |
551 | 38.3k | Status add_array_nulls(const uint8_t* null_map, size_t num_rows) override { |
552 | 38.3k | return Status::OK(); |
553 | 38.3k | } |
554 | 38.3k | Status finish() override { return Status::OK(); } |
555 | 38.3k | int64_t size() const override { return 0; } |
556 | 38.3k | void close_on_error() override {} |
557 | 38.3k | }; |
558 | | |
559 | 38.3k | _inverted_index_builders[i] = |
560 | 38.3k | std::make_unique<InvertedIndexColumnWriterEmpty>(); |
561 | | |
562 | 38.3k | break; |
563 | 38.3k | }); |
564 | | |
565 | 38.3k | RETURN_IF_ERROR(IndexColumnWriter::create( |
566 | 38.3k | get_column(), &_inverted_index_builders[i], _opts.index_file_writer, |
567 | 38.3k | _opts.inverted_indexes[i])); |
568 | 38.3k | } |
569 | 38.1k | } while (false); |
570 | 38.1k | } |
571 | 865k | if (_opts.need_bloom_filter) { |
572 | 5.48k | if (_opts.is_ngram_bf_index) { |
573 | 3.37k | RETURN_IF_ERROR(NGramBloomFilterIndexWriterImpl::create( |
574 | 3.37k | BloomFilterOptions(), get_column()->type(), _opts.gram_size, _opts.gram_bf_size, |
575 | 3.37k | &_bloom_filter_index_builder)); |
576 | 3.37k | } else { |
577 | 2.10k | RETURN_IF_ERROR(BloomFilterIndexWriter::create(_opts.bf_options, get_column()->type(), |
578 | 2.10k | &_bloom_filter_index_builder)); |
579 | 2.10k | } |
580 | 5.48k | } |
581 | 865k | return Status::OK(); |
582 | 865k | } |
583 | | |
584 | 2.54M | Status ScalarColumnWriter::append_nulls(size_t num_rows) { |
585 | 2.54M | _null_bitmap_builder->add_run(true, num_rows); |
586 | 2.54M | _next_rowid += num_rows; |
587 | 2.54M | if (_opts.need_zone_map) { |
588 | 2.47M | _zone_map_index_builder->add_nulls(cast_set<uint32_t>(num_rows)); |
589 | 2.47M | } |
590 | 2.54M | if (_opts.need_inverted_index) { |
591 | 1.62M | for (const auto& builder : _inverted_index_builders) { |
592 | 1.62M | RETURN_IF_ERROR(builder->add_nulls(cast_set<uint32_t>(num_rows))); |
593 | 1.62M | } |
594 | 1.62M | } |
595 | 2.54M | if (_opts.need_bloom_filter) { |
596 | 593 | _bloom_filter_index_builder->add_nulls(cast_set<uint32_t>(num_rows)); |
597 | 593 | } |
598 | 2.54M | return Status::OK(); |
599 | 2.54M | } |
600 | | |
601 | | // append data to page builder. this function will make sure that |
602 | | // num_rows must be written before return. And ptr will be modified |
603 | | // to next data should be written |
604 | 3.78M | Status ScalarColumnWriter::append_data(const uint8_t** ptr, size_t num_rows) { |
605 | 3.78M | size_t remaining = num_rows; |
606 | 7.65M | while (remaining > 0) { |
607 | 3.86M | size_t num_written = remaining; |
608 | 3.86M | RETURN_IF_ERROR(append_data_in_current_page(ptr, &num_written)); |
609 | | |
610 | 3.86M | remaining -= num_written; |
611 | | |
612 | 3.86M | if (_page_builder->is_page_full()) { |
613 | 76.5k | RETURN_IF_ERROR(finish_current_page()); |
614 | 76.5k | } |
615 | 3.86M | } |
616 | 3.78M | return Status::OK(); |
617 | 3.78M | } |
618 | | |
619 | | Status ScalarColumnWriter::_internal_append_data_in_current_page(const uint8_t* data, |
620 | 3.92M | size_t* num_written) { |
621 | 3.92M | RETURN_IF_ERROR(_page_builder->add(data, num_written)); |
622 | 3.92M | if (_opts.need_zone_map) { |
623 | 3.63M | _zone_map_index_builder->add_values(data, *num_written); |
624 | 3.63M | } |
625 | 3.92M | if (_opts.need_inverted_index) { |
626 | 1.64M | for (const auto& builder : _inverted_index_builders) { |
627 | 1.64M | RETURN_IF_ERROR(builder->add_values(get_column()->name(), data, *num_written)); |
628 | 1.64M | } |
629 | 1.64M | } |
630 | 3.92M | if (_opts.need_bloom_filter) { |
631 | 5.79k | RETURN_IF_ERROR(_bloom_filter_index_builder->add_values(data, *num_written)); |
632 | 5.79k | } |
633 | | |
634 | 3.92M | _next_rowid += *num_written; |
635 | | |
636 | | // we must write null bits after write data, because we don't |
637 | | // know how many rows can be written into current page |
638 | 3.92M | if (is_nullable()) { |
639 | 3.44M | _null_bitmap_builder->add_run(false, *num_written); |
640 | 3.44M | } |
641 | 3.92M | return Status::OK(); |
642 | 3.92M | } |
643 | | |
644 | 3.92M | Status ScalarColumnWriter::append_data_in_current_page(const uint8_t** data, size_t* num_written) { |
645 | 3.92M | RETURN_IF_ERROR(append_data_in_current_page(*data, num_written)); |
646 | 3.92M | *data += cell_size() * (*num_written); |
647 | 3.92M | return Status::OK(); |
648 | 3.92M | } |
649 | | |
650 | | Status ScalarColumnWriter::append_nullable(const uint8_t* null_map, const uint8_t** ptr, |
651 | 2.34M | size_t num_rows) { |
652 | | // When optimization is disabled, use base class implementation |
653 | 2.34M | if (!config::enable_rle_batch_put_optimization) { |
654 | 0 | return ColumnWriter::append_nullable(null_map, ptr, num_rows); |
655 | 0 | } |
656 | | |
657 | 2.34M | if (UNLIKELY(num_rows == 0)) { |
658 | 0 | return Status::OK(); |
659 | 0 | } |
660 | | |
661 | | // Build run-length encoded null runs using memchr for fast boundary detection |
662 | 2.34M | _null_run_buffer.clear(); |
663 | 2.34M | if (_null_run_buffer.capacity() < num_rows) { |
664 | 471k | _null_run_buffer.reserve(std::min(num_rows, size_t(256))); |
665 | 471k | } |
666 | | |
667 | 2.34M | size_t non_null_count = 0; |
668 | 2.34M | size_t offset = 0; |
669 | 5.77M | while (offset < num_rows) { |
670 | 3.42M | bool is_null = null_map[offset] != 0; |
671 | 3.42M | size_t remaining = num_rows - offset; |
672 | 3.42M | const uint8_t* run_end = |
673 | 3.42M | static_cast<const uint8_t*>(memchr(null_map + offset, is_null ? 0 : 1, remaining)); |
674 | 3.42M | size_t run_length = run_end != nullptr ? (run_end - (null_map + offset)) : remaining; |
675 | 3.42M | _null_run_buffer.push_back(NullRun {is_null, static_cast<uint32_t>(run_length)}); |
676 | 3.42M | if (!is_null) { |
677 | 2.78M | non_null_count += run_length; |
678 | 2.78M | } |
679 | 3.42M | offset += run_length; |
680 | 3.42M | } |
681 | | |
682 | | // Pre-allocate buffer based on estimated size |
683 | 2.35M | if (_null_bitmap_builder != nullptr) { |
684 | 2.35M | size_t current_rows = _next_rowid - _first_rowid; |
685 | 2.35M | size_t expected_rows = current_rows + num_rows; |
686 | 2.35M | size_t est_non_null = non_null_count; |
687 | 2.35M | if (num_rows > 0 && expected_rows > num_rows) { |
688 | 1.91M | est_non_null = (non_null_count * expected_rows) / num_rows; |
689 | 1.91M | } |
690 | 2.35M | _null_bitmap_builder->reserve_for_write(expected_rows, est_non_null); |
691 | 2.35M | } |
692 | | |
693 | 2.34M | if (non_null_count == 0) { |
694 | | // All NULL: skip data writing, only update null bitmap and indexes |
695 | 54.4k | RETURN_IF_ERROR(append_nulls(num_rows)); |
696 | 54.4k | *ptr += cell_size() * num_rows; |
697 | 54.4k | return Status::OK(); |
698 | 54.4k | } |
699 | | |
700 | 2.29M | if (non_null_count == num_rows) { |
701 | | // All non-NULL: use normal append_data which handles both data and null bitmap |
702 | 2.22M | return append_data(ptr, num_rows); |
703 | 2.22M | } |
704 | | |
705 | | // Process by runs |
706 | 1.14M | for (const auto& run : _null_run_buffer) { |
707 | 1.14M | size_t run_length = run.len; |
708 | 1.14M | if (run.is_null) { |
709 | 588k | RETURN_IF_ERROR(append_nulls(run_length)); |
710 | 588k | *ptr += cell_size() * run_length; |
711 | 588k | } else { |
712 | | // TODO: |
713 | | // 1. `*ptr += cell_size() * step;` should do in this function, not append_data; |
714 | | // 2. support array vectorized load and ptr offset add |
715 | 560k | RETURN_IF_ERROR(append_data(ptr, run_length)); |
716 | 560k | } |
717 | 1.14M | } |
718 | | |
719 | 68.5k | return Status::OK(); |
720 | 68.5k | } |
721 | | |
722 | 42.6k | uint64_t ScalarColumnWriter::estimate_buffer_size() { |
723 | 42.6k | uint64_t size = _data_size; |
724 | 42.6k | size += _page_builder->size(); |
725 | 42.6k | if (is_nullable()) { |
726 | 28.5k | size += _null_bitmap_builder->size(); |
727 | 28.5k | } |
728 | 42.6k | size += _ordinal_index_builder->size(); |
729 | 42.6k | if (_opts.need_zone_map) { |
730 | 32.4k | size += _zone_map_index_builder->size(); |
731 | 32.4k | } |
732 | 42.6k | if (_opts.need_bloom_filter) { |
733 | 267 | size += _bloom_filter_index_builder->size(); |
734 | 267 | } |
735 | 42.6k | return size; |
736 | 42.6k | } |
737 | | |
738 | 868k | Status ScalarColumnWriter::finish() { |
739 | 868k | RETURN_IF_ERROR(finish_current_page()); |
740 | 868k | _opts.meta->set_num_rows(_next_rowid); |
741 | 868k | return Status::OK(); |
742 | 868k | } |
743 | | |
744 | 868k | Status ScalarColumnWriter::write_data() { |
745 | 868k | auto offset = _file_writer->bytes_appended(); |
746 | 1.16M | auto collect_uncompressed_bytes = [](const PageFooterPB& footer) { |
747 | 1.16M | return footer.uncompressed_size() + footer.ByteSizeLong() + |
748 | 1.16M | sizeof(uint32_t) /* footer size */ + sizeof(uint32_t) /* checksum */; |
749 | 1.16M | }; |
750 | 909k | for (auto& page : _pages) { |
751 | 909k | _total_uncompressed_data_pages_size += collect_uncompressed_bytes(page->footer); |
752 | 909k | RETURN_IF_ERROR(_write_data_page(page.get())); |
753 | 909k | } |
754 | 868k | _pages.clear(); |
755 | | // write column dict |
756 | 868k | if (_encoding_info->encoding() == DICT_ENCODING) { |
757 | 258k | OwnedSlice dict_body; |
758 | 258k | RETURN_IF_ERROR(_page_builder->get_dictionary_page(&dict_body)); |
759 | 258k | EncodingTypePB dict_word_page_encoding; |
760 | 258k | RETURN_IF_ERROR(_page_builder->get_dictionary_page_encoding(&dict_word_page_encoding)); |
761 | | |
762 | 258k | PageFooterPB footer; |
763 | 258k | footer.set_type(DICTIONARY_PAGE); |
764 | 258k | footer.set_uncompressed_size(cast_set<uint32_t>(dict_body.slice().get_size())); |
765 | 258k | footer.mutable_dict_page_footer()->set_encoding(dict_word_page_encoding); |
766 | 258k | _total_uncompressed_data_pages_size += collect_uncompressed_bytes(footer); |
767 | | |
768 | 258k | PagePointer dict_pp; |
769 | 258k | RETURN_IF_ERROR(PageIO::compress_and_write_page( |
770 | 258k | _compress_codec, _opts.compression_min_space_saving, _file_writer, |
771 | 258k | {dict_body.slice()}, footer, &dict_pp)); |
772 | 258k | dict_pp.to_proto(_opts.meta->mutable_dict_page()); |
773 | 258k | } |
774 | 868k | _total_compressed_data_pages_size += _file_writer->bytes_appended() - offset; |
775 | 868k | _page_builder.reset(); |
776 | 868k | return Status::OK(); |
777 | 868k | } |
778 | | |
779 | 832k | Status ScalarColumnWriter::write_ordinal_index() { |
780 | 832k | return _ordinal_index_builder->finish(_file_writer, _opts.meta->add_indexes()); |
781 | 832k | } |
782 | | |
783 | 669k | Status ScalarColumnWriter::write_zone_map() { |
784 | 669k | if (_opts.need_zone_map) { |
785 | 617k | return _zone_map_index_builder->finish(_file_writer, _opts.meta->add_indexes()); |
786 | 617k | } |
787 | 52.5k | return Status::OK(); |
788 | 669k | } |
789 | | |
790 | 611k | Status ScalarColumnWriter::write_inverted_index() { |
791 | 611k | if (_opts.need_inverted_index) { |
792 | 38.3k | for (const auto& builder : _inverted_index_builders) { |
793 | 38.3k | RETURN_IF_ERROR(builder->finish()); |
794 | 38.3k | } |
795 | 38.1k | } |
796 | 611k | return Status::OK(); |
797 | 611k | } |
798 | | |
799 | 604k | Status ScalarColumnWriter::write_bloom_filter_index() { |
800 | 604k | if (_opts.need_bloom_filter) { |
801 | 5.47k | return _bloom_filter_index_builder->finish(_file_writer, _opts.meta->add_indexes()); |
802 | 5.47k | } |
803 | 599k | return Status::OK(); |
804 | 604k | } |
805 | | |
806 | | // write a data page into file and update ordinal index |
807 | 909k | Status ScalarColumnWriter::_write_data_page(Page* page) { |
808 | 909k | PagePointer pp; |
809 | 909k | std::vector<Slice> compressed_body; |
810 | 1.71M | for (auto& data : page->data) { |
811 | 1.71M | compressed_body.push_back(data.slice()); |
812 | 1.71M | } |
813 | 909k | RETURN_IF_ERROR(PageIO::write_page(_file_writer, compressed_body, page->footer, &pp)); |
814 | 909k | _ordinal_index_builder->append_entry(page->footer.data_page_footer().first_ordinal(), pp); |
815 | 909k | return Status::OK(); |
816 | 909k | } |
817 | | |
818 | 945k | Status ScalarColumnWriter::finish_current_page() { |
819 | 945k | if (_next_rowid == _first_rowid) { |
820 | 35.9k | return Status::OK(); |
821 | 35.9k | } |
822 | 909k | if (_opts.need_zone_map) { |
823 | | // If the number of rows in the current page is less than the threshold, |
824 | | // we will invalidate zone map index for this page by set pass_all to true. |
825 | 682k | if (_next_rowid - _first_rowid < config::zone_map_row_num_threshold) { |
826 | 478k | _zone_map_index_builder->invalid_page_zone_map(); |
827 | 478k | } |
828 | 682k | RETURN_IF_ERROR(_zone_map_index_builder->flush()); |
829 | 682k | } |
830 | | |
831 | 909k | if (_opts.need_bloom_filter) { |
832 | 5.72k | RETURN_IF_ERROR(_bloom_filter_index_builder->flush()); |
833 | 5.72k | } |
834 | | |
835 | 909k | _raw_data_bytes += _page_builder->get_raw_data_size(); |
836 | | |
837 | | // build data page body : encoded values + [nullmap] |
838 | 909k | std::vector<Slice> body; |
839 | 909k | OwnedSlice encoded_values; |
840 | 909k | RETURN_IF_ERROR(_page_builder->finish(&encoded_values)); |
841 | 909k | RETURN_IF_ERROR(_page_builder->reset()); |
842 | 909k | body.push_back(encoded_values.slice()); |
843 | | |
844 | 909k | OwnedSlice nullmap; |
845 | 909k | if (_null_bitmap_builder != nullptr) { |
846 | 460k | if (is_nullable() && _null_bitmap_builder->has_null()) { |
847 | 128k | RETURN_IF_ERROR(_null_bitmap_builder->finish(&nullmap)); |
848 | 128k | body.push_back(nullmap.slice()); |
849 | 128k | } |
850 | 460k | _null_bitmap_builder->reset(); |
851 | 460k | } |
852 | | |
853 | | // prepare data page footer |
854 | 909k | std::unique_ptr<Page> page(new Page()); |
855 | 909k | page->footer.set_type(DATA_PAGE); |
856 | 909k | page->footer.set_uncompressed_size(cast_set<uint32_t>(Slice::compute_total_size(body))); |
857 | 909k | auto* data_page_footer = page->footer.mutable_data_page_footer(); |
858 | 909k | data_page_footer->set_first_ordinal(_first_rowid); |
859 | 909k | data_page_footer->set_num_values(_next_rowid - _first_rowid); |
860 | 909k | data_page_footer->set_nullmap_size(cast_set<uint32_t>(nullmap.slice().size)); |
861 | 909k | if (_new_page_callback != nullptr) { |
862 | 63.8k | _new_page_callback->put_extra_info_in_page(data_page_footer); |
863 | 63.8k | } |
864 | | // trying to compress page body |
865 | 909k | OwnedSlice compressed_body; |
866 | 909k | RETURN_IF_ERROR(PageIO::compress_page_body(_compress_codec, _opts.compression_min_space_saving, |
867 | 909k | body, &compressed_body)); |
868 | 909k | if (compressed_body.slice().empty()) { |
869 | | // page body is uncompressed |
870 | 809k | page->data.emplace_back(std::move(encoded_values)); |
871 | 809k | page->data.emplace_back(std::move(nullmap)); |
872 | 809k | } else { |
873 | | // page body is compressed |
874 | 100k | page->data.emplace_back(std::move(compressed_body)); |
875 | 100k | } |
876 | | |
877 | 909k | _push_back_page(std::move(page)); |
878 | 909k | _first_rowid = _next_rowid; |
879 | 909k | return Status::OK(); |
880 | 909k | } |
881 | | |
882 | | //////////////////////////////////////////////////////////////////////////////// |
883 | | |
884 | | //////////////////////////////////////////////////////////////////////////////// |
885 | | // offset column writer |
886 | | //////////////////////////////////////////////////////////////////////////////// |
887 | | |
888 | | OffsetColumnWriter::OffsetColumnWriter(const ColumnWriterOptions& opts, TabletColumnPtr column, |
889 | | io::FileWriter* file_writer) |
890 | 64.0k | : ScalarColumnWriter(opts, std::move(column), file_writer) { |
891 | | // now we only explain data in offset column as uint64 |
892 | 64.0k | DCHECK(get_column()->type() == FieldType::OLAP_FIELD_TYPE_UNSIGNED_BIGINT); |
893 | 64.0k | } |
894 | | |
895 | 64.0k | OffsetColumnWriter::~OffsetColumnWriter() = default; |
896 | | |
897 | 63.9k | Status OffsetColumnWriter::init() { |
898 | 63.9k | RETURN_IF_ERROR(ScalarColumnWriter::init()); |
899 | 63.9k | register_flush_page_callback(this); |
900 | 63.9k | _next_offset = 0; |
901 | 63.9k | return Status::OK(); |
902 | 63.9k | } |
903 | | |
904 | 63.8k | Status OffsetColumnWriter::append_data(const uint8_t** ptr, size_t num_rows) { |
905 | 63.8k | size_t remaining = num_rows; |
906 | 128k | while (remaining > 0) { |
907 | 64.3k | size_t num_written = remaining; |
908 | 64.3k | RETURN_IF_ERROR(append_data_in_current_page(ptr, &num_written)); |
909 | | // Callers provide one extra tail offset after the written rows so the page footer can |
910 | | // store the next array item ordinal for the current page. |
911 | 64.3k | _next_offset = *(const uint64_t*)(*ptr); |
912 | 64.3k | remaining -= num_written; |
913 | | |
914 | 64.3k | if (_page_builder->is_page_full()) { |
915 | | // get next data for next array_item_rowid |
916 | 482 | RETURN_IF_ERROR(finish_current_page()); |
917 | 482 | } |
918 | 64.3k | } |
919 | 63.8k | return Status::OK(); |
920 | 63.8k | } |
921 | | |
922 | 63.8k | void OffsetColumnWriter::put_extra_info_in_page(DataPageFooterPB* footer) { |
923 | 63.8k | footer->set_next_array_item_ordinal(_next_offset); |
924 | 63.8k | } |
925 | | |
926 | | StructColumnWriter::StructColumnWriter( |
927 | | const ColumnWriterOptions& opts, TabletColumnPtr column, ScalarColumnWriter* null_writer, |
928 | | std::vector<std::unique_ptr<ColumnWriter>>& sub_column_writers) |
929 | 2.12k | : ColumnWriter(std::move(column), opts.meta->is_nullable(), opts.meta), _opts(opts) { |
930 | 9.26k | for (auto& sub_column_writer : sub_column_writers) { |
931 | 9.26k | _sub_column_writers.push_back(std::move(sub_column_writer)); |
932 | 9.26k | } |
933 | 2.12k | _num_sub_column_writers = _sub_column_writers.size(); |
934 | 2.12k | DCHECK(_num_sub_column_writers >= 1); |
935 | 2.12k | if (is_nullable()) { |
936 | 1.85k | _null_writer.reset(null_writer); |
937 | 1.85k | } |
938 | 2.12k | } |
939 | | |
940 | 2.12k | Status StructColumnWriter::init() { |
941 | 9.26k | for (auto& column_writer : _sub_column_writers) { |
942 | 9.26k | RETURN_IF_ERROR(column_writer->init()); |
943 | 9.26k | } |
944 | 2.12k | if (is_nullable()) { |
945 | 1.85k | RETURN_IF_ERROR(_null_writer->init()); |
946 | 1.85k | } |
947 | 2.12k | return Status::OK(); |
948 | 2.12k | } |
949 | | |
950 | 1.50k | Status StructColumnWriter::write_inverted_index() { |
951 | 1.50k | if (_opts.need_inverted_index) { |
952 | 0 | for (auto& column_writer : _sub_column_writers) { |
953 | 0 | RETURN_IF_ERROR(column_writer->write_inverted_index()); |
954 | 0 | } |
955 | 0 | } |
956 | 1.50k | return Status::OK(); |
957 | 1.50k | } |
958 | | |
959 | | Status StructColumnWriter::append_nullable(const uint8_t* null_map, const uint8_t** ptr, |
960 | 1.81k | size_t num_rows) { |
961 | 1.81k | RETURN_IF_ERROR(append_data(ptr, num_rows)); |
962 | 1.81k | RETURN_IF_ERROR(_null_writer->append_data(&null_map, num_rows)); |
963 | 1.81k | return Status::OK(); |
964 | 1.81k | } |
965 | | |
966 | 2.08k | Status StructColumnWriter::append_data(const uint8_t** ptr, size_t num_rows) { |
967 | 2.08k | const auto* results = reinterpret_cast<const uint64_t*>(*ptr); |
968 | 11.0k | for (size_t i = 0; i < _num_sub_column_writers; ++i) { |
969 | 8.92k | auto nullmap = *(results + _num_sub_column_writers + i); |
970 | 8.92k | auto data = *(results + i); |
971 | 8.92k | RETURN_IF_ERROR(_sub_column_writers[i]->append(reinterpret_cast<const uint8_t*>(nullmap), |
972 | 8.92k | reinterpret_cast<const void*>(data), |
973 | 8.92k | num_rows)); |
974 | 8.92k | } |
975 | 2.08k | return Status::OK(); |
976 | 2.08k | } |
977 | | |
978 | 186 | uint64_t StructColumnWriter::estimate_buffer_size() { |
979 | 186 | uint64_t size = 0; |
980 | 780 | for (auto& column_writer : _sub_column_writers) { |
981 | 780 | size += column_writer->estimate_buffer_size(); |
982 | 780 | } |
983 | 186 | size += is_nullable() ? _null_writer->estimate_buffer_size() : 0; |
984 | 186 | return size; |
985 | 186 | } |
986 | | |
987 | 2.12k | Status StructColumnWriter::finish() { |
988 | 9.26k | for (auto& column_writer : _sub_column_writers) { |
989 | 9.26k | RETURN_IF_ERROR(column_writer->finish()); |
990 | 9.26k | } |
991 | 2.12k | if (is_nullable()) { |
992 | 1.85k | RETURN_IF_ERROR(_null_writer->finish()); |
993 | 1.85k | } |
994 | 2.12k | _opts.meta->set_num_rows(get_next_rowid()); |
995 | 2.12k | return Status::OK(); |
996 | 2.12k | } |
997 | | |
998 | 2.12k | Status StructColumnWriter::write_data() { |
999 | 9.26k | for (auto& column_writer : _sub_column_writers) { |
1000 | 9.26k | RETURN_IF_ERROR(column_writer->write_data()); |
1001 | 9.26k | } |
1002 | 2.12k | if (is_nullable()) { |
1003 | 1.85k | RETURN_IF_ERROR(_null_writer->write_data()); |
1004 | 1.85k | } |
1005 | 2.12k | return Status::OK(); |
1006 | 2.12k | } |
1007 | | |
1008 | 2.08k | Status StructColumnWriter::write_ordinal_index() { |
1009 | 8.92k | for (auto& column_writer : _sub_column_writers) { |
1010 | 8.92k | RETURN_IF_ERROR(column_writer->write_ordinal_index()); |
1011 | 8.92k | } |
1012 | 2.08k | if (is_nullable()) { |
1013 | 1.81k | RETURN_IF_ERROR(_null_writer->write_ordinal_index()); |
1014 | 1.81k | } |
1015 | 2.08k | return Status::OK(); |
1016 | 2.08k | } |
1017 | | |
1018 | 0 | Status StructColumnWriter::append_nulls(size_t num_rows) { |
1019 | 0 | for (auto& column_writer : _sub_column_writers) { |
1020 | 0 | RETURN_IF_ERROR(column_writer->append_nulls(num_rows)); |
1021 | 0 | } |
1022 | 0 | if (is_nullable()) { |
1023 | 0 | std::vector<UInt8> null_signs(num_rows, 1); |
1024 | 0 | const uint8_t* null_sign_ptr = null_signs.data(); |
1025 | 0 | RETURN_IF_ERROR(_null_writer->append_data(&null_sign_ptr, num_rows)); |
1026 | 0 | } |
1027 | 0 | return Status::OK(); |
1028 | 0 | } |
1029 | | |
1030 | 0 | Status StructColumnWriter::finish_current_page() { |
1031 | 0 | return Status::NotSupported("struct writer has no data, can not finish_current_page"); |
1032 | 0 | } |
1033 | | |
1034 | | ArrayColumnWriter::ArrayColumnWriter(const ColumnWriterOptions& opts, TabletColumnPtr column, |
1035 | | OffsetColumnWriter* offset_writer, |
1036 | | ScalarColumnWriter* null_writer, |
1037 | | std::unique_ptr<ColumnWriter> item_writer) |
1038 | 42.8k | : ColumnWriter(std::move(column), opts.meta->is_nullable(), opts.meta), |
1039 | 42.8k | _item_writer(std::move(item_writer)), |
1040 | 42.8k | _opts(opts) { |
1041 | 42.8k | _offset_writer.reset(offset_writer); |
1042 | 42.8k | if (is_nullable()) { |
1043 | 26.9k | _null_writer.reset(null_writer); |
1044 | 26.9k | } |
1045 | 42.8k | } |
1046 | | |
1047 | 42.7k | Status ArrayColumnWriter::init() { |
1048 | 42.7k | RETURN_IF_ERROR(_offset_writer->init()); |
1049 | 42.7k | if (is_nullable()) { |
1050 | 27.0k | RETURN_IF_ERROR(_null_writer->init()); |
1051 | 27.0k | } |
1052 | 42.7k | RETURN_IF_ERROR(_item_writer->init()); |
1053 | 42.7k | if (_opts.need_inverted_index) { |
1054 | 2.03k | auto* writer = dynamic_cast<ScalarColumnWriter*>(_item_writer.get()); |
1055 | 2.03k | if (writer != nullptr) { |
1056 | 2.03k | RETURN_IF_ERROR(IndexColumnWriter::create(get_column(), &_inverted_index_writer, |
1057 | 2.03k | _opts.index_file_writer, |
1058 | 2.03k | _opts.inverted_indexes[0])); |
1059 | 2.03k | } |
1060 | 2.03k | } |
1061 | 42.7k | if (_opts.need_ann_index) { |
1062 | 60 | auto* writer = dynamic_cast<ScalarColumnWriter*>(_item_writer.get()); |
1063 | 60 | if (writer != nullptr) { |
1064 | 60 | _ann_index_writer = std::make_unique<AnnIndexColumnWriter>(_opts.index_file_writer, |
1065 | 60 | _opts.ann_index); |
1066 | 60 | RETURN_IF_ERROR(_ann_index_writer->init()); |
1067 | 60 | } |
1068 | 60 | } |
1069 | 42.7k | return Status::OK(); |
1070 | 42.7k | } |
1071 | | |
1072 | 39.7k | Status ArrayColumnWriter::write_inverted_index() { |
1073 | 39.7k | if (_opts.need_inverted_index) { |
1074 | 2.03k | return _inverted_index_writer->finish(); |
1075 | 2.03k | } |
1076 | 37.7k | return Status::OK(); |
1077 | 39.7k | } |
1078 | | |
1079 | 39.7k | Status ArrayColumnWriter::write_ann_index() { |
1080 | 39.7k | if (_opts.need_ann_index) { |
1081 | 57 | return _ann_index_writer->finish(); |
1082 | 57 | } |
1083 | 39.7k | return Status::OK(); |
1084 | 39.7k | } |
1085 | | |
1086 | | // batch append data for array |
1087 | 42.6k | Status ArrayColumnWriter::append_data(const uint8_t** ptr, size_t num_rows) { |
1088 | | // data_ptr contains |
1089 | | // [size, offset_ptr, item_data_ptr, item_nullmap_ptr] |
1090 | 42.6k | auto data_ptr = reinterpret_cast<const uint64_t*>(*ptr); |
1091 | | // total number length |
1092 | 42.6k | size_t element_cnt = size_t((unsigned long)(*data_ptr)); |
1093 | 42.6k | auto offset_data = *(data_ptr + 1); |
1094 | 42.6k | const uint8_t* offsets_ptr = (const uint8_t*)offset_data; |
1095 | 42.6k | auto data = *(data_ptr + 2); |
1096 | 42.6k | auto nested_null_map = *(data_ptr + 3); |
1097 | 42.6k | if (element_cnt > 0) { |
1098 | 26.6k | RETURN_IF_ERROR(_item_writer->append(reinterpret_cast<const uint8_t*>(nested_null_map), |
1099 | 26.6k | reinterpret_cast<const void*>(data), element_cnt)); |
1100 | 26.6k | } |
1101 | 42.6k | if (_opts.need_inverted_index) { |
1102 | 2.12k | auto* writer = dynamic_cast<ScalarColumnWriter*>(_item_writer.get()); |
1103 | | // now only support nested type is scala |
1104 | 2.12k | if (writer != nullptr) { |
1105 | | //NOTE: use array field name as index field, but item_writer size should be used when moving item_data_ptr |
1106 | 2.12k | RETURN_IF_ERROR(_inverted_index_writer->add_array_values( |
1107 | 2.12k | field_type_size(_item_writer->get_column()->type()), |
1108 | 2.12k | reinterpret_cast<const void*>(data), |
1109 | 2.12k | reinterpret_cast<const uint8_t*>(nested_null_map), offsets_ptr, num_rows)); |
1110 | 2.12k | } |
1111 | 2.12k | } |
1112 | | |
1113 | 42.6k | if (_opts.need_ann_index) { |
1114 | 60 | auto* writer = dynamic_cast<ScalarColumnWriter*>(_item_writer.get()); |
1115 | | // now only support nested type is scala |
1116 | 60 | if (writer != nullptr) { |
1117 | | //NOTE: use array field name as index field, but item_writer size should be used when moving item_data_ptr |
1118 | 60 | RETURN_IF_ERROR(_ann_index_writer->add_array_values( |
1119 | 60 | field_type_size(_item_writer->get_column()->type()), |
1120 | 60 | reinterpret_cast<const void*>(data), |
1121 | 60 | reinterpret_cast<const uint8_t*>(nested_null_map), offsets_ptr, num_rows)); |
1122 | 60 | } else { |
1123 | 0 | return Status::NotSupported( |
1124 | 0 | "Ann index can only be build on array with scalar type. but got {} as " |
1125 | 0 | "nested", |
1126 | 0 | _item_writer->get_column()->type()); |
1127 | 0 | } |
1128 | 60 | } |
1129 | | |
1130 | 42.5k | RETURN_IF_ERROR(_offset_writer->append_data(&offsets_ptr, num_rows)); |
1131 | 42.5k | return Status::OK(); |
1132 | 42.5k | } |
1133 | | |
1134 | 277 | uint64_t ArrayColumnWriter::estimate_buffer_size() { |
1135 | 277 | return _offset_writer->estimate_buffer_size() + |
1136 | 277 | (is_nullable() ? _null_writer->estimate_buffer_size() : 0) + |
1137 | 277 | _item_writer->estimate_buffer_size(); |
1138 | 277 | } |
1139 | | |
1140 | | Status ArrayColumnWriter::append_nullable(const uint8_t* null_map, const uint8_t** ptr, |
1141 | 26.7k | size_t num_rows) { |
1142 | 26.7k | RETURN_IF_ERROR(append_data(ptr, num_rows)); |
1143 | 26.7k | if (is_nullable()) { |
1144 | 26.7k | if (_opts.need_inverted_index) { |
1145 | 1.59k | RETURN_IF_ERROR(_inverted_index_writer->add_array_nulls(null_map, num_rows)); |
1146 | 1.59k | } |
1147 | 26.7k | RETURN_IF_ERROR(_null_writer->append_data(&null_map, num_rows)); |
1148 | 26.7k | } |
1149 | 26.7k | return Status::OK(); |
1150 | 26.7k | } |
1151 | | |
1152 | 42.9k | Status ArrayColumnWriter::finish() { |
1153 | 42.9k | RETURN_IF_ERROR(_offset_writer->finish()); |
1154 | 42.9k | if (is_nullable()) { |
1155 | 27.0k | RETURN_IF_ERROR(_null_writer->finish()); |
1156 | 27.0k | } |
1157 | 42.9k | RETURN_IF_ERROR(_item_writer->finish()); |
1158 | 42.9k | _opts.meta->set_num_rows(get_next_rowid()); |
1159 | 42.9k | return Status::OK(); |
1160 | 42.9k | } |
1161 | | |
1162 | 42.9k | Status ArrayColumnWriter::write_data() { |
1163 | 42.9k | RETURN_IF_ERROR(_offset_writer->write_data()); |
1164 | 42.9k | if (is_nullable()) { |
1165 | 27.0k | RETURN_IF_ERROR(_null_writer->write_data()); |
1166 | 27.0k | } |
1167 | 42.9k | RETURN_IF_ERROR(_item_writer->write_data()); |
1168 | 42.9k | return Status::OK(); |
1169 | 42.9k | } |
1170 | | |
1171 | 42.4k | Status ArrayColumnWriter::write_ordinal_index() { |
1172 | 42.4k | RETURN_IF_ERROR(_offset_writer->write_ordinal_index()); |
1173 | 42.4k | if (is_nullable()) { |
1174 | 26.5k | RETURN_IF_ERROR(_null_writer->write_ordinal_index()); |
1175 | 26.5k | } |
1176 | 42.4k | if (!has_empty_items()) { |
1177 | 26.5k | RETURN_IF_ERROR(_item_writer->write_ordinal_index()); |
1178 | 26.5k | } |
1179 | 42.4k | return Status::OK(); |
1180 | 42.4k | } |
1181 | | |
1182 | 0 | Status ArrayColumnWriter::append_nulls(size_t num_rows) { |
1183 | 0 | const UInt64 offset = cast_set<UInt64>(_item_writer->get_next_rowid()); |
1184 | 0 | std::vector<UInt64> offsets_data(num_rows + 1, offset); |
1185 | 0 | const uint8_t* offsets_ptr = reinterpret_cast<const uint8_t*>(offsets_data.data()); |
1186 | 0 | RETURN_IF_ERROR(_offset_writer->append_data(&offsets_ptr, num_rows)); |
1187 | 0 | return write_null_column(num_rows, true); |
1188 | 0 | } |
1189 | | |
1190 | 0 | Status ArrayColumnWriter::write_null_column(size_t num_rows, bool is_null) { |
1191 | 0 | uint8_t null_sign = is_null ? 1 : 0; |
1192 | 0 | while (is_nullable() && num_rows > 0) { |
1193 | | // TODO llj bulk write |
1194 | 0 | const uint8_t* null_sign_ptr = &null_sign; |
1195 | 0 | RETURN_IF_ERROR(_null_writer->append_data(&null_sign_ptr, 1)); |
1196 | 0 | --num_rows; |
1197 | 0 | } |
1198 | 0 | return Status::OK(); |
1199 | 0 | } |
1200 | | |
1201 | 0 | Status ArrayColumnWriter::finish_current_page() { |
1202 | 0 | return Status::NotSupported("array writer has no data, can not finish_current_page"); |
1203 | 0 | } |
1204 | | |
1205 | | /// ============================= MapColumnWriter =====================//// |
1206 | | MapColumnWriter::MapColumnWriter(const ColumnWriterOptions& opts, TabletColumnPtr column, |
1207 | | ScalarColumnWriter* null_writer, OffsetColumnWriter* offset_writer, |
1208 | | std::vector<std::unique_ptr<ColumnWriter>>& kv_writers) |
1209 | 21.1k | : ColumnWriter(std::move(column), opts.meta->is_nullable(), opts.meta), _opts(opts) { |
1210 | 21.1k | CHECK_EQ(kv_writers.size(), 2); |
1211 | 21.1k | _offsets_writer.reset(offset_writer); |
1212 | 21.1k | if (is_nullable()) { |
1213 | 7.76k | _null_writer.reset(null_writer); |
1214 | 7.76k | } |
1215 | 42.3k | for (auto& sub_writers : kv_writers) { |
1216 | 42.3k | _kv_writers.push_back(std::move(sub_writers)); |
1217 | 42.3k | } |
1218 | 21.1k | } |
1219 | | |
1220 | 21.1k | Status MapColumnWriter::init() { |
1221 | 21.1k | RETURN_IF_ERROR(_offsets_writer->init()); |
1222 | 21.1k | if (is_nullable()) { |
1223 | 7.77k | RETURN_IF_ERROR(_null_writer->init()); |
1224 | 7.77k | } |
1225 | | // here register_flush_page_callback to call this.put_extra_info_in_page() |
1226 | | // when finish cur data page |
1227 | 42.4k | for (auto& sub_writer : _kv_writers) { |
1228 | 42.4k | RETURN_IF_ERROR(sub_writer->init()); |
1229 | 42.4k | } |
1230 | 21.1k | return Status::OK(); |
1231 | 21.1k | } |
1232 | | |
1233 | 431 | uint64_t MapColumnWriter::estimate_buffer_size() { |
1234 | 431 | size_t estimate = 0; |
1235 | 862 | for (auto& sub_writer : _kv_writers) { |
1236 | 862 | estimate += sub_writer->estimate_buffer_size(); |
1237 | 862 | } |
1238 | 431 | estimate += _offsets_writer->estimate_buffer_size(); |
1239 | 431 | if (is_nullable()) { |
1240 | 410 | estimate += _null_writer->estimate_buffer_size(); |
1241 | 410 | } |
1242 | 431 | return estimate; |
1243 | 431 | } |
1244 | | |
1245 | 21.1k | Status MapColumnWriter::finish() { |
1246 | 21.1k | RETURN_IF_ERROR(_offsets_writer->finish()); |
1247 | 21.1k | if (is_nullable()) { |
1248 | 7.77k | RETURN_IF_ERROR(_null_writer->finish()); |
1249 | 7.77k | } |
1250 | 42.3k | for (auto& sub_writer : _kv_writers) { |
1251 | 42.3k | RETURN_IF_ERROR(sub_writer->finish()); |
1252 | 42.3k | } |
1253 | 21.1k | _opts.meta->set_num_rows(get_next_rowid()); |
1254 | 21.1k | return Status::OK(); |
1255 | 21.1k | } |
1256 | | |
1257 | | Status MapColumnWriter::append_nullable(const uint8_t* null_map, const uint8_t** ptr, |
1258 | 7.59k | size_t num_rows) { |
1259 | 7.59k | RETURN_IF_ERROR(append_data(ptr, num_rows)); |
1260 | 7.59k | if (is_nullable()) { |
1261 | 7.59k | RETURN_IF_ERROR(_null_writer->append_data(&null_map, num_rows)); |
1262 | 7.59k | } |
1263 | 7.59k | return Status::OK(); |
1264 | 7.59k | } |
1265 | | |
1266 | | // write key value data with offsets |
1267 | 21.3k | Status MapColumnWriter::append_data(const uint8_t** ptr, size_t num_rows) { |
1268 | | // data_ptr contains |
1269 | | // [size, offset_ptr, key_data_ptr, val_data_ptr, k_nullmap_ptr, v_nullmap_pr] |
1270 | | // which converted results from olap_map_convertor and later will use a structure to replace it |
1271 | 21.3k | auto data_ptr = reinterpret_cast<const uint64_t*>(*ptr); |
1272 | | // total number length |
1273 | 21.3k | size_t element_cnt = size_t((unsigned long)(*data_ptr)); |
1274 | 21.3k | auto offset_data = *(data_ptr + 1); |
1275 | 21.3k | const uint8_t* offsets_ptr = (const uint8_t*)offset_data; |
1276 | | |
1277 | 21.3k | if (element_cnt > 0) { |
1278 | 37.3k | for (size_t i = 0; i < 2; ++i) { |
1279 | 24.8k | auto data = *(data_ptr + 2 + i); |
1280 | 24.8k | auto nested_null_map = *(data_ptr + 2 + 2 + i); |
1281 | 24.8k | RETURN_IF_ERROR( |
1282 | 24.8k | _kv_writers[i]->append(reinterpret_cast<const uint8_t*>(nested_null_map), |
1283 | 24.8k | reinterpret_cast<const void*>(data), element_cnt)); |
1284 | 24.8k | } |
1285 | 12.4k | } |
1286 | | // make sure the order : offset writer flush next_array_item_ordinal after kv_writers append_data |
1287 | | // because we use _kv_writers[0]->get_next_rowid() to set next_array_item_ordinal in offset page footer |
1288 | 21.3k | RETURN_IF_ERROR(_offsets_writer->append_data(&offsets_ptr, num_rows)); |
1289 | 21.3k | return Status::OK(); |
1290 | 21.3k | } |
1291 | | |
1292 | 21.1k | Status MapColumnWriter::write_data() { |
1293 | 21.1k | RETURN_IF_ERROR(_offsets_writer->write_data()); |
1294 | 21.1k | if (is_nullable()) { |
1295 | 7.77k | RETURN_IF_ERROR(_null_writer->write_data()); |
1296 | 7.77k | } |
1297 | 42.3k | for (auto& sub_writer : _kv_writers) { |
1298 | 42.3k | RETURN_IF_ERROR(sub_writer->write_data()); |
1299 | 42.3k | } |
1300 | 21.1k | return Status::OK(); |
1301 | 21.1k | } |
1302 | | |
1303 | 20.9k | Status MapColumnWriter::write_ordinal_index() { |
1304 | 20.9k | RETURN_IF_ERROR(_offsets_writer->write_ordinal_index()); |
1305 | 20.9k | if (is_nullable()) { |
1306 | 7.50k | RETURN_IF_ERROR(_null_writer->write_ordinal_index()); |
1307 | 7.50k | } |
1308 | 41.8k | for (auto& sub_writer : _kv_writers) { |
1309 | 41.8k | if (sub_writer->get_next_rowid() != 0) { |
1310 | 24.1k | RETURN_IF_ERROR(sub_writer->write_ordinal_index()); |
1311 | 24.1k | } |
1312 | 41.8k | } |
1313 | 20.9k | return Status::OK(); |
1314 | 20.9k | } |
1315 | | |
1316 | 0 | Status MapColumnWriter::append_nulls(size_t num_rows) { |
1317 | 0 | for (auto& sub_writer : _kv_writers) { |
1318 | 0 | RETURN_IF_ERROR(sub_writer->append_nulls(num_rows)); |
1319 | 0 | } |
1320 | 0 | const UInt64 offset = cast_set<UInt64>(_kv_writers[0]->get_next_rowid()); |
1321 | 0 | std::vector<UInt64> offsets_data(num_rows + 1, offset); |
1322 | 0 | const uint8_t* offsets_ptr = reinterpret_cast<const uint8_t*>(offsets_data.data()); |
1323 | 0 | RETURN_IF_ERROR(_offsets_writer->append_data(&offsets_ptr, num_rows)); |
1324 | | |
1325 | 0 | if (is_nullable()) { |
1326 | 0 | std::vector<UInt8> null_signs(num_rows, 1); |
1327 | 0 | const uint8_t* null_sign_ptr = null_signs.data(); |
1328 | 0 | RETURN_IF_ERROR(_null_writer->append_data(&null_sign_ptr, num_rows)); |
1329 | 0 | } |
1330 | 0 | return Status::OK(); |
1331 | 0 | } |
1332 | | |
1333 | 0 | Status MapColumnWriter::finish_current_page() { |
1334 | 0 | return Status::NotSupported("map writer has no data, can not finish_current_page"); |
1335 | 0 | } |
1336 | | |
1337 | 9.03k | Status MapColumnWriter::write_inverted_index() { |
1338 | 9.03k | if (_opts.need_inverted_index) { |
1339 | 0 | return _index_builder->finish(); |
1340 | 0 | } |
1341 | 9.03k | return Status::OK(); |
1342 | 9.03k | } |
1343 | | |
1344 | | VariantColumnWriter::VariantColumnWriter(const ColumnWriterOptions& opts, TabletColumnPtr column) |
1345 | 5.90k | : ColumnWriter(std::move(column), opts.meta->is_nullable(), opts.meta) { |
1346 | 5.90k | _impl = std::make_unique<VariantColumnWriterImpl>(opts, get_column()); |
1347 | 5.90k | } |
1348 | | |
1349 | 5.89k | Status VariantColumnWriter::init() { |
1350 | 5.89k | return _impl->init(); |
1351 | 5.89k | } |
1352 | | |
1353 | 674 | Status VariantColumnWriter::append_data(const uint8_t** ptr, size_t num_rows) { |
1354 | 674 | _next_rowid += num_rows; |
1355 | 674 | return _impl->append_data(ptr, num_rows); |
1356 | 674 | } |
1357 | | |
1358 | 869 | uint64_t VariantColumnWriter::estimate_buffer_size() { |
1359 | 869 | return _impl->estimate_buffer_size(); |
1360 | 869 | } |
1361 | | |
1362 | 5.93k | Status VariantColumnWriter::finish() { |
1363 | 5.93k | return _impl->finish(); |
1364 | 5.93k | } |
1365 | 5.93k | Status VariantColumnWriter::write_data() { |
1366 | 5.93k | return _impl->write_data(); |
1367 | 5.93k | } |
1368 | 5.93k | Status VariantColumnWriter::write_ordinal_index() { |
1369 | 5.93k | return _impl->write_ordinal_index(); |
1370 | 5.93k | } |
1371 | | |
1372 | 5.92k | Status VariantColumnWriter::write_zone_map() { |
1373 | 5.92k | return _impl->write_zone_map(); |
1374 | 5.92k | } |
1375 | | |
1376 | 5.91k | Status VariantColumnWriter::write_inverted_index() { |
1377 | 5.91k | return _impl->write_inverted_index(); |
1378 | 5.91k | } |
1379 | 5.90k | Status VariantColumnWriter::write_bloom_filter_index() { |
1380 | 5.90k | return _impl->write_bloom_filter_index(); |
1381 | 5.90k | } |
1382 | | |
1383 | | Status VariantColumnWriter::append_nullable(const uint8_t* null_map, const uint8_t** ptr, |
1384 | 5.47k | size_t num_rows) { |
1385 | 5.47k | return _impl->append_nullable(null_map, ptr, num_rows); |
1386 | 5.47k | } |
1387 | | |
1388 | | } // namespace doris::segment_v2 |