Coverage Report

Created: 2026-05-09 09:25

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