Coverage Report

Created: 2026-05-25 14:29

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