Coverage Report

Created: 2026-06-09 13:37

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/segment/column_reader.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_reader.h"
19
20
#include <assert.h>
21
#include <gen_cpp/Descriptors_types.h>
22
#include <gen_cpp/segment_v2.pb.h>
23
#include <glog/logging.h>
24
25
#include <algorithm>
26
#include <memory>
27
#include <ostream>
28
#include <set>
29
#include <utility>
30
31
#include "common/compiler_util.h" // IWYU pragma: keep
32
#include "common/status.h"
33
#include "core/assert_cast.h"
34
#include "core/binary_cast.hpp"
35
#include "core/column/column.h"
36
#include "core/column/column_array.h"
37
#include "core/column/column_map.h"
38
#include "core/column/column_nullable.h"
39
#include "core/column/column_struct.h"
40
#include "core/column/column_vector.h"
41
#include "core/data_type/data_type_agg_state.h"
42
#include "core/data_type/data_type_factory.hpp"
43
#include "core/data_type/data_type_nullable.h"
44
#include "core/data_type/define_primitive_type.h"
45
#include "core/decimal12.h"
46
#include "core/string_ref.h"
47
#include "core/types.h"
48
#include "core/value/decimalv2_value.h"
49
#include "core/value/vdatetime_value.h" //for VecDateTime
50
#include "io/fs/file_reader.h"
51
#include "storage/index/ann/ann_index_reader.h"
52
#include "storage/index/bloom_filter/bloom_filter.h"
53
#include "storage/index/bloom_filter/bloom_filter_index_reader.h"
54
#include "storage/index/index_file_reader.h"
55
#include "storage/index/index_reader.h"
56
#include "storage/index/inverted/analyzer/analyzer.h"
57
#include "storage/index/inverted/inverted_index_reader.h"
58
#include "storage/index/zone_map/zone_map_index.h"
59
#include "storage/iterators.h"
60
#include "storage/olap_common.h"
61
#include "storage/predicate/block_column_predicate.h"
62
#include "storage/predicate/column_predicate.h"
63
#include "storage/segment/binary_dict_page.h" // for BinaryDictPageDecoder
64
#include "storage/segment/binary_plain_page.h"
65
#include "storage/segment/column_meta_accessor.h"
66
#include "storage/segment/encoding_info.h" // for EncodingInfo
67
#include "storage/segment/page_decoder.h"
68
#include "storage/segment/page_handle.h" // for PageHandle
69
#include "storage/segment/page_io.h"
70
#include "storage/segment/page_pointer.h" // for PagePointer
71
#include "storage/segment/row_ranges.h"
72
#include "storage/segment/segment.h"
73
#include "storage/segment/segment_prefetcher.h"
74
#include "storage/segment/variant/variant_column_reader.h"
75
#include "storage/tablet/tablet_schema.h"
76
#include "storage/types.h" // for TypeInfo
77
#include "util/bitmap.h"
78
#include "util/block_compression.h"
79
#include "util/concurrency_stats.h"
80
#include "util/defer_op.h"
81
#include "util/rle_encoding.h" // for RleDecoder
82
#include "util/slice.h"
83
84
namespace doris::segment_v2 {
85
#include "storage/segment/column_reader.h"
86
87
887
inline bool read_as_string(PrimitiveType type) {
88
887
    return type == PrimitiveType::TYPE_STRING || type == PrimitiveType::INVALID_TYPE ||
89
887
           type == PrimitiveType::TYPE_BITMAP || type == PrimitiveType::TYPE_FIXED_LENGTH_OBJECT;
90
887
}
91
92
Status ColumnReader::create_array(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,
93
                                  const io::FileReaderSPtr& file_reader,
94
63.4k
                                  std::shared_ptr<ColumnReader>* reader) {
95
63.4k
    DCHECK(meta.children_columns_size() == 2 || meta.children_columns_size() == 3);
96
97
63.4k
    std::shared_ptr<ColumnReader> item_reader;
98
63.4k
    RETURN_IF_ERROR(ColumnReader::create(opts, meta.children_columns(0),
99
63.4k
                                         meta.children_columns(0).num_rows(), file_reader,
100
63.4k
                                         &item_reader));
101
102
63.4k
    std::shared_ptr<ColumnReader> offset_reader;
103
63.4k
    RETURN_IF_ERROR(ColumnReader::create(opts, meta.children_columns(1),
104
63.4k
                                         meta.children_columns(1).num_rows(), file_reader,
105
63.4k
                                         &offset_reader));
106
107
63.4k
    std::shared_ptr<ColumnReader> null_reader;
108
63.4k
    if (meta.is_nullable()) {
109
45.1k
        RETURN_IF_ERROR(ColumnReader::create(opts, meta.children_columns(2),
110
45.1k
                                             meta.children_columns(2).num_rows(), file_reader,
111
45.1k
                                             &null_reader));
112
45.1k
    }
113
114
    // The num rows of the array reader equals to the num rows of the length reader.
115
63.4k
    uint64_t array_num_rows = meta.children_columns(1).num_rows();
116
63.4k
    std::shared_ptr<ColumnReader> array_reader(
117
63.4k
            new ColumnReader(opts, meta, array_num_rows, file_reader));
118
    //  array reader do not need to init
119
63.4k
    array_reader->_sub_readers.resize(meta.children_columns_size());
120
63.4k
    array_reader->_sub_readers[0] = std::move(item_reader);
121
63.4k
    array_reader->_sub_readers[1] = std::move(offset_reader);
122
63.4k
    if (meta.is_nullable()) {
123
45.0k
        array_reader->_sub_readers[2] = std::move(null_reader);
124
45.0k
    }
125
63.4k
    array_reader->_meta_type = FieldType::OLAP_FIELD_TYPE_ARRAY;
126
63.4k
    *reader = std::move(array_reader);
127
63.4k
    return Status::OK();
128
63.4k
}
129
130
Status ColumnReader::create_map(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,
131
                                const io::FileReaderSPtr& file_reader,
132
73.9k
                                std::shared_ptr<ColumnReader>* reader) {
133
    // map reader now has 3 sub readers for key, value, offsets(scalar), null(scala)
134
73.9k
    DCHECK(meta.children_columns_size() == 3 || meta.children_columns_size() == 4);
135
73.9k
    std::shared_ptr<ColumnReader> key_reader;
136
73.9k
    RETURN_IF_ERROR(ColumnReader::create(opts, meta.children_columns(0),
137
73.9k
                                         meta.children_columns(0).num_rows(), file_reader,
138
73.9k
                                         &key_reader));
139
73.9k
    std::shared_ptr<ColumnReader> val_reader;
140
73.9k
    RETURN_IF_ERROR(ColumnReader::create(opts, meta.children_columns(1),
141
73.9k
                                         meta.children_columns(1).num_rows(), file_reader,
142
73.9k
                                         &val_reader));
143
73.9k
    std::shared_ptr<ColumnReader> offset_reader;
144
73.9k
    RETURN_IF_ERROR(ColumnReader::create(opts, meta.children_columns(2),
145
73.9k
                                         meta.children_columns(2).num_rows(), file_reader,
146
73.9k
                                         &offset_reader));
147
73.9k
    std::shared_ptr<ColumnReader> null_reader;
148
73.9k
    if (meta.is_nullable()) {
149
15.7k
        RETURN_IF_ERROR(ColumnReader::create(opts, meta.children_columns(3),
150
15.7k
                                             meta.children_columns(3).num_rows(), file_reader,
151
15.7k
                                             &null_reader));
152
15.7k
    }
153
154
    // The num rows of the map reader equals to the num rows of the length reader.
155
73.9k
    uint64_t map_num_rows = meta.children_columns(2).num_rows();
156
73.9k
    std::shared_ptr<ColumnReader> map_reader(
157
73.9k
            new ColumnReader(opts, meta, map_num_rows, file_reader));
158
73.9k
    map_reader->_sub_readers.resize(meta.children_columns_size());
159
160
73.9k
    map_reader->_sub_readers[0] = std::move(key_reader);
161
73.9k
    map_reader->_sub_readers[1] = std::move(val_reader);
162
73.9k
    map_reader->_sub_readers[2] = std::move(offset_reader);
163
73.9k
    if (meta.is_nullable()) {
164
15.7k
        map_reader->_sub_readers[3] = std::move(null_reader);
165
15.7k
    }
166
73.9k
    map_reader->_meta_type = FieldType::OLAP_FIELD_TYPE_MAP;
167
73.9k
    *reader = std::move(map_reader);
168
73.9k
    return Status::OK();
169
73.9k
}
170
171
Status ColumnReader::create_struct(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,
172
                                   uint64_t num_rows, const io::FileReaderSPtr& file_reader,
173
7.99k
                                   std::shared_ptr<ColumnReader>* reader) {
174
    // not support empty struct
175
7.99k
    DCHECK(meta.children_columns_size() >= 1);
176
    // create struct column reader
177
7.99k
    std::shared_ptr<ColumnReader> struct_reader(
178
7.99k
            new ColumnReader(opts, meta, num_rows, file_reader));
179
7.99k
    struct_reader->_sub_readers.reserve(meta.children_columns_size());
180
    // now we support struct column can add the children columns according to the schema-change behavior
181
42.8k
    for (int i = 0; i < meta.children_columns_size(); i++) {
182
34.8k
        std::shared_ptr<ColumnReader> sub_reader;
183
34.8k
        RETURN_IF_ERROR(ColumnReader::create(opts, meta.children_columns(i),
184
34.8k
                                             meta.children_columns(i).num_rows(), file_reader,
185
34.8k
                                             &sub_reader));
186
34.8k
        struct_reader->_sub_readers.push_back(std::move(sub_reader));
187
34.8k
    }
188
7.99k
    struct_reader->_meta_type = FieldType::OLAP_FIELD_TYPE_STRUCT;
189
7.99k
    *reader = std::move(struct_reader);
190
7.99k
    return Status::OK();
191
7.99k
}
192
193
Status ColumnReader::create_agg_state(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,
194
                                      uint64_t num_rows, const io::FileReaderSPtr& file_reader,
195
892
                                      std::shared_ptr<ColumnReader>* reader) {
196
892
    if (!meta.has_function_name()) { // meet old version ColumnMetaPB
197
0
        std::shared_ptr<ColumnReader> reader_local(
198
0
                new ColumnReader(opts, meta, num_rows, file_reader));
199
0
        RETURN_IF_ERROR(reader_local->init(&meta));
200
0
        *reader = std::move(reader_local);
201
0
        return Status::OK();
202
0
    }
203
204
892
    auto data_type = DataTypeFactory::instance().create_data_type(meta);
205
892
    const auto* agg_state_type = assert_cast<const DataTypeAggState*>(data_type.get());
206
892
    agg_state_type->check_function_compatibility(opts.be_exec_version);
207
892
    auto type = agg_state_type->get_serialized_type()->get_primitive_type();
208
209
892
    if (read_as_string(type)) {
210
836
        std::shared_ptr<ColumnReader> reader_local(
211
836
                new ColumnReader(opts, meta, num_rows, file_reader));
212
836
        RETURN_IF_ERROR(reader_local->init(&meta));
213
836
        *reader = std::move(reader_local);
214
836
        return Status::OK();
215
836
    } else if (type == PrimitiveType::TYPE_MAP) {
216
38
        return create_map(opts, meta, file_reader, reader);
217
38
    } else if (type == PrimitiveType::TYPE_ARRAY) {
218
13
        return create_array(opts, meta, file_reader, reader);
219
13
    } else if (type == PrimitiveType::TYPE_STRUCT) {
220
0
        return create_struct(opts, meta, num_rows, file_reader, reader);
221
0
    }
222
223
5
    return Status::InternalError("Not supported type: {}, serialized type: {}",
224
5
                                 agg_state_type->get_name(), int(type));
225
892
}
226
227
102k
bool ColumnReader::is_compaction_reader_type(ReaderType type) {
228
102k
    return type == ReaderType::READER_BASE_COMPACTION ||
229
102k
           type == ReaderType::READER_CUMULATIVE_COMPACTION ||
230
102k
           type == ReaderType::READER_COLD_DATA_COMPACTION ||
231
102k
           type == ReaderType::READER_SEGMENT_COMPACTION ||
232
102k
           type == ReaderType::READER_FULL_COMPACTION;
233
102k
}
234
235
Status ColumnReader::create(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,
236
                            uint64_t num_rows, const io::FileReaderSPtr& file_reader,
237
29.0M
                            std::shared_ptr<ColumnReader>* reader) {
238
29.0M
    if (is_scalar_type((FieldType)meta.type())) {
239
28.9M
        std::shared_ptr<ColumnReader> reader_local(
240
28.9M
                new ColumnReader(opts, meta, num_rows, file_reader));
241
28.9M
        RETURN_IF_ERROR(reader_local->init(&meta));
242
28.9M
        *reader = std::move(reader_local);
243
28.9M
        return Status::OK();
244
28.9M
    } else {
245
158k
        auto type = (FieldType)meta.type();
246
158k
        switch (type) {
247
891
        case FieldType::OLAP_FIELD_TYPE_AGG_STATE: {
248
891
            return create_agg_state(opts, meta, num_rows, file_reader, reader);
249
0
        }
250
7.99k
        case FieldType::OLAP_FIELD_TYPE_STRUCT: {
251
7.99k
            return create_struct(opts, meta, num_rows, file_reader, reader);
252
0
        }
253
63.4k
        case FieldType::OLAP_FIELD_TYPE_ARRAY: {
254
63.4k
            return create_array(opts, meta, file_reader, reader);
255
0
        }
256
73.9k
        case FieldType::OLAP_FIELD_TYPE_MAP: {
257
73.9k
            return create_map(opts, meta, file_reader, reader);
258
0
        }
259
18.2k
        case FieldType::OLAP_FIELD_TYPE_VARIANT: {
260
            // Read variant only root data using a single ColumnReader
261
18.2k
            std::shared_ptr<ColumnReader> reader_local(
262
18.2k
                    new ColumnReader(opts, meta, num_rows, file_reader));
263
18.2k
            RETURN_IF_ERROR(reader_local->init(&meta));
264
18.2k
            *reader = std::move(reader_local);
265
18.2k
            return Status::OK();
266
18.2k
        }
267
0
        default:
268
0
            return Status::NotSupported("unsupported type for ColumnReader: {}",
269
0
                                        std::to_string(int(type)));
270
158k
        }
271
158k
    }
272
29.0M
}
273
274
18.2k
ColumnReader::ColumnReader() = default;
275
276
ColumnReader::ColumnReader(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,
277
                           uint64_t num_rows, io::FileReaderSPtr file_reader)
278
29.1M
        : _use_index_page_cache(!config::disable_storage_page_cache),
279
29.1M
          _opts(opts),
280
29.1M
          _num_rows(num_rows),
281
29.1M
          _file_reader(std::move(file_reader)),
282
29.1M
          _dict_encoding_type(UNKNOWN_DICT_ENCODING) {
283
29.1M
    _meta_length = meta.length();
284
29.1M
    _meta_type = (FieldType)meta.type();
285
29.1M
    if (_meta_type == FieldType::OLAP_FIELD_TYPE_ARRAY) {
286
63.4k
        _meta_children_column_type = (FieldType)meta.children_columns(0).type();
287
63.4k
    }
288
29.1M
    _data_type = DataTypeFactory::instance().create_data_type(meta);
289
29.1M
    _meta_is_nullable = meta.is_nullable();
290
29.1M
    _meta_dict_page = meta.dict_page();
291
29.1M
    _meta_compression = meta.compression();
292
29.1M
}
293
294
29.2M
ColumnReader::~ColumnReader() = default;
295
296
29.0M
int64_t ColumnReader::get_metadata_size() const {
297
29.0M
    return sizeof(ColumnReader) + (_segment_zone_map ? _segment_zone_map->ByteSizeLong() : 0);
298
29.0M
}
299
300
#ifdef BE_TEST
301
/// This function is only used in UT to verify the correctness of data read from zone map
302
/// See UT case 'SegCompactionMoWTest.SegCompactionInterleaveWithBig_ooooOOoOooooooooO'
303
/// be/test/olap/segcompaction_mow_test.cpp
304
void ColumnReader::check_data_by_zone_map_for_test(const MutableColumnPtr& dst) const {
305
    if (!_segment_zone_map) {
306
        return;
307
    }
308
309
    const auto rows = dst->size();
310
    if (rows == 0) {
311
        return;
312
    }
313
314
    FieldType type = _type;
315
316
    if (type != FieldType::OLAP_FIELD_TYPE_INT) {
317
        return;
318
    }
319
320
    auto* non_nullable_column =
321
            dst->is_nullable()
322
                    ? assert_cast<ColumnNullable*>(dst.get())->get_nested_column_ptr().get()
323
                    : dst.get();
324
325
    /// `PredicateColumnType<TYPE_INT>` does not support `void get(size_t n, Field& res)`,
326
    /// So here only check `CoumnVector<TYPE_INT>`
327
    if (check_and_get_column<ColumnVector<TYPE_INT>>(non_nullable_column) == nullptr) {
328
        return;
329
    }
330
331
    ZoneMap zone_map;
332
    THROW_IF_ERROR(ZoneMap::from_proto(*_segment_zone_map, _data_type, zone_map));
333
334
    if (zone_map.has_null) {
335
        return;
336
    }
337
338
    for (size_t i = 0; i != rows; ++i) {
339
        Field field;
340
        dst->get(i, field);
341
        DCHECK(!field.is_null());
342
        const auto v = field.get<TYPE_INT>();
343
        DCHECK_GE(v, zone_map.min_value.get<TYPE_INT>());
344
        DCHECK_LE(v, zone_map.max_value.get<TYPE_INT>());
345
    }
346
}
347
#endif
348
349
28.9M
Status ColumnReader::init(const ColumnMetaPB* meta) {
350
28.9M
    _type = (FieldType)meta->type();
351
352
28.9M
    if (meta->has_be_exec_version()) {
353
28.6M
        _be_exec_version = meta->be_exec_version();
354
28.6M
    }
355
356
28.9M
    if (_type == FieldType::OLAP_FIELD_TYPE_NONE || _type == FieldType::OLAP_FIELD_TYPE_UNKNOWN) {
357
0
        return Status::NotSupported("unsupported typeinfo, type={}", meta->type());
358
0
    }
359
28.9M
    RETURN_IF_ERROR(EncodingInfo::get(_type, meta->encoding(), &_encoding_info));
360
361
86.1M
    for (int i = 0; i < meta->indexes_size(); i++) {
362
57.2M
        const auto& index_meta = meta->indexes(i);
363
57.2M
        switch (index_meta.type()) {
364
0
        case BITMAP_INDEX:
365
0
            break;
366
28.8M
        case ORDINAL_INDEX:
367
28.8M
            _ordinal_index.reset(
368
28.8M
                    new OrdinalIndexReader(_file_reader, _num_rows, index_meta.ordinal_index()));
369
28.8M
            break;
370
28.3M
        case ZONE_MAP_INDEX:
371
28.3M
            _segment_zone_map =
372
28.3M
                    std::make_unique<ZoneMapPB>(index_meta.zone_map_index().segment_zone_map());
373
28.3M
            _zone_map_index.reset(new ZoneMapIndexReader(
374
28.3M
                    _file_reader, index_meta.zone_map_index().page_zone_maps()));
375
28.3M
            break;
376
5.05k
        case BLOOM_FILTER_INDEX:
377
5.05k
            _bloom_filter_index.reset(
378
5.05k
                    new BloomFilterIndexReader(_file_reader, index_meta.bloom_filter_index()));
379
5.05k
            break;
380
0
        case NESTED_OFFSETS_INDEX:
381
0
            break;
382
0
        default:
383
0
            return Status::Corruption("Bad file {}: invalid column index type {}",
384
0
                                      _file_reader->path().native(), index_meta.type());
385
57.2M
        }
386
57.2M
    }
387
28.8M
    update_metadata_size();
388
389
    // ArrayColumnWriter writes a single empty array and flushes. In this scenario,
390
    // the item writer doesn't write any data and the corresponding ordinal index is empty.
391
28.8M
    if (_ordinal_index == nullptr && !is_empty()) {
392
0
        return Status::Corruption("Bad file {}: missing ordinal index for column {}",
393
0
                                  _file_reader->path().native(), meta->column_id());
394
0
    }
395
396
28.8M
    return Status::OK();
397
28.8M
}
398
399
Status ColumnReader::new_index_iterator(const std::shared_ptr<IndexFileReader>& index_file_reader,
400
                                        const TabletIndex* index_meta, const std::string& rowset_id,
401
                                        uint32_t segment_id, size_t rows_of_segment,
402
70.5k
                                        std::unique_ptr<IndexIterator>* iterator) {
403
70.5k
    RETURN_IF_ERROR(
404
70.5k
            _load_index(index_file_reader, index_meta, rowset_id, segment_id, rows_of_segment));
405
70.5k
    {
406
70.5k
        std::shared_lock<std::shared_mutex> rlock(_load_index_lock);
407
70.5k
        auto iter = _index_readers.find(index_meta->index_id());
408
70.7k
        if (iter != _index_readers.end()) {
409
70.7k
            if (iter->second != nullptr) {
410
70.7k
                RETURN_IF_ERROR(iter->second->new_iterator(iterator));
411
70.7k
            }
412
70.7k
        }
413
70.5k
    }
414
70.5k
    return Status::OK();
415
70.5k
}
416
417
Status ColumnReader::read_page(const ColumnIteratorOptions& iter_opts, const PagePointer& pp,
418
                               PageHandle* handle, Slice* page_body, PageFooterPB* footer,
419
2.06M
                               BlockCompressionCodec* codec) const {
420
2.06M
    SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().column_reader_read_page);
421
2.06M
    iter_opts.sanity_check();
422
2.06M
    PageReadOptions opts(iter_opts.io_ctx);
423
2.06M
    opts.verify_checksum = _opts.verify_checksum;
424
2.06M
    opts.use_page_cache = iter_opts.use_page_cache;
425
2.06M
    opts.kept_in_memory = _opts.kept_in_memory;
426
2.06M
    opts.type = iter_opts.type;
427
2.06M
    opts.file_reader = iter_opts.file_reader;
428
2.06M
    opts.page_pointer = pp;
429
2.06M
    opts.codec = codec;
430
2.06M
    opts.stats = iter_opts.stats;
431
2.06M
    opts.encoding_info = _encoding_info;
432
433
2.06M
    return PageIO::read_and_decompress_page(opts, handle, page_body, footer);
434
2.06M
}
435
436
Status ColumnReader::get_row_ranges_by_zone_map(
437
        const AndBlockColumnPredicate* col_predicates,
438
        const std::vector<std::shared_ptr<const ColumnPredicate>>* delete_predicates,
439
134k
        RowRanges* row_ranges, const ColumnIteratorOptions& iter_opts) {
440
134k
    std::vector<uint32_t> page_indexes;
441
134k
    RETURN_IF_ERROR(
442
134k
            _get_filtered_pages(col_predicates, delete_predicates, &page_indexes, iter_opts));
443
134k
    RETURN_IF_ERROR(_calculate_row_ranges(page_indexes, row_ranges, iter_opts));
444
134k
    return Status::OK();
445
134k
}
446
447
18.3k
Status ColumnReader::next_batch_of_zone_map(size_t* n, MutableColumnPtr& dst) const {
448
18.3k
    if (_segment_zone_map == nullptr) {
449
0
        return Status::InternalError("segment zonemap not exist");
450
0
    }
451
    // TODO: this work to get min/max value seems should only do once
452
18.3k
    ZoneMap zone_map;
453
18.3k
    RETURN_IF_ERROR(ZoneMap::from_proto(*_segment_zone_map, _data_type, zone_map));
454
455
18.3k
    dst->reserve(*n);
456
18.3k
    if (!zone_map.has_not_null) {
457
1.14k
        assert_cast<ColumnNullable&>(*dst).insert_many_defaults(*n);
458
1.14k
        return Status::OK();
459
1.14k
    }
460
17.2k
    dst->insert(zone_map.max_value);
461
34.4k
    for (int i = 1; i < *n; ++i) {
462
17.2k
        dst->insert(zone_map.min_value);
463
17.2k
    }
464
17.2k
    return Status::OK();
465
18.3k
}
466
467
Status ColumnReader::match_condition(const AndBlockColumnPredicate* col_predicates,
468
1.84M
                                     bool* matched) const {
469
1.84M
    *matched = true;
470
1.84M
    if (_zone_map_index == nullptr) {
471
0
        return Status::OK();
472
0
    }
473
1.84M
    ZoneMap zone_map;
474
1.84M
    RETURN_IF_ERROR(ZoneMap::from_proto(*_segment_zone_map, _data_type, zone_map));
475
476
1.84M
    *matched = _zone_map_match_condition(zone_map, col_predicates);
477
1.84M
    return Status::OK();
478
1.84M
}
479
480
Status ColumnReader::prune_predicates_by_zone_map(
481
        std::vector<std::shared_ptr<ColumnPredicate>>& predicates, const int column_id,
482
1.73M
        bool* pruned) const {
483
1.73M
    *pruned = false;
484
1.73M
    if (_zone_map_index == nullptr) {
485
96
        return Status::OK();
486
96
    }
487
488
1.73M
    ZoneMap zone_map;
489
1.73M
    RETURN_IF_ERROR(ZoneMap::from_proto(*_segment_zone_map, _data_type, zone_map));
490
1.73M
    if (zone_map.pass_all) {
491
0
        return Status::OK();
492
0
    }
493
494
3.60M
    for (auto it = predicates.begin(); it != predicates.end();) {
495
1.86M
        auto predicate = *it;
496
1.86M
        if (predicate->column_id() == column_id && predicate->is_always_true(zone_map)) {
497
10.9k
            *pruned = true;
498
10.9k
            it = predicates.erase(it);
499
1.85M
        } else {
500
1.85M
            ++it;
501
1.85M
        }
502
1.86M
    }
503
1.73M
    return Status::OK();
504
1.73M
}
505
506
bool ColumnReader::_zone_map_match_condition(const ZoneMap& zone_map,
507
1.93M
                                             const AndBlockColumnPredicate* col_predicates) const {
508
1.93M
    if (zone_map.pass_all) {
509
0
        return true;
510
0
    }
511
512
1.93M
    return col_predicates->evaluate_and(zone_map);
513
1.93M
}
514
515
Status ColumnReader::_get_filtered_pages(
516
        const AndBlockColumnPredicate* col_predicates,
517
        const std::vector<std::shared_ptr<const ColumnPredicate>>* delete_predicates,
518
134k
        std::vector<uint32_t>* page_indexes, const ColumnIteratorOptions& iter_opts) {
519
134k
    RETURN_IF_ERROR(_load_zone_map_index(_use_index_page_cache, _opts.kept_in_memory, iter_opts));
520
521
134k
    const std::vector<ZoneMapPB>& zone_maps = _zone_map_index->page_zone_maps();
522
134k
    size_t page_size = _zone_map_index->num_pages();
523
317k
    for (size_t i = 0; i < page_size; ++i) {
524
182k
        if (zone_maps[i].pass_all()) {
525
86.3k
            page_indexes->push_back(cast_set<uint32_t>(i));
526
96.3k
        } else {
527
96.3k
            segment_v2::ZoneMap zone_map;
528
96.3k
            RETURN_IF_ERROR(ZoneMap::from_proto(zone_maps[i], _data_type, zone_map));
529
96.3k
            if (_zone_map_match_condition(zone_map, col_predicates)) {
530
95.7k
                bool should_read = true;
531
95.7k
                if (delete_predicates != nullptr) {
532
4
                    for (auto del_pred : *delete_predicates) {
533
                        // TODO: Both `min_value` and `max_value` should be 0 or neither should be 0.
534
                        //  So nullable only need to judge once.
535
4
                        if (del_pred->evaluate_del(zone_map)) {
536
1
                            should_read = false;
537
1
                            break;
538
1
                        }
539
4
                    }
540
3
                }
541
95.7k
                if (should_read) {
542
95.7k
                    page_indexes->push_back(cast_set<uint32_t>(i));
543
95.7k
                }
544
95.7k
            }
545
96.3k
        }
546
182k
    }
547
18.4E
    VLOG(1) << "total-pages: " << page_size << " not-filtered-pages: " << page_indexes->size()
548
18.4E
            << " filtered-percent:"
549
18.4E
            << 1.0 - (static_cast<double>(page_indexes->size()) /
550
18.4E
                      (static_cast<double>(page_size) * 1.0));
551
134k
    return Status::OK();
552
134k
}
553
554
Status ColumnReader::_calculate_row_ranges(const std::vector<uint32_t>& page_indexes,
555
                                           RowRanges* row_ranges,
556
135k
                                           const ColumnIteratorOptions& iter_opts) {
557
135k
    row_ranges->clear();
558
135k
    RETURN_IF_ERROR(_load_ordinal_index(_use_index_page_cache, _opts.kept_in_memory, iter_opts));
559
181k
    for (auto i : page_indexes) {
560
181k
        ordinal_t page_first_id = _ordinal_index->get_first_ordinal(i);
561
181k
        ordinal_t page_last_id = _ordinal_index->get_last_ordinal(i);
562
181k
        RowRanges page_row_ranges(RowRanges::create_single(page_first_id, page_last_id + 1));
563
181k
        RowRanges::ranges_union(*row_ranges, page_row_ranges, row_ranges);
564
181k
    }
565
135k
    return Status::OK();
566
135k
}
567
568
Status ColumnReader::get_row_ranges_by_bloom_filter(const AndBlockColumnPredicate* col_predicates,
569
                                                    RowRanges* row_ranges,
570
107
                                                    const ColumnIteratorOptions& iter_opts) {
571
107
    RETURN_IF_ERROR(_load_ordinal_index(_use_index_page_cache, _opts.kept_in_memory, iter_opts));
572
107
    RETURN_IF_ERROR(
573
107
            _load_bloom_filter_index(_use_index_page_cache, _opts.kept_in_memory, iter_opts));
574
107
    RowRanges bf_row_ranges;
575
107
    std::unique_ptr<BloomFilterIndexIterator> bf_iter;
576
107
    RETURN_IF_ERROR(_bloom_filter_index->new_iterator(&bf_iter, iter_opts.stats));
577
107
    size_t range_size = row_ranges->range_size();
578
    // get covered page ids
579
107
    std::set<uint32_t> page_ids;
580
212
    for (int i = 0; i < range_size; ++i) {
581
105
        int64_t from = row_ranges->get_range_from(i);
582
105
        int64_t idx = from;
583
105
        int64_t to = row_ranges->get_range_to(i);
584
105
        auto iter = _ordinal_index->seek_at_or_before(from);
585
392
        while (idx < to && iter.valid()) {
586
287
            page_ids.insert(iter.page_index());
587
287
            idx = iter.last_ordinal() + 1;
588
287
            iter.next();
589
287
        }
590
105
    }
591
333
    for (auto& pid : page_ids) {
592
333
        std::unique_ptr<BloomFilter> bf;
593
333
        RETURN_IF_ERROR(bf_iter->read_bloom_filter(pid, &bf));
594
333
        if (col_predicates->evaluate_and(bf.get())) {
595
21
            bf_row_ranges.add(RowRange(_ordinal_index->get_first_ordinal(pid),
596
21
                                       _ordinal_index->get_last_ordinal(pid) + 1));
597
21
        }
598
333
    }
599
107
    RowRanges::ranges_intersection(*row_ranges, bf_row_ranges, row_ranges);
600
107
    return Status::OK();
601
107
}
602
603
Status ColumnReader::_load_ordinal_index(bool use_page_cache, bool kept_in_memory,
604
1.72M
                                         const ColumnIteratorOptions& iter_opts) {
605
1.72M
    if (!_ordinal_index) {
606
0
        return Status::InternalError("ordinal_index not inited");
607
0
    }
608
1.72M
    return _ordinal_index->load(use_page_cache, kept_in_memory, iter_opts.stats);
609
1.72M
}
610
611
Status ColumnReader::_load_zone_map_index(bool use_page_cache, bool kept_in_memory,
612
134k
                                          const ColumnIteratorOptions& iter_opts) {
613
134k
    if (_zone_map_index != nullptr) {
614
134k
        return _zone_map_index->load(use_page_cache, kept_in_memory, iter_opts.stats);
615
134k
    }
616
18.4E
    return Status::OK();
617
134k
}
618
619
Status ColumnReader::_load_index(const std::shared_ptr<IndexFileReader>& index_file_reader,
620
                                 const TabletIndex* index_meta, const std::string& rowset_id,
621
70.4k
                                 uint32_t segment_id, size_t rows_of_segment) {
622
70.4k
    std::unique_lock<std::shared_mutex> wlock(_load_index_lock);
623
624
70.4k
    if (index_meta == nullptr) {
625
0
        return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
626
0
                "Failed to load inverted index: index metadata is null");
627
0
    }
628
629
70.4k
    auto it = _index_readers.find(index_meta->index_id());
630
70.4k
    if (it != _index_readers.end()) {
631
0
        return Status::OK();
632
0
    }
633
634
70.4k
    bool should_analyzer =
635
70.4k
            inverted_index::InvertedIndexAnalyzer::should_analyzer(index_meta->properties());
636
637
70.4k
    FieldType type;
638
70.4k
    if (_meta_type == FieldType::OLAP_FIELD_TYPE_ARRAY) {
639
6.16k
        type = _meta_children_column_type;
640
64.3k
    } else {
641
64.3k
        type = _type;
642
64.3k
    }
643
644
70.4k
    if (index_meta->index_type() == IndexType::ANN) {
645
155
        _index_readers[index_meta->index_id()] = std::make_shared<AnnIndexReader>(
646
155
                index_meta, index_file_reader, rowset_id, segment_id, rows_of_segment);
647
155
        return Status::OK();
648
155
    }
649
650
70.3k
    IndexReaderPtr index_reader;
651
652
70.3k
    if (is_string_type(type)) {
653
38.9k
        if (should_analyzer) {
654
22.5k
            try {
655
22.5k
                index_reader = FullTextIndexReader::create_shared(index_meta, index_file_reader);
656
22.5k
            } catch (const CLuceneError& e) {
657
0
                return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
658
0
                        "create FullTextIndexReader error: {}", e.what());
659
0
            }
660
22.5k
        } else {
661
16.4k
            try {
662
16.4k
                index_reader =
663
16.4k
                        StringTypeInvertedIndexReader::create_shared(index_meta, index_file_reader);
664
16.4k
            } catch (const CLuceneError& e) {
665
0
                return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
666
0
                        "create StringTypeInvertedIndexReader error: {}", e.what());
667
0
            }
668
16.4k
        }
669
38.9k
    } else if (is_numeric_type(type)) {
670
31.5k
        try {
671
31.5k
            index_reader = BkdIndexReader::create_shared(index_meta, index_file_reader);
672
31.5k
        } catch (const CLuceneError& e) {
673
0
            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(
674
0
                    "create BkdIndexReader error: {}", e.what());
675
0
        }
676
18.4E
    } else {
677
18.4E
        return Status::Error<ErrorCode::INVERTED_INDEX_NOT_SUPPORTED>(
678
18.4E
                "Field type {} is not supported for inverted index", type);
679
18.4E
    }
680
70.4k
    _index_readers[index_meta->index_id()] = index_reader;
681
70.4k
    return Status::OK();
682
70.3k
}
683
684
108k
bool ColumnReader::has_bloom_filter_index(bool ngram) const {
685
108k
    if (_bloom_filter_index == nullptr) return false;
686
687
18.4E
    if (ngram) {
688
77
        return _bloom_filter_index->algorithm() == BloomFilterAlgorithmPB::NGRAM_BLOOM_FILTER;
689
18.4E
    } else {
690
18.4E
        return _bloom_filter_index->algorithm() != BloomFilterAlgorithmPB::NGRAM_BLOOM_FILTER;
691
18.4E
    }
692
18.4E
}
693
694
Status ColumnReader::_load_bloom_filter_index(bool use_page_cache, bool kept_in_memory,
695
108
                                              const ColumnIteratorOptions& iter_opts) {
696
108
    if (_bloom_filter_index != nullptr) {
697
108
        return _bloom_filter_index->load(use_page_cache, kept_in_memory, iter_opts.stats);
698
108
    }
699
0
    return Status::OK();
700
108
}
701
702
Status ColumnReader::seek_at_or_before(ordinal_t ordinal, OrdinalPageIndexIterator* iter,
703
1.59M
                                       const ColumnIteratorOptions& iter_opts) {
704
1.59M
    RETURN_IF_ERROR(_load_ordinal_index(_use_index_page_cache, _opts.kept_in_memory, iter_opts));
705
1.59M
    *iter = _ordinal_index->seek_at_or_before(ordinal);
706
1.59M
    if (!iter->valid()) {
707
0
        return Status::NotFound("Failed to seek to ordinal {}, ", ordinal);
708
0
    }
709
1.59M
    return Status::OK();
710
1.59M
}
711
712
Status ColumnReader::get_ordinal_index_reader(OrdinalIndexReader*& reader,
713
299k
                                              OlapReaderStatistics* index_load_stats) {
714
18.4E
    CHECK(_ordinal_index) << fmt::format("ordinal index is null for column reader of type {}",
715
18.4E
                                         std::to_string(int(_meta_type)));
716
299k
    RETURN_IF_ERROR(
717
299k
            _ordinal_index->load(_use_index_page_cache, _opts.kept_in_memory, index_load_stats));
718
299k
    reader = _ordinal_index.get();
719
299k
    return Status::OK();
720
299k
}
721
722
386k
Status ColumnReader::new_iterator(ColumnIteratorUPtr* iterator, const TabletColumn* tablet_column) {
723
386k
    return new_iterator(iterator, tablet_column, nullptr);
724
386k
}
725
726
Status ColumnReader::new_iterator(ColumnIteratorUPtr* iterator, const TabletColumn* tablet_column,
727
27.1M
                                  const StorageReadOptions* opt) {
728
27.1M
    if (is_empty()) {
729
46.0k
        *iterator = std::make_unique<EmptyFileColumnIterator>();
730
46.0k
        return Status::OK();
731
46.0k
    }
732
27.1M
    if (is_scalar_type(_meta_type)) {
733
27.0M
        if (is_string_type(_meta_type)) {
734
16.2M
            *iterator = std::make_unique<StringFileColumnIterator>(shared_from_this());
735
16.2M
        } else {
736
10.7M
            *iterator = std::make_unique<FileColumnIterator>(shared_from_this());
737
10.7M
        }
738
27.0M
        (*iterator)->set_column_name(tablet_column ? tablet_column->name() : "");
739
27.0M
        return Status::OK();
740
27.0M
    } else {
741
85.4k
        auto type = _meta_type;
742
85.4k
        switch (type) {
743
839
        case FieldType::OLAP_FIELD_TYPE_AGG_STATE: {
744
839
            return new_agg_state_iterator(iterator);
745
0
        }
746
7.92k
        case FieldType::OLAP_FIELD_TYPE_STRUCT: {
747
7.92k
            return new_struct_iterator(iterator, tablet_column);
748
0
        }
749
62.7k
        case FieldType::OLAP_FIELD_TYPE_ARRAY: {
750
62.7k
            return new_array_iterator(iterator, tablet_column);
751
0
        }
752
36.7k
        case FieldType::OLAP_FIELD_TYPE_MAP: {
753
36.7k
            return new_map_iterator(iterator, tablet_column);
754
0
        }
755
0
        default:
756
0
            return Status::NotSupported("unsupported type to create iterator: {}",
757
0
                                        std::to_string(int(type)));
758
85.4k
        }
759
85.4k
    }
760
27.1M
}
761
762
836
Status ColumnReader::new_agg_state_iterator(ColumnIteratorUPtr* iterator) {
763
836
    *iterator = std::make_unique<FileColumnIterator>(shared_from_this());
764
836
    return Status::OK();
765
836
}
766
767
Status ColumnReader::new_array_iterator(ColumnIteratorUPtr* iterator,
768
62.8k
                                        const TabletColumn* tablet_column) {
769
62.8k
    ColumnIteratorUPtr item_iterator;
770
62.8k
    RETURN_IF_ERROR(_sub_readers[0]->new_iterator(
771
62.8k
            &item_iterator, tablet_column && tablet_column->get_subtype_count() > 0
772
62.8k
                                    ? &tablet_column->get_sub_column(0)
773
62.8k
                                    : nullptr));
774
775
62.8k
    item_iterator->set_column_name(tablet_column ? tablet_column->get_sub_column(0).name() : "");
776
777
62.8k
    ColumnIteratorUPtr offset_iterator;
778
62.8k
    RETURN_IF_ERROR(_sub_readers[1]->new_iterator(&offset_iterator, nullptr));
779
62.8k
    auto* file_iter = static_cast<FileColumnIterator*>(offset_iterator.release());
780
62.8k
    OffsetFileColumnIteratorUPtr ofcIter = std::make_unique<OffsetFileColumnIterator>(
781
62.8k
            std::unique_ptr<FileColumnIterator>(file_iter));
782
783
62.8k
    ColumnIteratorUPtr null_iterator;
784
62.8k
    if (is_nullable()) {
785
44.5k
        RETURN_IF_ERROR(_sub_readers[2]->new_iterator(&null_iterator, nullptr));
786
44.5k
    }
787
62.8k
    *iterator = std::make_unique<ArrayFileColumnIterator>(shared_from_this(), std::move(ofcIter),
788
62.8k
                                                          std::move(item_iterator),
789
62.8k
                                                          std::move(null_iterator));
790
62.8k
    return Status::OK();
791
62.8k
}
792
793
Status ColumnReader::new_map_iterator(ColumnIteratorUPtr* iterator,
794
36.7k
                                      const TabletColumn* tablet_column) {
795
36.7k
    ColumnIteratorUPtr key_iterator;
796
36.7k
    RETURN_IF_ERROR(_sub_readers[0]->new_iterator(
797
36.7k
            &key_iterator, tablet_column && tablet_column->get_subtype_count() > 1
798
36.7k
                                   ? &tablet_column->get_sub_column(0)
799
36.7k
                                   : nullptr));
800
36.7k
    key_iterator->set_column_name(tablet_column ? tablet_column->get_sub_column(0).name() : "");
801
36.7k
    ColumnIteratorUPtr val_iterator;
802
36.7k
    RETURN_IF_ERROR(_sub_readers[1]->new_iterator(
803
36.7k
            &val_iterator, tablet_column && tablet_column->get_subtype_count() > 1
804
36.7k
                                   ? &tablet_column->get_sub_column(1)
805
36.7k
                                   : nullptr));
806
36.7k
    val_iterator->set_column_name(tablet_column ? tablet_column->get_sub_column(1).name() : "");
807
36.7k
    ColumnIteratorUPtr offsets_iterator;
808
36.7k
    RETURN_IF_ERROR(_sub_readers[2]->new_iterator(&offsets_iterator, nullptr));
809
36.7k
    auto* file_iter = static_cast<FileColumnIterator*>(offsets_iterator.release());
810
36.7k
    OffsetFileColumnIteratorUPtr ofcIter = std::make_unique<OffsetFileColumnIterator>(
811
36.7k
            std::unique_ptr<FileColumnIterator>(file_iter));
812
813
36.7k
    ColumnIteratorUPtr null_iterator;
814
36.7k
    if (is_nullable()) {
815
15.6k
        RETURN_IF_ERROR(_sub_readers[3]->new_iterator(&null_iterator, nullptr));
816
15.6k
    }
817
36.7k
    *iterator = std::make_unique<MapFileColumnIterator>(
818
36.7k
            shared_from_this(), std::move(null_iterator), std::move(ofcIter),
819
36.7k
            std::move(key_iterator), std::move(val_iterator));
820
36.7k
    return Status::OK();
821
36.7k
}
822
823
Status ColumnReader::new_struct_iterator(ColumnIteratorUPtr* iterator,
824
7.92k
                                         const TabletColumn* tablet_column) {
825
7.92k
    std::vector<ColumnIteratorUPtr> sub_column_iterators;
826
7.92k
    size_t child_size = is_nullable() ? _sub_readers.size() - 1 : _sub_readers.size();
827
18.4E
    size_t tablet_column_size = tablet_column ? tablet_column->get_sub_columns().size() : 0;
828
7.92k
    sub_column_iterators.reserve(child_size);
829
830
35.2k
    for (uint64_t i = 0; i < child_size; i++) {
831
27.2k
        ColumnIteratorUPtr sub_column_iterator;
832
27.2k
        RETURN_IF_ERROR(_sub_readers[i]->new_iterator(
833
27.2k
                &sub_column_iterator, tablet_column ? &tablet_column->get_sub_column(i) : nullptr));
834
27.2k
        sub_column_iterator->set_column_name(tablet_column ? tablet_column->get_sub_column(i).name()
835
18.4E
                                                           : "");
836
27.2k
        sub_column_iterators.emplace_back(std::move(sub_column_iterator));
837
27.2k
    }
838
839
    // create default_iterator for schema-change behavior which increase column
840
9.22k
    for (size_t i = child_size; i < tablet_column_size; i++) {
841
1.30k
        TabletColumn column = tablet_column->get_sub_column(i);
842
1.30k
        ColumnIteratorUPtr it;
843
1.30k
        RETURN_IF_ERROR(Segment::new_default_iterator(column, &it));
844
1.30k
        it->set_column_name(column.name());
845
1.30k
        sub_column_iterators.emplace_back(std::move(it));
846
1.30k
    }
847
848
7.92k
    ColumnIteratorUPtr null_iterator;
849
7.92k
    if (is_nullable()) {
850
7.08k
        RETURN_IF_ERROR(_sub_readers[child_size]->new_iterator(&null_iterator, nullptr));
851
7.08k
    }
852
7.92k
    *iterator = std::make_unique<StructFileColumnIterator>(
853
7.92k
            shared_from_this(), std::move(null_iterator), std::move(sub_column_iterators));
854
7.92k
    return Status::OK();
855
7.92k
}
856
857
Result<TColumnAccessPaths> ColumnIterator::_get_sub_access_paths(
858
117k
        const TColumnAccessPaths& access_paths) {
859
117k
    TColumnAccessPaths sub_access_paths = access_paths;
860
181k
    for (auto it = sub_access_paths.begin(); it != sub_access_paths.end();) {
861
63.3k
        TColumnAccessPath& name_path = *it;
862
63.3k
        if (name_path.data_access_path.path.empty()) {
863
1
            return ResultError(
864
1
                    Status::InternalError("Invalid access path for struct column: path is empty"));
865
1
        }
866
867
63.3k
        if (!StringCaseEqual()(name_path.data_access_path.path[0], _column_name)) {
868
1
            return ResultError(Status::InternalError(
869
1
                    R"(Invalid access path for column: expected name "{}", got "{}")", _column_name,
870
1
                    name_path.data_access_path.path[0]));
871
1
        }
872
873
63.3k
        name_path.data_access_path.path.erase(name_path.data_access_path.path.begin());
874
63.3k
        if (!name_path.data_access_path.path.empty()) {
875
6.83k
            ++it;
876
56.4k
        } else {
877
56.4k
            set_need_to_read();
878
56.4k
            it = sub_access_paths.erase(it);
879
56.4k
        }
880
63.3k
    }
881
117k
    return sub_access_paths;
882
117k
}
883
884
///====================== MapFileColumnIterator ============================////
885
MapFileColumnIterator::MapFileColumnIterator(std::shared_ptr<ColumnReader> reader,
886
                                             ColumnIteratorUPtr null_iterator,
887
                                             OffsetFileColumnIteratorUPtr offsets_iterator,
888
                                             ColumnIteratorUPtr key_iterator,
889
                                             ColumnIteratorUPtr val_iterator)
890
36.6k
        : _map_reader(reader),
891
36.6k
          _offsets_iterator(std::move(offsets_iterator)),
892
36.6k
          _key_iterator(std::move(key_iterator)),
893
36.6k
          _val_iterator(std::move(val_iterator)) {
894
36.6k
    if (_map_reader->is_nullable()) {
895
15.6k
        _null_iterator = std::move(null_iterator);
896
15.6k
    }
897
36.6k
}
898
899
36.5k
Status MapFileColumnIterator::init(const ColumnIteratorOptions& opts) {
900
36.5k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
901
32
        DLOG(INFO) << "Map column iterator column " << _column_name << " skip reading.";
902
32
        return Status::OK();
903
32
    }
904
36.5k
    RETURN_IF_ERROR(_key_iterator->init(opts));
905
36.5k
    RETURN_IF_ERROR(_val_iterator->init(opts));
906
36.5k
    RETURN_IF_ERROR(_offsets_iterator->init(opts));
907
36.5k
    if (_map_reader->is_nullable()) {
908
15.5k
        RETURN_IF_ERROR(_null_iterator->init(opts));
909
15.5k
    }
910
36.5k
    return Status::OK();
911
36.5k
}
912
913
19.4k
Status MapFileColumnIterator::seek_to_ordinal(ordinal_t ord) {
914
19.4k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
915
0
        DLOG(INFO) << "Map column iterator column " << _column_name << " skip reading.";
916
0
        return Status::OK();
917
0
    }
918
919
19.4k
    if (read_null_map_only()) {
920
        // In NULL_MAP_ONLY mode, only seek the null iterator; skip offset/key/val iterators
921
5
        if (_map_reader->is_nullable() && _null_iterator) {
922
5
            RETURN_IF_ERROR(_null_iterator->seek_to_ordinal(ord));
923
5
        }
924
5
        return Status::OK();
925
5
    }
926
927
19.4k
    if (_map_reader->is_nullable()) {
928
11.2k
        RETURN_IF_ERROR(_null_iterator->seek_to_ordinal(ord));
929
11.2k
    }
930
19.4k
    RETURN_IF_ERROR(_offsets_iterator->seek_to_ordinal(ord));
931
19.4k
    if (read_offset_only()) {
932
        // In OFFSET_ONLY mode, key/value iterators are SKIP_READING, no need to seek them
933
468
        return Status::OK();
934
468
    }
935
    // here to use offset info
936
19.0k
    ordinal_t offset = 0;
937
19.0k
    RETURN_IF_ERROR(_offsets_iterator->_peek_one_offset(&offset));
938
19.0k
    RETURN_IF_ERROR(_key_iterator->seek_to_ordinal(offset));
939
19.0k
    RETURN_IF_ERROR(_val_iterator->seek_to_ordinal(offset));
940
19.0k
    return Status::OK();
941
19.0k
}
942
943
1.42k
Status MapFileColumnIterator::init_prefetcher(const SegmentPrefetchParams& params) {
944
1.42k
    RETURN_IF_ERROR(_offsets_iterator->init_prefetcher(params));
945
1.42k
    if (_map_reader->is_nullable()) {
946
1.41k
        RETURN_IF_ERROR(_null_iterator->init_prefetcher(params));
947
1.41k
    }
948
1.42k
    RETURN_IF_ERROR(_key_iterator->init_prefetcher(params));
949
1.42k
    RETURN_IF_ERROR(_val_iterator->init_prefetcher(params));
950
1.42k
    return Status::OK();
951
1.42k
}
952
953
void MapFileColumnIterator::collect_prefetchers(
954
        std::map<PrefetcherInitMethod, std::vector<SegmentPrefetcher*>>& prefetchers,
955
1.42k
        PrefetcherInitMethod init_method) {
956
1.42k
    _offsets_iterator->collect_prefetchers(prefetchers, init_method);
957
1.42k
    if (_map_reader->is_nullable()) {
958
1.41k
        _null_iterator->collect_prefetchers(prefetchers, init_method);
959
1.41k
    }
960
    // the actual data pages to read of key/value column depends on the read result of offset column,
961
    // so we can't init prefetch blocks according to rowids, just prefetch all data blocks here.
962
1.42k
    _key_iterator->collect_prefetchers(prefetchers, PrefetcherInitMethod::ALL_DATA_BLOCKS);
963
1.42k
    _val_iterator->collect_prefetchers(prefetchers, PrefetcherInitMethod::ALL_DATA_BLOCKS);
964
1.42k
}
965
966
19.4k
Status MapFileColumnIterator::next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) {
967
19.4k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
968
0
        DLOG(INFO) << "Map column iterator column " << _column_name << " skip reading.";
969
0
        dst->insert_many_defaults(*n);
970
0
        return Status::OK();
971
0
    }
972
973
19.4k
    if (read_null_map_only()) {
974
        // NULL_MAP_ONLY mode: read null map, fill nested ColumnMap with empty defaults
975
5
        DORIS_CHECK(dst->is_nullable());
976
5
        auto& nullable_col = assert_cast<ColumnNullable&>(*dst);
977
5
        auto null_map_ptr = nullable_col.get_null_map_column_ptr();
978
5
        size_t num_read = *n;
979
5
        if (_null_iterator) {
980
5
            bool null_signs_has_null = false;
981
5
            MutableColumnPtr null_map_column = std::move(null_map_ptr);
982
5
            RETURN_IF_ERROR(
983
5
                    _null_iterator->next_batch(&num_read, null_map_column, &null_signs_has_null));
984
5
        } else {
985
            // schema-change: column became nullable but old segment has no null data
986
0
            null_map_ptr->insert_many_vals(0, num_read);
987
0
        }
988
5
        DCHECK(num_read == *n);
989
        // fill nested ColumnMap with empty (zero-element) maps
990
5
        auto& column_map = assert_cast<ColumnMap&, TypeCheckOnRelease::DISABLE>(
991
5
                nullable_col.get_nested_column());
992
5
        column_map.insert_many_defaults(num_read);
993
5
        *has_null = true;
994
5
        return Status::OK();
995
5
    }
996
997
19.4k
    auto& column_map = assert_cast<ColumnMap&, TypeCheckOnRelease::DISABLE>(
998
19.4k
            dst->is_nullable() ? static_cast<ColumnNullable&>(*dst).get_nested_column() : *dst);
999
19.4k
    auto column_offsets_ptr = IColumn::mutate(std::move(column_map.get_offsets_ptr()));
1000
19.4k
    Defer defer_offsets {[&] {
1001
19.4k
        auto typed_column_offsets_ptr = ColumnMap::COffsets::cast_to_column_mutptr(
1002
19.4k
                assert_cast<ColumnMap::COffsets*, TypeCheckOnRelease::DISABLE>(
1003
19.4k
                        column_offsets_ptr.get()));
1004
19.4k
        column_offsets_ptr = nullptr;
1005
19.4k
        column_map.get_offsets_ptr() = std::move(typed_column_offsets_ptr);
1006
19.4k
    }};
1007
19.4k
    bool offsets_has_null = false;
1008
19.4k
    ssize_t start = column_offsets_ptr->size();
1009
19.4k
    RETURN_IF_ERROR(_offsets_iterator->next_batch(n, column_offsets_ptr, &offsets_has_null));
1010
19.4k
    if (*n == 0) {
1011
0
        return Status::OK();
1012
0
    }
1013
19.4k
    auto& column_offsets = static_cast<ColumnArray::ColumnOffsets&>(*column_offsets_ptr);
1014
19.4k
    RETURN_IF_ERROR(_offsets_iterator->_calculate_offsets(start, column_offsets));
1015
19.4k
    DCHECK(column_offsets.get_data().back() >= column_offsets.get_data()[start - 1]);
1016
19.4k
    size_t num_items =
1017
19.4k
            column_offsets.get_data().back() - column_offsets.get_data()[start - 1]; // -1 is valid
1018
1019
19.4k
    if (num_items > 0) {
1020
15.4k
        auto key_ptr = IColumn::mutate(std::move(column_map.get_keys_ptr()));
1021
15.4k
        auto val_ptr = IColumn::mutate(std::move(column_map.get_values_ptr()));
1022
15.4k
        Defer defer_keys {[&] { column_map.get_keys_ptr() = std::move(key_ptr); }};
1023
15.4k
        Defer defer_values {[&] { column_map.get_values_ptr() = std::move(val_ptr); }};
1024
15.4k
        if (read_offset_only()) {
1025
            // OFFSET_ONLY mode: skip reading actual key/value data, fill with defaults
1026
468
            key_ptr->insert_many_defaults(num_items);
1027
468
            val_ptr->insert_many_defaults(num_items);
1028
14.9k
        } else {
1029
14.9k
            size_t num_read = num_items;
1030
14.9k
            bool key_has_null = false;
1031
14.9k
            bool val_has_null = false;
1032
14.9k
            RETURN_IF_ERROR(_key_iterator->next_batch(&num_read, key_ptr, &key_has_null));
1033
14.9k
            RETURN_IF_ERROR(_val_iterator->next_batch(&num_read, val_ptr, &val_has_null));
1034
14.9k
            DCHECK(num_read == num_items);
1035
14.9k
        }
1036
15.4k
    }
1037
1038
19.4k
    if (dst->is_nullable()) {
1039
11.3k
        size_t num_read = *n;
1040
11.3k
        auto null_map_ptr = static_cast<ColumnNullable&>(*dst).get_null_map_column_ptr();
1041
        // in not-null to null linked-schemachange mode,
1042
        // actually we do not change dat data include meta in footer,
1043
        // so may dst from changed meta which is nullable but old data is not nullable,
1044
        // if so, we should set null_map to all null by default
1045
11.3k
        if (_null_iterator) {
1046
11.3k
            bool null_signs_has_null = false;
1047
11.3k
            MutableColumnPtr null_map_column = std::move(null_map_ptr);
1048
11.3k
            RETURN_IF_ERROR(
1049
11.3k
                    _null_iterator->next_batch(&num_read, null_map_column, &null_signs_has_null));
1050
11.3k
        } else {
1051
1
            null_map_ptr->insert_many_vals(0, num_read);
1052
1
        }
1053
11.3k
        DCHECK(num_read == *n);
1054
11.3k
    }
1055
19.4k
    return Status::OK();
1056
19.4k
}
1057
1058
Status MapFileColumnIterator::read_by_rowids(const rowid_t* rowids, const size_t count,
1059
17.0k
                                             MutableColumnPtr& dst) {
1060
17.0k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
1061
1
        DLOG(INFO) << "File column iterator column " << _column_name << " skip reading.";
1062
1
        dst->insert_many_defaults(count);
1063
1
        return Status::OK();
1064
1
    }
1065
1066
17.0k
    if (read_null_map_only()) {
1067
        // NULL_MAP_ONLY mode: read null map by rowids, fill nested ColumnMap with empty defaults
1068
7
        DORIS_CHECK(dst->is_nullable());
1069
7
        auto& nullable_col = assert_cast<ColumnNullable&>(*dst);
1070
7
        if (_null_iterator) {
1071
7
            auto null_map_ptr = nullable_col.get_null_map_column_ptr();
1072
7
            MutableColumnPtr null_map_column = std::move(null_map_ptr);
1073
7
            RETURN_IF_ERROR(_null_iterator->read_by_rowids(rowids, count, null_map_column));
1074
7
        } else {
1075
            // schema-change: column became nullable but old segment has no null data
1076
0
            auto null_map_ptr = nullable_col.get_null_map_column_ptr();
1077
0
            null_map_ptr->insert_many_vals(0, count);
1078
0
        }
1079
        // fill nested ColumnMap with empty (zero-element) maps
1080
7
        auto& column_map = assert_cast<ColumnMap&, TypeCheckOnRelease::DISABLE>(
1081
7
                nullable_col.get_nested_column());
1082
7
        column_map.insert_many_defaults(count);
1083
7
        return Status::OK();
1084
7
    }
1085
1086
17.0k
    if (count == 0) {
1087
0
        return Status::OK();
1088
0
    }
1089
    // resolve ColumnMap and nullable wrapper
1090
17.0k
    auto& column_map = assert_cast<ColumnMap&, TypeCheckOnRelease::DISABLE>(
1091
17.0k
            dst->is_nullable() ? static_cast<ColumnNullable&>(*dst).get_nested_column() : *dst);
1092
17.0k
    auto offsets_ptr = IColumn::mutate(std::move(column_map.get_offsets_ptr()));
1093
17.0k
    Defer defer_offsets {[&] {
1094
17.0k
        auto typed_offsets_ptr = ColumnMap::COffsets::cast_to_column_mutptr(
1095
17.0k
                assert_cast<ColumnMap::COffsets*, TypeCheckOnRelease::DISABLE>(offsets_ptr.get()));
1096
17.0k
        offsets_ptr = nullptr;
1097
17.0k
        column_map.get_offsets_ptr() = std::move(typed_offsets_ptr);
1098
17.0k
    }};
1099
17.0k
    auto& offsets = static_cast<ColumnArray::ColumnOffsets&>(*offsets_ptr);
1100
17.0k
    size_t base = offsets.get_data().empty() ? 0 : offsets.get_data().back();
1101
1102
    // 1. bulk read null-map if nullable
1103
17.0k
    std::vector<uint8_t> null_mask; // 0: not null, 1: null
1104
17.0k
    if (_map_reader->is_nullable()) {
1105
        // For nullable map columns, the destination column must also be nullable.
1106
3.13k
        if (UNLIKELY(!dst->is_nullable())) {
1107
0
            return Status::InternalError(
1108
0
                    "unexpected non-nullable destination column for nullable map reader");
1109
0
        }
1110
3.13k
        auto null_map_ptr = static_cast<ColumnNullable&>(*dst).get_null_map_column_ptr();
1111
3.13k
        size_t null_before = null_map_ptr->size();
1112
3.13k
        auto* null_map_col = null_map_ptr.get();
1113
3.13k
        MutableColumnPtr null_map_column = std::move(null_map_ptr);
1114
3.13k
        RETURN_IF_ERROR(_null_iterator->read_by_rowids(rowids, count, null_map_column));
1115
        // extract a light-weight view to decide element reads
1116
3.13k
        null_mask.reserve(count);
1117
56.5k
        for (size_t i = 0; i < count; ++i) {
1118
53.3k
            null_mask.push_back(null_map_col->get_element(null_before + i));
1119
53.3k
        }
1120
13.9k
    } else if (dst->is_nullable()) {
1121
        // in not-null to null linked-schemachange mode,
1122
        // actually we do not change dat data include meta in footer,
1123
        // so may dst from changed meta which is nullable but old data is not nullable,
1124
        // if so, we should set null_map to all null by default
1125
1
        auto null_map_ptr = static_cast<ColumnNullable&>(*dst).get_null_map_column_ptr();
1126
1
        null_map_ptr->insert_many_vals(0, count);
1127
1
    }
1128
1129
    // 2. bulk read start ordinals for requested rows
1130
17.0k
    MutableColumnPtr starts_col = ColumnOffset64::create();
1131
17.0k
    starts_col->reserve(count);
1132
17.0k
    RETURN_IF_ERROR(_offsets_iterator->read_by_rowids(rowids, count, starts_col));
1133
1134
    // 3. bulk read next-start ordinals for rowid+1 (within bounds)
1135
17.0k
    std::vector<rowid_t> next_rowids(count);
1136
2.16M
    for (size_t i = 0; i < count; ++i) {
1137
2.14M
        uint64_t nr = rowids[i] + 1;
1138
2.14M
        next_rowids[i] = nr < _map_reader->num_rows() ? static_cast<rowid_t>(nr)
1139
2.14M
                                                      : static_cast<rowid_t>(0); // placeholder
1140
2.14M
    }
1141
17.0k
    MutableColumnPtr next_starts_col = ColumnOffset64::create();
1142
17.0k
    next_starts_col->reserve(count);
1143
    // read for all; we'll fix out-of-bound cases below
1144
17.0k
    RETURN_IF_ERROR(_offsets_iterator->read_by_rowids(next_rowids.data(), count, next_starts_col));
1145
1146
    // 4. fix next_start for rows whose next_rowid is out-of-bound (rowid == num_rows-1)
1147
2.16M
    for (size_t i = 0; i < count; ++i) {
1148
2.14M
        if (rowids[i] + 1 >= _map_reader->num_rows()) {
1149
            // seek to the last row and consume one to move decoder to end-of-page,
1150
            // then peek page-tail sentinel next_array_item_ordinal as next_start
1151
15.0k
            RETURN_IF_ERROR(_offsets_iterator->seek_to_ordinal(rowids[i]));
1152
15.0k
            size_t one = 1;
1153
15.0k
            bool has_null_unused = false;
1154
15.0k
            MutableColumnPtr tmp = ColumnOffset64::create();
1155
15.0k
            RETURN_IF_ERROR(_offsets_iterator->next_batch(&one, tmp, &has_null_unused));
1156
15.0k
            ordinal_t ns = 0;
1157
15.0k
            RETURN_IF_ERROR(_offsets_iterator->_peek_one_offset(&ns));
1158
            // overwrite with sentinel
1159
15.0k
            assert_cast<ColumnOffset64&, TypeCheckOnRelease::DISABLE>(*next_starts_col)
1160
15.0k
                    .get_data()[i] = ns;
1161
15.0k
        }
1162
2.14M
    }
1163
1164
    // 5. compute sizes and append offsets prefix-sum
1165
17.0k
    auto& starts_data = assert_cast<ColumnOffset64&>(*starts_col).get_data();
1166
17.0k
    auto& next_starts_data = assert_cast<ColumnOffset64&>(*next_starts_col).get_data();
1167
17.0k
    std::vector<size_t> sizes(count, 0);
1168
17.0k
    size_t acc = base;
1169
17.0k
    const auto original_size = offsets.get_data().back();
1170
17.0k
    offsets.get_data().reserve(offsets.get_data().size() + count);
1171
2.15M
    for (size_t i = 0; i < count; ++i) {
1172
2.13M
        size_t sz = static_cast<size_t>(next_starts_data[i] - starts_data[i]);
1173
2.13M
        if (_map_reader->is_nullable() && !null_mask.empty() && null_mask[i]) {
1174
731
            sz = 0; // null rows do not consume elements
1175
731
        }
1176
2.13M
        sizes[i] = sz;
1177
2.13M
        acc += sz;
1178
2.13M
        offsets.get_data().push_back(acc);
1179
2.13M
    }
1180
1181
    // 6. read key/value elements for non-empty sizes
1182
17.0k
    auto keys_ptr = IColumn::mutate(std::move(column_map.get_keys_ptr()));
1183
17.0k
    auto vals_ptr = IColumn::mutate(std::move(column_map.get_values_ptr()));
1184
17.0k
    Defer defer_keys {[&] { column_map.get_keys_ptr() = std::move(keys_ptr); }};
1185
17.0k
    Defer defer_values {[&] { column_map.get_values_ptr() = std::move(vals_ptr); }};
1186
1187
17.0k
    size_t this_run = sizes[0];
1188
17.0k
    auto start_idx = starts_data[0];
1189
17.0k
    auto last_idx = starts_data[0] + this_run;
1190
2.14M
    for (size_t i = 1; i < count; ++i) {
1191
2.12M
        size_t sz = sizes[i];
1192
2.12M
        if (sz == 0) {
1193
142k
            continue;
1194
142k
        }
1195
1.98M
        auto start = static_cast<ordinal_t>(starts_data[i]);
1196
1.98M
        if (start != last_idx) {
1197
288k
            size_t n = this_run;
1198
288k
            bool dummy_has_null = false;
1199
1200
288k
            if (this_run != 0) {
1201
288k
                if (_key_iterator->reading_flag() != ReadingFlag::SKIP_READING) {
1202
288k
                    RETURN_IF_ERROR(_key_iterator->seek_to_ordinal(start_idx));
1203
288k
                    RETURN_IF_ERROR(_key_iterator->next_batch(&n, keys_ptr, &dummy_has_null));
1204
288k
                    DCHECK(n == this_run);
1205
288k
                }
1206
1207
288k
                if (_val_iterator->reading_flag() != ReadingFlag::SKIP_READING) {
1208
288k
                    n = this_run;
1209
288k
                    RETURN_IF_ERROR(_val_iterator->seek_to_ordinal(start_idx));
1210
288k
                    RETURN_IF_ERROR(_val_iterator->next_batch(&n, vals_ptr, &dummy_has_null));
1211
288k
                    DCHECK(n == this_run);
1212
288k
                }
1213
288k
            }
1214
288k
            start_idx = start;
1215
288k
            this_run = sz;
1216
288k
            last_idx = start + sz;
1217
288k
            continue;
1218
288k
        }
1219
1220
1.69M
        this_run += sz;
1221
1.69M
        last_idx += sz;
1222
1.69M
    }
1223
1224
17.0k
    size_t n = this_run;
1225
17.0k
    const size_t total_count = offsets.get_data().back() - original_size;
1226
17.0k
    bool dummy_has_null = false;
1227
17.0k
    if (_key_iterator->reading_flag() != ReadingFlag::SKIP_READING) {
1228
17.0k
        if (this_run != 0) {
1229
6.73k
            RETURN_IF_ERROR(_key_iterator->seek_to_ordinal(start_idx));
1230
6.73k
            RETURN_IF_ERROR(_key_iterator->next_batch(&n, keys_ptr, &dummy_has_null));
1231
6.73k
            DCHECK(n == this_run);
1232
6.73k
        }
1233
17.0k
    } else {
1234
9
        keys_ptr->insert_many_defaults(total_count);
1235
9
    }
1236
1237
17.0k
    if (_val_iterator->reading_flag() != ReadingFlag::SKIP_READING) {
1238
17.0k
        if (this_run != 0) {
1239
6.72k
            n = this_run;
1240
6.72k
            RETURN_IF_ERROR(_val_iterator->seek_to_ordinal(start_idx));
1241
6.72k
            RETURN_IF_ERROR(_val_iterator->next_batch(&n, vals_ptr, &dummy_has_null));
1242
6.72k
            DCHECK(n == this_run);
1243
6.72k
        }
1244
17.0k
    } else {
1245
18
        vals_ptr->insert_many_defaults(total_count);
1246
18
    }
1247
1248
17.0k
    return Status::OK();
1249
17.0k
}
1250
1251
5.71k
void MapFileColumnIterator::set_need_to_read() {
1252
5.71k
    set_reading_flag(ReadingFlag::NEED_TO_READ);
1253
5.71k
    _key_iterator->set_need_to_read();
1254
5.71k
    _val_iterator->set_need_to_read();
1255
5.71k
}
1256
1257
6.88k
void MapFileColumnIterator::remove_pruned_sub_iterators() {
1258
6.88k
    _key_iterator->remove_pruned_sub_iterators();
1259
6.88k
    _val_iterator->remove_pruned_sub_iterators();
1260
6.88k
}
1261
1262
Status MapFileColumnIterator::set_access_paths(const TColumnAccessPaths& all_access_paths,
1263
6.66k
                                               const TColumnAccessPaths& predicate_access_paths) {
1264
6.66k
    if (all_access_paths.empty()) {
1265
1.51k
        return Status::OK();
1266
1.51k
    }
1267
1268
5.15k
    if (!predicate_access_paths.empty()) {
1269
31
        set_reading_flag(ReadingFlag::READING_FOR_PREDICATE);
1270
31
        DLOG(INFO) << "Map column iterator set sub-column " << _column_name
1271
31
                   << " to READING_FOR_PREDICATE";
1272
31
    }
1273
1274
5.15k
    auto sub_all_access_paths = DORIS_TRY(_get_sub_access_paths(all_access_paths));
1275
5.15k
    auto sub_predicate_access_paths = DORIS_TRY(_get_sub_access_paths(predicate_access_paths));
1276
1277
5.15k
    if (sub_all_access_paths.empty()) {
1278
3.97k
        return Status::OK();
1279
3.97k
    }
1280
1281
    // Check for meta-only modes (OFFSET_ONLY or NULL_MAP_ONLY)
1282
1.17k
    _check_and_set_meta_read_mode(sub_all_access_paths);
1283
1.17k
    if (read_offset_only()) {
1284
465
        _key_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1285
465
        _val_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1286
465
        DLOG(INFO) << "Map column iterator set column " << _column_name
1287
465
                   << " to OFFSET_ONLY reading mode, key/value columns set to SKIP_READING";
1288
465
        return Status::OK();
1289
465
    }
1290
712
    if (read_null_map_only()) {
1291
12
        _key_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1292
12
        _val_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1293
12
        DLOG(INFO) << "Map column iterator set column " << _column_name
1294
12
                   << " to NULL_MAP_ONLY reading mode, key/value columns set to SKIP_READING";
1295
12
        return Status::OK();
1296
12
    }
1297
1298
700
    TColumnAccessPaths key_all_access_paths;
1299
700
    TColumnAccessPaths val_all_access_paths;
1300
700
    TColumnAccessPaths key_predicate_access_paths;
1301
700
    TColumnAccessPaths val_predicate_access_paths;
1302
1303
753
    for (auto paths : sub_all_access_paths) {
1304
753
        if (paths.data_access_path.path[0] == ACCESS_ALL) {
1305
            // ACCESS_ALL means element_at(map, key) style access: the key column must be
1306
            // fully read so that the runtime can match the requested key, while any sub-path
1307
            // qualifiers (e.g. OFFSET) apply only to the value column.
1308
            // For key: create a path with just the column name (= full data access).
1309
202
            TColumnAccessPath key_path;
1310
202
            key_path.__set_type(paths.type);
1311
202
            TDataAccessPath key_data_path;
1312
202
            key_data_path.__set_path({_key_iterator->column_name()});
1313
202
            key_path.__set_data_access_path(key_data_path);
1314
202
            key_all_access_paths.emplace_back(std::move(key_path));
1315
            // For value: pass the full sub-path so qualifiers like OFFSET propagate.
1316
202
            paths.data_access_path.path[0] = _val_iterator->column_name();
1317
202
            val_all_access_paths.emplace_back(paths);
1318
551
        } else if (paths.data_access_path.path[0] == ACCESS_MAP_KEYS) {
1319
279
            paths.data_access_path.path[0] = _key_iterator->column_name();
1320
279
            key_all_access_paths.emplace_back(paths);
1321
279
        } else if (paths.data_access_path.path[0] == ACCESS_MAP_VALUES) {
1322
272
            paths.data_access_path.path[0] = _val_iterator->column_name();
1323
272
            val_all_access_paths.emplace_back(paths);
1324
272
        }
1325
753
    }
1326
700
    const auto need_read_keys = !key_all_access_paths.empty();
1327
700
    const auto need_read_values = !val_all_access_paths.empty();
1328
1329
700
    for (auto paths : sub_predicate_access_paths) {
1330
19
        if (paths.data_access_path.path[0] == ACCESS_ALL) {
1331
            // Same logic as above: key needs full data, value gets the sub-path.
1332
19
            TColumnAccessPath key_path;
1333
19
            key_path.__set_type(paths.type);
1334
19
            TDataAccessPath key_data_path;
1335
19
            key_data_path.__set_path({_key_iterator->column_name()});
1336
19
            key_path.__set_data_access_path(key_data_path);
1337
19
            key_predicate_access_paths.emplace_back(std::move(key_path));
1338
19
            paths.data_access_path.path[0] = _val_iterator->column_name();
1339
19
            val_predicate_access_paths.emplace_back(paths);
1340
19
        } else if (paths.data_access_path.path[0] == ACCESS_MAP_KEYS) {
1341
0
            paths.data_access_path.path[0] = _key_iterator->column_name();
1342
0
            key_predicate_access_paths.emplace_back(paths);
1343
0
        } else if (paths.data_access_path.path[0] == ACCESS_MAP_VALUES) {
1344
0
            paths.data_access_path.path[0] = _val_iterator->column_name();
1345
0
            val_predicate_access_paths.emplace_back(paths);
1346
0
        }
1347
19
    }
1348
1349
700
    if (need_read_keys) {
1350
478
        _key_iterator->set_reading_flag(ReadingFlag::NEED_TO_READ);
1351
478
        RETURN_IF_ERROR(
1352
478
                _key_iterator->set_access_paths(key_all_access_paths, key_predicate_access_paths));
1353
478
    } else {
1354
222
        _key_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1355
222
        DLOG(INFO) << "Map column iterator set key column to SKIP_READING";
1356
222
    }
1357
1358
700
    if (need_read_values) {
1359
472
        _val_iterator->set_reading_flag(ReadingFlag::NEED_TO_READ);
1360
472
        RETURN_IF_ERROR(
1361
472
                _val_iterator->set_access_paths(val_all_access_paths, val_predicate_access_paths));
1362
472
    } else {
1363
228
        _val_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1364
228
        DLOG(INFO) << "Map column iterator set value column to SKIP_READING";
1365
228
    }
1366
700
    return Status::OK();
1367
700
}
1368
1369
////////////////////////////////////////////////////////////////////////////////
1370
1371
StructFileColumnIterator::StructFileColumnIterator(
1372
        std::shared_ptr<ColumnReader> reader, ColumnIteratorUPtr null_iterator,
1373
        std::vector<ColumnIteratorUPtr>&& sub_column_iterators)
1374
7.91k
        : _struct_reader(reader), _sub_column_iterators(std::move(sub_column_iterators)) {
1375
7.91k
    if (_struct_reader->is_nullable()) {
1376
7.06k
        _null_iterator = std::move(null_iterator);
1377
7.06k
    }
1378
7.91k
}
1379
1380
7.89k
Status StructFileColumnIterator::init(const ColumnIteratorOptions& opts) {
1381
7.89k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
1382
0
        DLOG(INFO) << "Struct column iterator column " << _column_name << " skip reading.";
1383
0
        return Status::OK();
1384
0
    }
1385
1386
27.6k
    for (auto& column_iterator : _sub_column_iterators) {
1387
27.6k
        RETURN_IF_ERROR(column_iterator->init(opts));
1388
27.6k
    }
1389
7.89k
    if (_struct_reader->is_nullable()) {
1390
7.07k
        RETURN_IF_ERROR(_null_iterator->init(opts));
1391
7.07k
    }
1392
7.89k
    return Status::OK();
1393
7.89k
}
1394
1395
4.63k
Status StructFileColumnIterator::next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) {
1396
4.63k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
1397
0
        DLOG(INFO) << "Struct column iterator column " << _column_name << " skip reading.";
1398
0
        dst->insert_many_defaults(*n);
1399
0
        return Status::OK();
1400
0
    }
1401
1402
4.63k
    if (read_null_map_only()) {
1403
        // NULL_MAP_ONLY mode: read null map, fill nested ColumnStruct with empty defaults
1404
2
        DORIS_CHECK(dst->is_nullable());
1405
2
        auto& nullable_col = assert_cast<ColumnNullable&>(*dst);
1406
2
        auto null_map_ptr = nullable_col.get_null_map_column_ptr();
1407
2
        size_t num_read = *n;
1408
2
        if (_null_iterator) {
1409
2
            bool null_signs_has_null = false;
1410
2
            MutableColumnPtr null_map_column = std::move(null_map_ptr);
1411
2
            RETURN_IF_ERROR(
1412
2
                    _null_iterator->next_batch(&num_read, null_map_column, &null_signs_has_null));
1413
2
        } else {
1414
            // schema-change: column became nullable but old segment has no null data
1415
0
            null_map_ptr->insert_many_vals(0, num_read);
1416
0
        }
1417
2
        DCHECK(num_read == *n);
1418
        // fill nested ColumnStruct with defaults to maintain consistent column sizes
1419
2
        auto& column_struct = assert_cast<ColumnStruct&, TypeCheckOnRelease::DISABLE>(
1420
2
                nullable_col.get_nested_column());
1421
2
        column_struct.insert_many_defaults(num_read);
1422
2
        *has_null = true;
1423
2
        return Status::OK();
1424
2
    }
1425
1426
4.62k
    auto& column_struct = assert_cast<ColumnStruct&, TypeCheckOnRelease::DISABLE>(
1427
4.62k
            dst->is_nullable() ? static_cast<ColumnNullable&>(*dst).get_nested_column() : *dst);
1428
19.3k
    for (size_t i = 0; i < column_struct.tuple_size(); i++) {
1429
14.7k
        size_t num_read = *n;
1430
14.7k
        auto sub_column_ptr = IColumn::mutate(std::move(column_struct.get_column_ptr(i)));
1431
14.7k
        Defer defer_sub_column {
1432
14.7k
                [&] { column_struct.get_column_ptr(i) = std::move(sub_column_ptr); }};
1433
14.7k
        bool column_has_null = false;
1434
14.7k
        RETURN_IF_ERROR(
1435
14.7k
                _sub_column_iterators[i]->next_batch(&num_read, sub_column_ptr, &column_has_null));
1436
14.7k
        DCHECK(num_read == *n);
1437
14.7k
    }
1438
1439
4.62k
    if (dst->is_nullable()) {
1440
4.31k
        size_t num_read = *n;
1441
4.31k
        auto null_map_ptr = static_cast<ColumnNullable&>(*dst).get_null_map_column_ptr();
1442
        // in not-null to null linked-schemachange mode,
1443
        // actually we do not change dat data include meta in footer,
1444
        // so may dst from changed meta which is nullable but old data is not nullable,
1445
        // if so, we should set null_map to all null by default
1446
4.31k
        if (_null_iterator) {
1447
4.22k
            bool null_signs_has_null = false;
1448
4.22k
            MutableColumnPtr null_map_column = std::move(null_map_ptr);
1449
4.22k
            RETURN_IF_ERROR(
1450
4.22k
                    _null_iterator->next_batch(&num_read, null_map_column, &null_signs_has_null));
1451
4.22k
        } else {
1452
89
            null_map_ptr->insert_many_vals(0, num_read);
1453
89
        }
1454
4.31k
        DCHECK(num_read == *n);
1455
4.31k
    }
1456
1457
4.62k
    return Status::OK();
1458
4.62k
}
1459
1460
4.63k
Status StructFileColumnIterator::seek_to_ordinal(ordinal_t ord) {
1461
4.63k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
1462
0
        DLOG(INFO) << "Struct column iterator column " << _column_name << " skip reading.";
1463
0
        return Status::OK();
1464
0
    }
1465
1466
4.63k
    if (read_null_map_only()) {
1467
        // In NULL_MAP_ONLY mode, only seek the null iterator; skip all sub-column iterators
1468
2
        if (_struct_reader->is_nullable() && _null_iterator) {
1469
2
            RETURN_IF_ERROR(_null_iterator->seek_to_ordinal(ord));
1470
2
        }
1471
2
        return Status::OK();
1472
2
    }
1473
1474
14.7k
    for (auto& column_iterator : _sub_column_iterators) {
1475
14.7k
        RETURN_IF_ERROR(column_iterator->seek_to_ordinal(ord));
1476
14.7k
    }
1477
4.63k
    if (_struct_reader->is_nullable()) {
1478
4.22k
        RETURN_IF_ERROR(_null_iterator->seek_to_ordinal(ord));
1479
4.22k
    }
1480
4.63k
    return Status::OK();
1481
4.63k
}
1482
1483
439
Status StructFileColumnIterator::init_prefetcher(const SegmentPrefetchParams& params) {
1484
1.32k
    for (auto& column_iterator : _sub_column_iterators) {
1485
1.32k
        RETURN_IF_ERROR(column_iterator->init_prefetcher(params));
1486
1.32k
    }
1487
439
    if (_struct_reader->is_nullable()) {
1488
433
        RETURN_IF_ERROR(_null_iterator->init_prefetcher(params));
1489
433
    }
1490
439
    return Status::OK();
1491
439
}
1492
1493
void StructFileColumnIterator::collect_prefetchers(
1494
        std::map<PrefetcherInitMethod, std::vector<SegmentPrefetcher*>>& prefetchers,
1495
439
        PrefetcherInitMethod init_method) {
1496
1.32k
    for (auto& column_iterator : _sub_column_iterators) {
1497
1.32k
        column_iterator->collect_prefetchers(prefetchers, init_method);
1498
1.32k
    }
1499
439
    if (_struct_reader->is_nullable()) {
1500
432
        _null_iterator->collect_prefetchers(prefetchers, init_method);
1501
432
    }
1502
439
}
1503
1504
Status StructFileColumnIterator::read_by_rowids(const rowid_t* rowids, const size_t count,
1505
2.72k
                                                MutableColumnPtr& dst) {
1506
2.72k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
1507
0
        DLOG(INFO) << "Struct column iterator column " << _column_name << " skip reading.";
1508
0
        dst->insert_many_defaults(count);
1509
0
        return Status::OK();
1510
0
    }
1511
1512
2.72k
    if (count == 0) {
1513
0
        return Status::OK();
1514
0
    }
1515
1516
2.72k
    size_t this_run = 1;
1517
2.72k
    auto start_idx = rowids[0];
1518
2.72k
    auto last_idx = rowids[0];
1519
2.83k
    for (size_t i = 1; i < count; ++i) {
1520
114
        if (last_idx == rowids[i] - 1) {
1521
108
            last_idx = rowids[i];
1522
108
            this_run++;
1523
108
            continue;
1524
108
        }
1525
6
        RETURN_IF_ERROR(seek_to_ordinal(start_idx));
1526
6
        size_t num_read = this_run;
1527
6
        RETURN_IF_ERROR(next_batch(&num_read, dst));
1528
6
        DCHECK_EQ(num_read, this_run);
1529
1530
6
        start_idx = rowids[i];
1531
6
        last_idx = rowids[i];
1532
6
        this_run = 1;
1533
6
    }
1534
1535
2.72k
    RETURN_IF_ERROR(seek_to_ordinal(start_idx));
1536
2.72k
    size_t num_read = this_run;
1537
2.72k
    RETURN_IF_ERROR(next_batch(&num_read, dst));
1538
2.72k
    DCHECK_EQ(num_read, this_run);
1539
2.72k
    return Status::OK();
1540
2.72k
}
1541
1542
6.90k
void StructFileColumnIterator::set_need_to_read() {
1543
6.90k
    set_reading_flag(ReadingFlag::NEED_TO_READ);
1544
24.9k
    for (auto& sub_iterator : _sub_column_iterators) {
1545
24.9k
        sub_iterator->set_need_to_read();
1546
24.9k
    }
1547
6.90k
}
1548
1549
7.05k
void StructFileColumnIterator::remove_pruned_sub_iterators() {
1550
33.2k
    for (auto it = _sub_column_iterators.begin(); it != _sub_column_iterators.end();) {
1551
26.1k
        auto& sub_iterator = *it;
1552
26.1k
        if (sub_iterator->reading_flag() == ReadingFlag::SKIP_READING) {
1553
857
            DLOG(INFO) << "Struct column iterator remove pruned sub-column "
1554
857
                       << sub_iterator->column_name();
1555
857
            it = _sub_column_iterators.erase(it);
1556
25.3k
        } else {
1557
25.3k
            sub_iterator->remove_pruned_sub_iterators();
1558
25.3k
            ++it;
1559
25.3k
        }
1560
26.1k
    }
1561
7.05k
}
1562
1563
Status StructFileColumnIterator::set_access_paths(
1564
        const TColumnAccessPaths& all_access_paths,
1565
6.51k
        const TColumnAccessPaths& predicate_access_paths) {
1566
6.51k
    if (all_access_paths.empty()) {
1567
1.87k
        return Status::OK();
1568
1.87k
    }
1569
1570
4.64k
    if (!predicate_access_paths.empty()) {
1571
61
        set_reading_flag(ReadingFlag::READING_FOR_PREDICATE);
1572
61
        DLOG(INFO) << "Struct column iterator set sub-column " << _column_name
1573
61
                   << " to READING_FOR_PREDICATE";
1574
61
    }
1575
4.64k
    auto sub_all_access_paths = DORIS_TRY(_get_sub_access_paths(all_access_paths));
1576
4.63k
    auto sub_predicate_access_paths = DORIS_TRY(_get_sub_access_paths(predicate_access_paths));
1577
1578
    // Check for NULL_MAP_ONLY mode: only read null map, skip all sub-columns
1579
4.63k
    _check_and_set_meta_read_mode(sub_all_access_paths);
1580
4.63k
    if (read_null_map_only()) {
1581
6
        for (auto& sub_iterator : _sub_column_iterators) {
1582
6
            sub_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1583
6
        }
1584
3
        DLOG(INFO) << "Struct column iterator set column " << _column_name
1585
3
                   << " to NULL_MAP_ONLY reading mode, all sub-columns set to SKIP_READING";
1586
3
        return Status::OK();
1587
3
    }
1588
1589
4.63k
    const auto no_sub_column_to_skip = sub_all_access_paths.empty();
1590
4.63k
    const auto no_predicate_sub_column = sub_predicate_access_paths.empty();
1591
1592
20.6k
    for (auto& sub_iterator : _sub_column_iterators) {
1593
20.6k
        const auto name = sub_iterator->column_name();
1594
20.6k
        bool need_to_read = no_sub_column_to_skip;
1595
20.6k
        TColumnAccessPaths sub_all_access_paths_of_this;
1596
20.6k
        if (!need_to_read) {
1597
1.75k
            for (const auto& paths : sub_all_access_paths) {
1598
1.75k
                if (paths.data_access_path.path[0] == name) {
1599
328
                    sub_all_access_paths_of_this.emplace_back(paths);
1600
328
                }
1601
1.75k
            }
1602
1.18k
            need_to_read = !sub_all_access_paths_of_this.empty();
1603
1.18k
        }
1604
1605
20.6k
        if (!need_to_read) {
1606
854
            sub_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1607
854
            DLOG(INFO) << "Struct column iterator set sub-column " << name << " to SKIP_READING";
1608
854
            continue;
1609
854
        }
1610
19.8k
        set_reading_flag(ReadingFlag::NEED_TO_READ);
1611
19.8k
        sub_iterator->set_reading_flag(ReadingFlag::NEED_TO_READ);
1612
1613
19.8k
        TColumnAccessPaths sub_predicate_access_paths_of_this;
1614
1615
19.8k
        if (!no_predicate_sub_column) {
1616
59
            for (const auto& paths : sub_predicate_access_paths) {
1617
59
                if (StringCaseEqual()(paths.data_access_path.path[0], name)) {
1618
56
                    sub_predicate_access_paths_of_this.emplace_back(paths);
1619
56
                }
1620
59
            }
1621
59
        }
1622
1623
19.8k
        RETURN_IF_ERROR(sub_iterator->set_access_paths(sub_all_access_paths_of_this,
1624
19.8k
                                                       sub_predicate_access_paths_of_this));
1625
19.8k
    }
1626
4.63k
    return Status::OK();
1627
4.63k
}
1628
1629
////////////////////////////////////////////////////////////////////////////////
1630
99.3k
Status OffsetFileColumnIterator::init(const ColumnIteratorOptions& opts) {
1631
99.3k
    RETURN_IF_ERROR(_offset_iterator->init(opts));
1632
    // allocate peek tmp column once
1633
99.3k
    _peek_tmp_col = ColumnOffset64::create();
1634
99.3k
    return Status::OK();
1635
99.3k
}
1636
1637
158k
Status OffsetFileColumnIterator::next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) {
1638
158k
    RETURN_IF_ERROR(_offset_iterator->next_batch(n, dst, has_null));
1639
158k
    return Status::OK();
1640
158k
}
1641
1642
300k
Status OffsetFileColumnIterator::_peek_one_offset(ordinal_t* offset) {
1643
300k
    if (_offset_iterator->get_current_page()->has_remaining()) {
1644
213k
        PageDecoder* offset_page_decoder = _offset_iterator->get_current_page()->data_decoder.get();
1645
213k
        size_t n = 1;
1646
213k
        _peek_tmp_col->clear();
1647
213k
        RETURN_IF_ERROR(offset_page_decoder->peek_next_batch(&n, _peek_tmp_col)); // not null
1648
213k
        DCHECK(_peek_tmp_col->size() == 1);
1649
213k
        *offset =
1650
213k
                assert_cast<const ColumnOffset64*, TypeCheckOnRelease::DISABLE>(_peek_tmp_col.get())
1651
213k
                        ->get_element(0);
1652
213k
    } else {
1653
87.6k
        *offset = _offset_iterator->get_current_page()->next_array_item_ordinal;
1654
87.6k
    }
1655
300k
    return Status::OK();
1656
300k
}
1657
1658
6.73k
Status OffsetFileColumnIterator::init_prefetcher(const SegmentPrefetchParams& params) {
1659
6.73k
    return _offset_iterator->init_prefetcher(params);
1660
6.73k
}
1661
1662
void OffsetFileColumnIterator::collect_prefetchers(
1663
        std::map<PrefetcherInitMethod, std::vector<SegmentPrefetcher*>>& prefetchers,
1664
6.73k
        PrefetcherInitMethod init_method) {
1665
6.73k
    _offset_iterator->collect_prefetchers(prefetchers, init_method);
1666
6.73k
}
1667
1668
/**
1669
 *  first_storage_offset read from page should smaller than next_storage_offset which here call _peek_one_offset from page,
1670
    and first_column_offset is keep in memory data which is different dimension with (first_storage_offset and next_storage_offset)
1671
     eg. step1. read page: first_storage_offset = 16382
1672
         step2. read page below with _peek_one_offset(&last_offset): last_offset = 16387
1673
         step3. first_offset = 126 which is calculate in column offsets
1674
         for loop column offsets element in size
1675
            we can calculate from first_storage_offset to next_storage_offset one by one to fill with offsets_data in memory column offsets
1676
 * @param start
1677
 * @param column_offsets
1678
 * @return
1679
 */
1680
Status OffsetFileColumnIterator::_calculate_offsets(ssize_t start,
1681
143k
                                                    ColumnArray::ColumnOffsets& column_offsets) {
1682
143k
    ordinal_t next_storage_offset = 0;
1683
143k
    RETURN_IF_ERROR(_peek_one_offset(&next_storage_offset));
1684
1685
    // calculate real offsets
1686
143k
    auto& offsets_data = column_offsets.get_data();
1687
143k
    ordinal_t first_column_offset = offsets_data[start - 1]; // -1 is valid
1688
143k
    ordinal_t first_storage_offset = offsets_data[start];
1689
143k
    DCHECK(next_storage_offset >= first_storage_offset);
1690
8.06M
    for (ssize_t i = start; i < offsets_data.size() - 1; ++i) {
1691
7.91M
        offsets_data[i] = first_column_offset + (offsets_data[i + 1] - first_storage_offset);
1692
7.91M
    }
1693
    // last offset
1694
143k
    offsets_data[offsets_data.size() - 1] =
1695
143k
            first_column_offset + (next_storage_offset - first_storage_offset);
1696
143k
    return Status::OK();
1697
143k
}
1698
1699
////////////////////////////////////////////////////////////////////////////////
1700
ArrayFileColumnIterator::ArrayFileColumnIterator(std::shared_ptr<ColumnReader> reader,
1701
                                                 OffsetFileColumnIteratorUPtr offset_reader,
1702
                                                 ColumnIteratorUPtr item_iterator,
1703
                                                 ColumnIteratorUPtr null_iterator)
1704
62.8k
        : _array_reader(reader),
1705
62.8k
          _offset_iterator(std::move(offset_reader)),
1706
62.8k
          _item_iterator(std::move(item_iterator)) {
1707
62.8k
    if (_array_reader->is_nullable()) {
1708
44.4k
        _null_iterator = std::move(null_iterator);
1709
44.4k
    }
1710
62.8k
}
1711
1712
62.8k
Status ArrayFileColumnIterator::init(const ColumnIteratorOptions& opts) {
1713
62.8k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
1714
56
        DLOG(INFO) << "Array column iterator column " << _column_name << " skip readking.";
1715
56
        return Status::OK();
1716
56
    }
1717
1718
62.7k
    RETURN_IF_ERROR(_offset_iterator->init(opts));
1719
62.7k
    RETURN_IF_ERROR(_item_iterator->init(opts));
1720
62.7k
    if (_array_reader->is_nullable()) {
1721
44.4k
        RETURN_IF_ERROR(_null_iterator->init(opts));
1722
44.4k
    }
1723
62.7k
    return Status::OK();
1724
62.7k
}
1725
1726
124k
Status ArrayFileColumnIterator::_seek_by_offsets(ordinal_t ord) {
1727
124k
    if (read_offset_only()) {
1728
        // In OFFSET_ONLY mode, item iterator is SKIP_READING, no need to seek it
1729
1.18k
        return Status::OK();
1730
1.18k
    }
1731
    // using offsets info
1732
123k
    ordinal_t offset = 0;
1733
123k
    RETURN_IF_ERROR(_offset_iterator->_peek_one_offset(&offset));
1734
123k
    RETURN_IF_ERROR(_item_iterator->seek_to_ordinal(offset));
1735
123k
    return Status::OK();
1736
123k
}
1737
1738
124k
Status ArrayFileColumnIterator::seek_to_ordinal(ordinal_t ord) {
1739
124k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
1740
0
        DLOG(INFO) << "Array column iterator column " << _column_name << " skip reading.";
1741
0
        return Status::OK();
1742
0
    }
1743
1744
124k
    if (read_null_map_only()) {
1745
        // In NULL_MAP_ONLY mode, only seek the null iterator; skip offset and item iterators
1746
2
        if (_array_reader->is_nullable() && _null_iterator) {
1747
2
            RETURN_IF_ERROR(_null_iterator->seek_to_ordinal(ord));
1748
2
        }
1749
2
        return Status::OK();
1750
2
    }
1751
1752
124k
    RETURN_IF_ERROR(_offset_iterator->seek_to_ordinal(ord));
1753
124k
    if (_array_reader->is_nullable()) {
1754
105k
        RETURN_IF_ERROR(_null_iterator->seek_to_ordinal(ord));
1755
105k
    }
1756
124k
    return _seek_by_offsets(ord);
1757
124k
}
1758
1759
124k
Status ArrayFileColumnIterator::next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) {
1760
124k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
1761
0
        DLOG(INFO) << "Array column iterator column " << _column_name << " skip reading.";
1762
0
        dst->insert_many_defaults(*n);
1763
0
        return Status::OK();
1764
0
    }
1765
1766
124k
    if (read_null_map_only()) {
1767
        // NULL_MAP_ONLY mode: read null map, fill nested ColumnArray with empty defaults
1768
2
        DORIS_CHECK(dst->is_nullable());
1769
2
        auto& nullable_col = assert_cast<ColumnNullable&>(*dst);
1770
2
        auto null_map_ptr = nullable_col.get_null_map_column_ptr();
1771
2
        size_t num_read = *n;
1772
2
        if (_null_iterator) {
1773
2
            bool null_signs_has_null = false;
1774
2
            MutableColumnPtr null_map_column = std::move(null_map_ptr);
1775
2
            RETURN_IF_ERROR(
1776
2
                    _null_iterator->next_batch(&num_read, null_map_column, &null_signs_has_null));
1777
2
        } else {
1778
            // schema-change: column became nullable but old segment has no null data
1779
0
            null_map_ptr->insert_many_vals(0, num_read);
1780
0
        }
1781
2
        DCHECK(num_read == *n);
1782
        // fill nested ColumnArray with empty (zero-length) arrays
1783
2
        auto& column_array = assert_cast<ColumnArray&, TypeCheckOnRelease::DISABLE>(
1784
2
                nullable_col.get_nested_column());
1785
2
        column_array.insert_many_defaults(num_read);
1786
2
        *has_null = true;
1787
2
        return Status::OK();
1788
2
    }
1789
1790
124k
    auto& column_array = assert_cast<ColumnArray&, TypeCheckOnRelease::DISABLE>(
1791
124k
            dst->is_nullable() ? static_cast<ColumnNullable&>(*dst).get_nested_column() : *dst);
1792
1793
124k
    bool offsets_has_null = false;
1794
124k
    auto column_offsets_ptr = std::move(*column_array.get_offsets_ptr()).mutate();
1795
124k
    Defer defer_offsets {[&] {
1796
124k
        auto typed_column_offsets_ptr = ColumnArray::ColumnOffsets::cast_to_column_mutptr(
1797
124k
                assert_cast<ColumnArray::ColumnOffsets*, TypeCheckOnRelease::DISABLE>(
1798
124k
                        column_offsets_ptr.get()));
1799
124k
        column_offsets_ptr = nullptr;
1800
124k
        column_array.get_offsets_ptr() = std::move(typed_column_offsets_ptr);
1801
124k
    }};
1802
124k
    ssize_t start = column_offsets_ptr->size();
1803
124k
    RETURN_IF_ERROR(_offset_iterator->next_batch(n, column_offsets_ptr, &offsets_has_null));
1804
124k
    if (*n == 0) {
1805
0
        return Status::OK();
1806
0
    }
1807
124k
    auto& column_offsets = static_cast<ColumnArray::ColumnOffsets&>(*column_offsets_ptr);
1808
124k
    RETURN_IF_ERROR(_offset_iterator->_calculate_offsets(start, column_offsets));
1809
124k
    size_t num_items =
1810
124k
            column_offsets.get_data().back() - column_offsets.get_data()[start - 1]; // -1 is valid
1811
124k
    if (num_items > 0) {
1812
88.9k
        auto column_items_ptr = IColumn::mutate(std::move(column_array.get_data_ptr()));
1813
88.9k
        Defer defer_items {[&] { column_array.get_data_ptr() = std::move(column_items_ptr); }};
1814
88.9k
        if (read_offset_only()) {
1815
            // OFFSET_ONLY mode: skip reading actual item data, fill with defaults
1816
1.02k
            column_items_ptr->insert_many_defaults(num_items);
1817
87.9k
        } else {
1818
87.9k
            size_t num_read = num_items;
1819
87.9k
            bool items_has_null = false;
1820
87.9k
            RETURN_IF_ERROR(
1821
87.9k
                    _item_iterator->next_batch(&num_read, column_items_ptr, &items_has_null));
1822
87.9k
            DCHECK(num_read == num_items);
1823
87.9k
        }
1824
88.9k
    }
1825
1826
124k
    if (dst->is_nullable()) {
1827
105k
        auto null_map_ptr = static_cast<ColumnNullable&>(*dst).get_null_map_column_ptr();
1828
105k
        size_t num_read = *n;
1829
        // in not-null to null linked-schemachange mode,
1830
        // actually we do not change dat data include meta in footer,
1831
        // so may dst from changed meta which is nullable but old data is not nullable,
1832
        // if so, we should set null_map to all null by default
1833
105k
        if (_null_iterator) {
1834
105k
            bool null_signs_has_null = false;
1835
105k
            MutableColumnPtr null_map_column = std::move(null_map_ptr);
1836
105k
            RETURN_IF_ERROR(
1837
105k
                    _null_iterator->next_batch(&num_read, null_map_column, &null_signs_has_null));
1838
18.4E
        } else {
1839
18.4E
            null_map_ptr->insert_many_vals(0, num_read);
1840
18.4E
        }
1841
105k
        DCHECK(num_read == *n);
1842
105k
    }
1843
1844
124k
    return Status::OK();
1845
124k
}
1846
1847
5.30k
Status ArrayFileColumnIterator::init_prefetcher(const SegmentPrefetchParams& params) {
1848
5.30k
    RETURN_IF_ERROR(_offset_iterator->init_prefetcher(params));
1849
5.30k
    RETURN_IF_ERROR(_item_iterator->init_prefetcher(params));
1850
5.30k
    if (_array_reader->is_nullable()) {
1851
4.08k
        RETURN_IF_ERROR(_null_iterator->init_prefetcher(params));
1852
4.08k
    }
1853
5.30k
    return Status::OK();
1854
5.30k
}
1855
1856
void ArrayFileColumnIterator::collect_prefetchers(
1857
        std::map<PrefetcherInitMethod, std::vector<SegmentPrefetcher*>>& prefetchers,
1858
5.30k
        PrefetcherInitMethod init_method) {
1859
5.30k
    _offset_iterator->collect_prefetchers(prefetchers, init_method);
1860
    // the actual data pages to read of item column depends on the read result of offset column,
1861
    // so we can't init prefetch blocks according to rowids, just prefetch all data blocks here.
1862
5.30k
    _item_iterator->collect_prefetchers(prefetchers, PrefetcherInitMethod::ALL_DATA_BLOCKS);
1863
5.30k
    if (_array_reader->is_nullable()) {
1864
4.08k
        _null_iterator->collect_prefetchers(prefetchers, init_method);
1865
4.08k
    }
1866
5.30k
}
1867
1868
Status ArrayFileColumnIterator::read_by_rowids(const rowid_t* rowids, const size_t count,
1869
31.8k
                                               MutableColumnPtr& dst) {
1870
31.8k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
1871
0
        DLOG(INFO) << "Array column iterator column " << _column_name << " skip reading.";
1872
0
        dst->insert_many_defaults(count);
1873
0
        return Status::OK();
1874
0
    }
1875
1876
130k
    for (size_t i = 0; i < count; ++i) {
1877
        // TODO(cambyszju): now read array one by one, need optimize later
1878
98.6k
        RETURN_IF_ERROR(seek_to_ordinal(rowids[i]));
1879
98.6k
        size_t num_read = 1;
1880
98.6k
        RETURN_IF_ERROR(next_batch(&num_read, dst));
1881
98.6k
    }
1882
31.8k
    return Status::OK();
1883
31.8k
}
1884
1885
49.5k
void ArrayFileColumnIterator::set_need_to_read() {
1886
49.5k
    set_reading_flag(ReadingFlag::NEED_TO_READ);
1887
49.5k
    _item_iterator->set_need_to_read();
1888
49.5k
}
1889
1890
51.2k
void ArrayFileColumnIterator::remove_pruned_sub_iterators() {
1891
51.2k
    _item_iterator->remove_pruned_sub_iterators();
1892
51.2k
}
1893
1894
Status ArrayFileColumnIterator::set_access_paths(const TColumnAccessPaths& all_access_paths,
1895
50.3k
                                                 const TColumnAccessPaths& predicate_access_paths) {
1896
50.3k
    if (all_access_paths.empty()) {
1897
1.75k
        return Status::OK();
1898
1.75k
    }
1899
1900
48.5k
    if (!predicate_access_paths.empty()) {
1901
2.96k
        set_reading_flag(ReadingFlag::READING_FOR_PREDICATE);
1902
2.96k
        DLOG(INFO) << "Array column iterator set sub-column " << _column_name
1903
2.96k
                   << " to READING_FOR_PREDICATE";
1904
2.96k
    }
1905
1906
48.5k
    auto sub_all_access_paths = DORIS_TRY(_get_sub_access_paths(all_access_paths));
1907
48.5k
    auto sub_predicate_access_paths = DORIS_TRY(_get_sub_access_paths(predicate_access_paths));
1908
1909
    // Check for meta-only modes (OFFSET_ONLY or NULL_MAP_ONLY)
1910
48.5k
    _check_and_set_meta_read_mode(sub_all_access_paths);
1911
48.5k
    if (read_offset_only()) {
1912
1.16k
        _item_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1913
1.16k
        DLOG(INFO) << "Array column iterator set column " << _column_name
1914
1.16k
                   << " to OFFSET_ONLY reading mode, item column set to SKIP_READING";
1915
1.16k
        return Status::OK();
1916
1.16k
    }
1917
47.3k
    if (read_null_map_only()) {
1918
12
        _item_iterator->set_reading_flag(ReadingFlag::SKIP_READING);
1919
12
        DLOG(INFO) << "Array column iterator set column " << _column_name
1920
12
                   << " to NULL_MAP_ONLY reading mode, item column set to SKIP_READING";
1921
12
        return Status::OK();
1922
12
    }
1923
1924
47.3k
    const auto no_sub_column_to_skip = sub_all_access_paths.empty();
1925
47.3k
    const auto no_predicate_sub_column = sub_predicate_access_paths.empty();
1926
1927
47.3k
    if (!no_sub_column_to_skip) {
1928
3.46k
        for (auto& path : sub_all_access_paths) {
1929
3.46k
            if (path.data_access_path.path[0] == ACCESS_ALL) {
1930
3.46k
                path.data_access_path.path[0] = _item_iterator->column_name();
1931
3.46k
            }
1932
3.46k
        }
1933
3.44k
    }
1934
1935
47.3k
    if (!no_predicate_sub_column) {
1936
202
        for (auto& path : sub_predicate_access_paths) {
1937
202
            if (path.data_access_path.path[0] == ACCESS_ALL) {
1938
202
                path.data_access_path.path[0] = _item_iterator->column_name();
1939
202
            }
1940
202
        }
1941
202
    }
1942
1943
47.3k
    if (!no_sub_column_to_skip || !no_predicate_sub_column) {
1944
3.43k
        _item_iterator->set_reading_flag(ReadingFlag::NEED_TO_READ);
1945
3.43k
        RETURN_IF_ERROR(
1946
3.43k
                _item_iterator->set_access_paths(sub_all_access_paths, sub_predicate_access_paths));
1947
3.43k
    }
1948
47.3k
    return Status::OK();
1949
47.3k
}
1950
1951
////////////////////////////////////////////////////////////////////////////////
1952
// StringFileColumnIterator implementation
1953
////////////////////////////////////////////////////////////////////////////////
1954
1955
StringFileColumnIterator::StringFileColumnIterator(std::shared_ptr<ColumnReader> reader)
1956
16.3M
        : FileColumnIterator(std::move(reader)) {}
1957
1958
16.2M
Status StringFileColumnIterator::init(const ColumnIteratorOptions& opts) {
1959
16.2M
    if (read_offset_only()) {
1960
        // Propagate only_read_offsets to the FileColumnIterator's options
1961
121
        auto modified_opts = opts;
1962
121
        modified_opts.only_read_offsets = true;
1963
121
        return FileColumnIterator::init(modified_opts);
1964
121
    }
1965
16.2M
    return FileColumnIterator::init(opts);
1966
16.2M
}
1967
1968
Status StringFileColumnIterator::set_access_paths(
1969
        const TColumnAccessPaths& all_access_paths,
1970
4.80k
        const TColumnAccessPaths& predicate_access_paths) {
1971
4.80k
    if (all_access_paths.empty()) {
1972
3.76k
        return Status::OK();
1973
3.76k
    }
1974
1975
1.03k
    if (!predicate_access_paths.empty()) {
1976
185
        set_reading_flag(ReadingFlag::READING_FOR_PREDICATE);
1977
185
    }
1978
1979
    // Strip the column name from path[0] before checking for meta-only modes.
1980
    // Raw paths look like ["col_name", "OFFSET"] or ["col_name", "NULL"].
1981
1.03k
    auto sub_all_access_paths = DORIS_TRY(_get_sub_access_paths(all_access_paths));
1982
1.03k
    _check_and_set_meta_read_mode(sub_all_access_paths);
1983
    // OFFSET_ONLY mode is fundamentally incompatible with CHAR columns:
1984
    // CHAR is stored padded to its declared length (see
1985
    // OlapColumnDataConvertorChar::clone_and_padding), so the per-row length
1986
    // recorded in dict word info / page headers is always the padded length
1987
    // (e.g. 25 for CHAR(25)) — never the logical length expected by length().
1988
    // Recovering the logical length requires scanning the chars buffer with
1989
    // strnlen() (shrink_padding_chars), which OFFSET_ONLY by definition skips.
1990
    // There is no partial-benefit path: any optimization that still produces
1991
    // the correct length() result must read the chars buffer in full.
1992
    //
1993
    // FE (NestedColumnPruning) already filters CHAR slots out of the
1994
    // OFFSET-only access plan, so reaching this branch means an FE/BE
1995
    // contract violation. Fail loudly instead of silently falling back.
1996
1.03k
    if (read_offset_only() && get_reader() != nullptr &&
1997
1.03k
        get_reader()->get_meta_type() == FieldType::OLAP_FIELD_TYPE_CHAR) {
1998
0
        return Status::InternalError(
1999
0
                "OFFSET_ONLY access path is not supported on CHAR column '{}': CHAR is stored "
2000
0
                "padded so the per-row length information available without reading the chars "
2001
0
                "buffer is always the padded length, not the logical length. The FE planner "
2002
0
                "must not emit an OFFSET access path for CHAR columns.",
2003
0
                _column_name);
2004
0
    }
2005
1.03k
    if (read_offset_only()) {
2006
121
        DLOG(INFO) << "String column iterator set column " << _column_name
2007
121
                   << " to OFFSET_ONLY reading mode";
2008
914
    } else if (read_null_map_only()) {
2009
206
        DLOG(INFO) << "String column iterator set column " << _column_name
2010
206
                   << " to NULL_MAP_ONLY reading mode";
2011
206
    }
2012
2013
1.03k
    return Status::OK();
2014
1.03k
}
2015
2016
////////////////////////////////////////////////////////////////////////////////
2017
2018
27.1M
FileColumnIterator::FileColumnIterator(std::shared_ptr<ColumnReader> reader) : _reader(reader) {}
2019
2020
55.7k
void ColumnIterator::_check_and_set_meta_read_mode(const TColumnAccessPaths& sub_all_access_paths) {
2021
55.7k
    for (const auto& path : sub_all_access_paths) {
2022
6.49k
        if (!path.data_access_path.path.empty()) {
2023
6.49k
            if (StringCaseEqual()(path.data_access_path.path[0], ACCESS_OFFSET)) {
2024
1.74k
                _read_mode = ReadMode::OFFSET_ONLY;
2025
1.74k
                return;
2026
4.75k
            } else if (StringCaseEqual()(path.data_access_path.path[0], ACCESS_NULL)) {
2027
234
                _read_mode = ReadMode::NULL_MAP_ONLY;
2028
234
                return;
2029
234
            }
2030
6.49k
        }
2031
6.48k
    }
2032
53.7k
    _read_mode = ReadMode::DEFAULT;
2033
53.7k
}
2034
2035
26.9M
Status FileColumnIterator::init(const ColumnIteratorOptions& opts) {
2036
26.9M
    if (_reading_flag == ReadingFlag::SKIP_READING) {
2037
2.37k
        DLOG(INFO) << "File column iterator column " << _column_name << " skip reading.";
2038
2.37k
        return Status::OK();
2039
2.37k
    }
2040
2041
26.9M
    _opts = opts;
2042
26.9M
    if (!_opts.use_page_cache) {
2043
26.1M
        _reader->disable_index_meta_cache();
2044
26.1M
    }
2045
26.9M
    RETURN_IF_ERROR(get_block_compression_codec(_reader->get_compression(), &_compress_codec));
2046
26.9M
    if (config::enable_low_cardinality_optimize &&
2047
27.0M
        opts.io_ctx.reader_type == ReaderType::READER_QUERY &&
2048
26.9M
        _reader->encoding_info()->encoding() == DICT_ENCODING) {
2049
16.1M
        auto dict_encoding_type = _reader->get_dict_encoding_type();
2050
        // Only if the column is a predicate column, then we need check the all dict encoding flag
2051
        // because we could rewrite the predciate to accelarate query speed. But if it is not a
2052
        // predicate column, then it is useless. And it has a bad impact on cold read(first time read)
2053
        // because it will load the column's ordinal index and zonemap index and maybe other indices.
2054
        // it has bad impact on primary key query. For example, select * from table where pk = 1, and
2055
        // the table has 2000 columns.
2056
16.1M
        if (dict_encoding_type == ColumnReader::UNKNOWN_DICT_ENCODING && opts.is_predicate_column) {
2057
15.0k
            RETURN_IF_ERROR(seek_to_ordinal(_reader->num_rows() - 1));
2058
15.0k
            _is_all_dict_encoding = _page.is_dict_encoding;
2059
15.0k
            _reader->set_dict_encoding_type(_is_all_dict_encoding
2060
15.0k
                                                    ? ColumnReader::ALL_DICT_ENCODING
2061
15.0k
                                                    : ColumnReader::PARTIAL_DICT_ENCODING);
2062
16.1M
        } else {
2063
16.1M
            _is_all_dict_encoding = dict_encoding_type == ColumnReader::ALL_DICT_ENCODING;
2064
16.1M
        }
2065
16.1M
    }
2066
26.9M
    return Status::OK();
2067
26.9M
}
2068
2069
27.1M
FileColumnIterator::~FileColumnIterator() = default;
2070
2071
416k
void FileColumnIterator::_trigger_prefetch_if_eligible(ordinal_t ord) {
2072
416k
    std::vector<BlockRange> ranges;
2073
416k
    if (_prefetcher->need_prefetch(cast_set<uint32_t>(ord), &ranges)) {
2074
300k
        for (const auto& range : ranges) {
2075
300k
            _cached_remote_file_reader->prefetch_range(range.offset, range.size, &_opts.io_ctx);
2076
300k
        }
2077
299k
    }
2078
416k
}
2079
2080
3.75M
Status FileColumnIterator::seek_to_ordinal(ordinal_t ord) {
2081
3.75M
    if (_reading_flag == ReadingFlag::SKIP_READING) {
2082
449
        DLOG(INFO) << "File column iterator column " << _column_name << " skip reading.";
2083
449
        return Status::OK();
2084
449
    }
2085
2086
18.4E
    LOG_IF(INFO, config::enable_segment_prefetch_verbose_log) << fmt::format(
2087
18.4E
            "[verbose] FileColumnIterator::seek_to_ordinal seek to ordinal {}, enable_prefetch={}",
2088
18.4E
            ord, _enable_prefetch);
2089
3.75M
    if (_enable_prefetch) {
2090
416k
        _trigger_prefetch_if_eligible(ord);
2091
416k
    }
2092
2093
    // if current page contains this row, we don't need to seek
2094
3.75M
    if (!_page || !_page.contains(ord) || !_page_iter.valid()) {
2095
1.59M
        RETURN_IF_ERROR(_reader->seek_at_or_before(ord, &_page_iter, _opts));
2096
1.59M
        RETURN_IF_ERROR(_read_data_page(_page_iter));
2097
1.59M
    }
2098
3.75M
    RETURN_IF_ERROR(_seek_to_pos_in_page(&_page, ord - _page.first_ordinal));
2099
3.75M
    _current_ordinal = ord;
2100
3.75M
    return Status::OK();
2101
3.75M
}
2102
2103
0
Status FileColumnIterator::seek_to_page_start() {
2104
0
    return seek_to_ordinal(_page.first_ordinal);
2105
0
}
2106
2107
3.81M
Status FileColumnIterator::_seek_to_pos_in_page(ParsedPage* page, ordinal_t offset_in_page) const {
2108
3.81M
    if (page->offset_in_page == offset_in_page) {
2109
        // fast path, do nothing
2110
1.89M
        return Status::OK();
2111
1.89M
    }
2112
2113
1.92M
    ordinal_t pos_in_data = offset_in_page;
2114
1.92M
    if (_page.has_null) {
2115
120k
        ordinal_t offset_in_data = 0;
2116
120k
        ordinal_t skips = offset_in_page;
2117
2118
120k
        if (offset_in_page > page->offset_in_page) {
2119
            // forward, reuse null bitmap
2120
80.8k
            skips = offset_in_page - page->offset_in_page;
2121
80.8k
            offset_in_data = page->data_decoder->current_index();
2122
80.8k
        } else {
2123
            // rewind null bitmap, and
2124
39.4k
            page->null_decoder = RleDecoder<bool>((const uint8_t*)page->null_bitmap.data,
2125
39.4k
                                                  cast_set<int>(page->null_bitmap.size), 1);
2126
39.4k
        }
2127
2128
120k
        auto skip_nulls = page->null_decoder.Skip(skips);
2129
120k
        pos_in_data = offset_in_data + skips - skip_nulls;
2130
120k
    }
2131
2132
1.92M
    RETURN_IF_ERROR(page->data_decoder->seek_to_position_in_page(pos_in_data));
2133
1.92M
    page->offset_in_page = offset_in_page;
2134
1.92M
    return Status::OK();
2135
1.92M
}
2136
2137
18.3k
Status FileColumnIterator::next_batch_of_zone_map(size_t* n, MutableColumnPtr& dst) {
2138
18.3k
    return _reader->next_batch_of_zone_map(n, dst);
2139
18.3k
}
2140
2141
2.90M
Status FileColumnIterator::next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) {
2142
2.90M
    if (read_null_map_only()) {
2143
2.95k
        DLOG(INFO) << "File column iterator column " << _column_name
2144
2.95k
                   << " in NULL_MAP_ONLY mode, reading only null map.";
2145
2.95k
        DORIS_CHECK(dst->is_nullable());
2146
2.95k
        auto& nullable_col = assert_cast<ColumnNullable&>(*dst);
2147
2.95k
        auto& null_map_data = nullable_col.get_null_map_data();
2148
2149
2.95k
        size_t remaining = *n;
2150
2.95k
        *has_null = false;
2151
5.96k
        while (remaining > 0) {
2152
3.00k
            if (!_page.has_remaining()) {
2153
51
                bool eos = false;
2154
51
                RETURN_IF_ERROR(_load_next_page(&eos));
2155
51
                if (eos) {
2156
0
                    break;
2157
0
                }
2158
51
            }
2159
2160
3.00k
            size_t nrows_in_page = std::min(remaining, _page.remaining());
2161
3.00k
            size_t nrows_to_read = nrows_in_page;
2162
3.00k
            if (_page.has_null) {
2163
211k
                while (nrows_to_read > 0) {
2164
208k
                    bool is_null = false;
2165
208k
                    size_t this_run = _page.null_decoder.GetNextRun(&is_null, nrows_to_read);
2166
208k
                    const size_t cur_size = null_map_data.size();
2167
208k
                    null_map_data.resize(cur_size + this_run);
2168
208k
                    memset(null_map_data.data() + cur_size, is_null ? 1 : 0, this_run);
2169
208k
                    if (is_null) {
2170
105k
                        *has_null = true;
2171
105k
                    }
2172
208k
                    nrows_to_read -= this_run;
2173
208k
                    _page.offset_in_page += this_run;
2174
208k
                    _current_ordinal += this_run;
2175
208k
                }
2176
2.93k
            } else {
2177
75
                const size_t cur_size = null_map_data.size();
2178
75
                null_map_data.resize(cur_size + nrows_to_read);
2179
75
                memset(null_map_data.data() + cur_size, 0, nrows_to_read);
2180
75
                _page.offset_in_page += nrows_to_read;
2181
75
                _current_ordinal += nrows_to_read;
2182
75
            }
2183
3.00k
            remaining -= nrows_in_page;
2184
3.00k
        }
2185
2.95k
        *n -= remaining;
2186
2.95k
        nullable_col.get_nested_column().insert_many_defaults(*n);
2187
2.95k
        return Status::OK();
2188
2.95k
    }
2189
2190
2.90M
    if (_reading_flag == ReadingFlag::SKIP_READING) {
2191
449
        DLOG(INFO) << "File column iterator column " << _column_name << " skip reading.";
2192
449
        dst->insert_many_defaults(*n);
2193
449
        return Status::OK();
2194
449
    }
2195
2196
2.90M
    size_t curr_size = dst->byte_size();
2197
2.90M
    dst->reserve(*n);
2198
2.90M
    size_t remaining = *n;
2199
2.90M
    *has_null = false;
2200
5.87M
    while (remaining > 0) {
2201
2.96M
        if (!_page.has_remaining()) {
2202
62.9k
            bool eos = false;
2203
62.9k
            RETURN_IF_ERROR(_load_next_page(&eos));
2204
62.9k
            if (eos) {
2205
0
                break;
2206
0
            }
2207
62.9k
        }
2208
2209
        // number of rows to be read from this page
2210
2.96M
        size_t nrows_in_page = std::min(remaining, _page.remaining());
2211
2.96M
        size_t nrows_to_read = nrows_in_page;
2212
2.96M
        if (_page.has_null) {
2213
904k
            while (nrows_to_read > 0) {
2214
701k
                bool is_null = false;
2215
701k
                size_t this_run = _page.null_decoder.GetNextRun(&is_null, nrows_to_read);
2216
                // we use num_rows only for CHECK
2217
701k
                size_t num_rows = this_run;
2218
701k
                if (!is_null) {
2219
371k
                    RETURN_IF_ERROR(_page.data_decoder->next_batch(&num_rows, dst));
2220
371k
                    DCHECK_EQ(this_run, num_rows);
2221
371k
                } else {
2222
329k
                    *has_null = true;
2223
329k
                    auto* null_col = check_and_get_column<ColumnNullable>(dst.get());
2224
330k
                    if (null_col != nullptr) {
2225
330k
                        null_col->insert_many_defaults(this_run);
2226
18.4E
                    } else {
2227
18.4E
                        return Status::InternalError("unexpected column type in column reader");
2228
18.4E
                    }
2229
329k
                }
2230
2231
702k
                nrows_to_read -= this_run;
2232
702k
                _page.offset_in_page += this_run;
2233
702k
                _current_ordinal += this_run;
2234
702k
            }
2235
2.76M
        } else {
2236
2.76M
            RETURN_IF_ERROR(_page.data_decoder->next_batch(&nrows_to_read, dst));
2237
2.76M
            DCHECK_EQ(nrows_to_read, nrows_in_page);
2238
2239
2.76M
            _page.offset_in_page += nrows_to_read;
2240
2.76M
            _current_ordinal += nrows_to_read;
2241
2.76M
        }
2242
2.96M
        remaining -= nrows_in_page;
2243
2.96M
    }
2244
2.90M
    *n -= remaining;
2245
2.90M
    _opts.stats->bytes_read += (dst->byte_size() - curr_size) + BitmapSize(*n);
2246
2247
#ifdef BE_TEST
2248
    _reader->check_data_by_zone_map_for_test(dst);
2249
#endif
2250
2.90M
    return Status::OK();
2251
2.90M
}
2252
2253
Status FileColumnIterator::read_by_rowids(const rowid_t* rowids, const size_t count,
2254
784k
                                          MutableColumnPtr& dst) {
2255
784k
    if (read_null_map_only()) {
2256
15
        DLOG(INFO) << "File column iterator column " << _column_name
2257
15
                   << " in NULL_MAP_ONLY mode, reading only null map by rowids.";
2258
2259
15
        DORIS_CHECK(dst->is_nullable());
2260
15
        auto& nullable_col = assert_cast<ColumnNullable&>(*dst);
2261
15
        auto& null_map_data = nullable_col.get_null_map_data();
2262
15
        const size_t base_size = null_map_data.size();
2263
15
        null_map_data.resize(base_size + count);
2264
2265
15
        size_t remaining = count;
2266
15
        size_t total_read_count = 0;
2267
15
        size_t nrows_to_read = 0;
2268
30
        while (remaining > 0) {
2269
15
            RETURN_IF_ERROR(seek_to_ordinal(rowids[total_read_count]));
2270
2271
15
            nrows_to_read = std::min(remaining, _page.remaining());
2272
2273
15
            if (_page.has_null) {
2274
2
                size_t already_read = 0;
2275
4
                while ((nrows_to_read - already_read) > 0) {
2276
2
                    bool is_null = false;
2277
2
                    size_t this_run = std::min(nrows_to_read - already_read, _page.remaining());
2278
2
                    if (UNLIKELY(this_run == 0)) {
2279
0
                        break;
2280
0
                    }
2281
2
                    this_run = _page.null_decoder.GetNextRun(&is_null, this_run);
2282
2283
2
                    size_t offset = total_read_count + already_read;
2284
2
                    size_t this_read_count = 0;
2285
2
                    rowid_t current_ordinal_in_page =
2286
2
                            cast_set<uint32_t>(_page.offset_in_page + _page.first_ordinal);
2287
4
                    for (size_t i = 0; i < this_run; ++i) {
2288
2
                        if (rowids[offset + i] - current_ordinal_in_page >= this_run) {
2289
0
                            break;
2290
0
                        }
2291
2
                        this_read_count++;
2292
2
                    }
2293
2294
2
                    if (this_read_count > 0) {
2295
2
                        memset(null_map_data.data() + base_size + offset, is_null ? 1 : 0,
2296
2
                               this_read_count);
2297
2
                    }
2298
2299
2
                    already_read += this_read_count;
2300
2
                    _page.offset_in_page += this_run;
2301
2
                }
2302
2303
2
                nrows_to_read = already_read;
2304
2
                total_read_count += nrows_to_read;
2305
2
                remaining -= nrows_to_read;
2306
13
            } else {
2307
13
                memset(null_map_data.data() + base_size + total_read_count, 0, nrows_to_read);
2308
13
                total_read_count += nrows_to_read;
2309
13
                remaining -= nrows_to_read;
2310
13
            }
2311
15
        }
2312
2313
15
        null_map_data.resize(base_size + total_read_count);
2314
15
        nullable_col.get_nested_column().insert_many_defaults(total_read_count);
2315
15
        return Status::OK();
2316
15
    }
2317
2318
784k
    if (_reading_flag == ReadingFlag::SKIP_READING) {
2319
0
        DLOG(INFO) << "File column iterator column " << _column_name << " skip reading.";
2320
0
        dst->insert_many_defaults(count);
2321
0
        return Status::OK();
2322
0
    }
2323
2324
784k
    size_t remaining = count;
2325
784k
    size_t total_read_count = 0;
2326
784k
    size_t nrows_to_read = 0;
2327
1.59M
    while (remaining > 0) {
2328
813k
        RETURN_IF_ERROR(seek_to_ordinal(rowids[total_read_count]));
2329
2330
        // number of rows to be read from this page
2331
813k
        nrows_to_read = std::min(remaining, _page.remaining());
2332
2333
813k
        if (_page.has_null) {
2334
68.7k
            size_t already_read = 0;
2335
4.37M
            while ((nrows_to_read - already_read) > 0) {
2336
4.30M
                bool is_null = false;
2337
4.30M
                size_t this_run = std::min(nrows_to_read - already_read, _page.remaining());
2338
4.30M
                if (UNLIKELY(this_run == 0)) {
2339
183
                    break;
2340
183
                }
2341
4.30M
                this_run = _page.null_decoder.GetNextRun(&is_null, this_run);
2342
4.30M
                size_t offset = total_read_count + already_read;
2343
4.30M
                size_t this_read_count = 0;
2344
4.30M
                rowid_t current_ordinal_in_page =
2345
4.30M
                        cast_set<uint32_t>(_page.offset_in_page + _page.first_ordinal);
2346
20.8M
                for (size_t i = 0; i < this_run; ++i) {
2347
19.5M
                    if (rowids[offset + i] - current_ordinal_in_page >= this_run) {
2348
3.05M
                        break;
2349
3.05M
                    }
2350
16.5M
                    this_read_count++;
2351
16.5M
                }
2352
2353
4.30M
                auto origin_index = _page.data_decoder->current_index();
2354
4.30M
                if (this_read_count > 0) {
2355
1.25M
                    if (is_null) {
2356
657k
                        auto* null_col = check_and_get_column<ColumnNullable>(dst.get());
2357
657k
                        if (UNLIKELY(null_col == nullptr)) {
2358
0
                            return Status::InternalError("unexpected column type in column reader");
2359
0
                        }
2360
2361
657k
                        null_col->insert_many_defaults(this_read_count);
2362
657k
                    } else {
2363
600k
                        size_t read_count = this_read_count;
2364
2365
                        // ordinal in nullable columns' data buffer maybe be not continuously(the data doesn't contain null value),
2366
                        // so we need use `page_start_off_in_decoder` to calculate the actual offset in `data_decoder`
2367
600k
                        size_t page_start_off_in_decoder =
2368
600k
                                _page.first_ordinal + _page.offset_in_page - origin_index;
2369
600k
                        RETURN_IF_ERROR(_page.data_decoder->read_by_rowids(
2370
600k
                                &rowids[offset], page_start_off_in_decoder, &read_count, dst));
2371
600k
                        DCHECK_EQ(read_count, this_read_count);
2372
600k
                    }
2373
1.25M
                }
2374
2375
4.30M
                if (!is_null) {
2376
3.63M
                    RETURN_IF_ERROR(
2377
3.63M
                            _page.data_decoder->seek_to_position_in_page(origin_index + this_run));
2378
3.63M
                }
2379
2380
4.30M
                already_read += this_read_count;
2381
4.30M
                _page.offset_in_page += this_run;
2382
4.30M
                DCHECK(_page.offset_in_page <= _page.num_rows);
2383
4.30M
            }
2384
2385
68.7k
            nrows_to_read = already_read;
2386
68.7k
            total_read_count += nrows_to_read;
2387
68.7k
            remaining -= nrows_to_read;
2388
744k
        } else {
2389
744k
            RETURN_IF_ERROR(_page.data_decoder->read_by_rowids(
2390
744k
                    &rowids[total_read_count], _page.first_ordinal, &nrows_to_read, dst));
2391
744k
            total_read_count += nrows_to_read;
2392
744k
            remaining -= nrows_to_read;
2393
744k
        }
2394
813k
    }
2395
784k
    return Status::OK();
2396
784k
}
2397
2398
62.9k
Status FileColumnIterator::_load_next_page(bool* eos) {
2399
62.9k
    _page_iter.next();
2400
62.9k
    if (!_page_iter.valid()) {
2401
0
        *eos = true;
2402
0
        return Status::OK();
2403
0
    }
2404
2405
62.9k
    RETURN_IF_ERROR(_read_data_page(_page_iter));
2406
62.9k
    RETURN_IF_ERROR(_seek_to_pos_in_page(&_page, 0));
2407
62.9k
    *eos = false;
2408
62.9k
    return Status::OK();
2409
62.9k
}
2410
2411
1.65M
Status FileColumnIterator::_read_data_page(const OrdinalPageIndexIterator& iter) {
2412
1.65M
    PageHandle handle;
2413
1.65M
    Slice page_body;
2414
1.65M
    PageFooterPB footer;
2415
1.65M
    _opts.type = DATA_PAGE;
2416
1.65M
    PageDecoderOptions decoder_opts;
2417
1.65M
    decoder_opts.only_read_offsets = _opts.only_read_offsets;
2418
1.65M
    RETURN_IF_ERROR(
2419
1.65M
            _reader->read_page(_opts, iter.page(), &handle, &page_body, &footer, _compress_codec));
2420
    // parse data page
2421
1.65M
    auto st = ParsedPage::create(std::move(handle), page_body, footer.data_page_footer(),
2422
1.65M
                                 _reader->encoding_info(), iter.page(), iter.page_index(), &_page,
2423
1.65M
                                 decoder_opts);
2424
1.65M
    if (!st.ok()) {
2425
0
        LOG(WARNING) << "failed to create ParsedPage, file=" << _opts.file_reader->path().native()
2426
0
                     << ", page_offset=" << iter.page().offset << ", page_size=" << iter.page().size
2427
0
                     << ", page_index=" << iter.page_index() << ", error=" << st;
2428
0
        return st;
2429
0
    }
2430
2431
    // dictionary page is read when the first data page that uses it is read,
2432
    // this is to optimize the memory usage: when there is no query on one column, we could
2433
    // release the memory of dictionary page.
2434
    // note that concurrent iterators for the same column won't repeatedly read dictionary page
2435
    // because of page cache.
2436
1.65M
    if (_reader->encoding_info()->encoding() == DICT_ENCODING) {
2437
457k
        auto dict_page_decoder = reinterpret_cast<BinaryDictPageDecoder*>(_page.data_decoder.get());
2438
457k
        if (dict_page_decoder->is_dict_encoding()) {
2439
423k
            if (_dict_decoder == nullptr) {
2440
411k
                RETURN_IF_ERROR(_read_dict_data());
2441
411k
                CHECK_NOTNULL(_dict_decoder);
2442
411k
            }
2443
2444
423k
            dict_page_decoder->set_dict_decoder(cast_set<uint32_t>(_dict_decoder->count()),
2445
423k
                                                _dict_word_info.get());
2446
423k
        }
2447
457k
    }
2448
1.65M
    return Status::OK();
2449
1.65M
}
2450
2451
412k
Status FileColumnIterator::_read_dict_data() {
2452
412k
    CHECK_EQ(_reader->encoding_info()->encoding(), DICT_ENCODING);
2453
    // read dictionary page
2454
412k
    Slice dict_data;
2455
412k
    PageFooterPB dict_footer;
2456
412k
    _opts.type = INDEX_PAGE;
2457
2458
412k
    RETURN_IF_ERROR(_reader->read_page(_opts, _reader->get_dict_page_pointer(), &_dict_page_handle,
2459
412k
                                       &dict_data, &dict_footer, _compress_codec));
2460
412k
    const EncodingInfo* encoding_info;
2461
    // The dict pool stores strings of the outer column's type. Using the
2462
    // outer type (CHAR vs VARCHAR/STRING) lets the EncodingInfo pick a
2463
    // CHAR-strip pre-decoder so the cached dict page is already unpadded.
2464
412k
    RETURN_IF_ERROR(EncodingInfo::get(_reader->get_meta_type(),
2465
412k
                                      dict_footer.dict_page_footer().encoding(), &encoding_info));
2466
412k
    RETURN_IF_ERROR(encoding_info->create_page_decoder(dict_data, {}, _dict_decoder));
2467
412k
    RETURN_IF_ERROR(_dict_decoder->init());
2468
2469
412k
    _dict_word_info.reset(new StringRef[_dict_decoder->count()]);
2470
412k
    RETURN_IF_ERROR(_dict_decoder->get_dict_word_info(_dict_word_info.get()));
2471
412k
    return Status::OK();
2472
412k
}
2473
2474
Status FileColumnIterator::get_row_ranges_by_zone_map(
2475
        const AndBlockColumnPredicate* col_predicates,
2476
        const std::vector<std::shared_ptr<const ColumnPredicate>>* delete_predicates,
2477
135k
        RowRanges* row_ranges) {
2478
135k
    if (_reader->has_zone_map()) {
2479
134k
        RETURN_IF_ERROR(_reader->get_row_ranges_by_zone_map(col_predicates, delete_predicates,
2480
134k
                                                            row_ranges, _opts));
2481
134k
    }
2482
135k
    return Status::OK();
2483
135k
}
2484
2485
Status FileColumnIterator::get_row_ranges_by_bloom_filter(
2486
135k
        const AndBlockColumnPredicate* col_predicates, RowRanges* row_ranges) {
2487
135k
    if ((col_predicates->can_do_bloom_filter(false) && _reader->has_bloom_filter_index(false)) ||
2488
135k
        (col_predicates->can_do_bloom_filter(true) && _reader->has_bloom_filter_index(true))) {
2489
110
        RETURN_IF_ERROR(_reader->get_row_ranges_by_bloom_filter(col_predicates, row_ranges, _opts));
2490
110
    }
2491
135k
    return Status::OK();
2492
135k
}
2493
2494
Status FileColumnIterator::get_row_ranges_by_dict(const AndBlockColumnPredicate* col_predicates,
2495
137k
                                                  RowRanges* row_ranges) {
2496
137k
    if (!_is_all_dict_encoding) {
2497
126k
        return Status::OK();
2498
126k
    }
2499
2500
10.6k
    if (!_dict_decoder) {
2501
0
        RETURN_IF_ERROR(_read_dict_data());
2502
0
        CHECK_NOTNULL(_dict_decoder);
2503
0
    }
2504
2505
10.6k
    if (!col_predicates->evaluate_and(_dict_word_info.get(), _dict_decoder->count())) {
2506
1.14k
        row_ranges->clear();
2507
1.14k
    }
2508
10.6k
    return Status::OK();
2509
10.6k
}
2510
2511
299k
Status FileColumnIterator::init_prefetcher(const SegmentPrefetchParams& params) {
2512
299k
    if (_cached_remote_file_reader =
2513
299k
                std::dynamic_pointer_cast<io::CachedRemoteFileReader>(_reader->_file_reader);
2514
299k
        !_cached_remote_file_reader) {
2515
0
        return Status::OK();
2516
0
    }
2517
299k
    _enable_prefetch = true;
2518
299k
    _prefetcher = std::make_unique<SegmentPrefetcher>(params.config);
2519
299k
    RETURN_IF_ERROR(_prefetcher->init(_reader, params.read_options));
2520
299k
    return Status::OK();
2521
299k
}
2522
2523
void FileColumnIterator::collect_prefetchers(
2524
        std::map<PrefetcherInitMethod, std::vector<SegmentPrefetcher*>>& prefetchers,
2525
299k
        PrefetcherInitMethod init_method) {
2526
299k
    if (_prefetcher) {
2527
299k
        prefetchers[init_method].emplace_back(_prefetcher.get());
2528
299k
    }
2529
299k
}
2530
2531
21.7k
Status DefaultValueColumnIterator::init(const ColumnIteratorOptions& opts) {
2532
21.7k
    _opts = opts;
2533
    // be consistent with segment v1
2534
    // if _has_default_value, we should create default column iterator for this column, and
2535
    // "NULL" is a special default value which means the default value is null.
2536
21.7k
    if (_has_default_value) {
2537
1.80k
        if (_default_value == "NULL") {
2538
1.03k
            _default_value_field = Field::create_field<TYPE_NULL>(Null {});
2539
1.03k
        } else {
2540
773
            if (_type == FieldType::OLAP_FIELD_TYPE_ARRAY) {
2541
8
                if (_default_value != "[]") {
2542
0
                    return Status::NotSupported("Array default {} is unsupported", _default_value);
2543
8
                } else {
2544
8
                    _default_value_field = Field::create_field<TYPE_ARRAY>(Array {});
2545
8
                    return Status::OK();
2546
8
                }
2547
765
            } else if (_type == FieldType::OLAP_FIELD_TYPE_STRUCT) {
2548
0
                return Status::NotSupported("STRUCT default type is unsupported");
2549
765
            } else if (_type == FieldType::OLAP_FIELD_TYPE_MAP) {
2550
0
                return Status::NotSupported("MAP default type is unsupported");
2551
0
            }
2552
765
            const auto t = _type;
2553
765
            const auto serde = DataTypeFactory::instance()
2554
765
                                       .create_data_type(t, _precision, _scale, _len)
2555
765
                                       ->get_serde();
2556
765
            RETURN_IF_ERROR(serde->from_fe_string(_default_value, _default_value_field));
2557
765
        }
2558
19.9k
    } else if (_is_nullable) {
2559
19.9k
        _default_value_field = Field::create_field<TYPE_NULL>(Null {});
2560
18.4E
    } else {
2561
18.4E
        return Status::InternalError(
2562
18.4E
                "invalid default value column for no default value and not nullable");
2563
18.4E
    }
2564
21.7k
    return Status::OK();
2565
21.7k
}
2566
2567
2.31k
Status DefaultValueColumnIterator::next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) {
2568
2.31k
    *has_null = _default_value_field.is_null();
2569
2.31k
    _insert_many_default(dst, *n);
2570
2.31k
    return Status::OK();
2571
2.31k
}
2572
2573
Status DefaultValueColumnIterator::read_by_rowids(const rowid_t* rowids, const size_t count,
2574
6.21k
                                                  MutableColumnPtr& dst) {
2575
6.21k
    _insert_many_default(dst, count);
2576
6.21k
    return Status::OK();
2577
6.21k
}
2578
2579
8.52k
void DefaultValueColumnIterator::_insert_many_default(MutableColumnPtr& dst, size_t n) {
2580
8.52k
    if (_default_value_field.is_null()) {
2581
8.10k
        dst->insert_many_defaults(n);
2582
8.10k
    } else {
2583
426
        dst = dst->convert_to_predicate_column_if_dictionary();
2584
426
        dst->insert_duplicate_fields(_default_value_field, n);
2585
426
    }
2586
8.52k
}
2587
2588
3.54k
Status RowIdColumnIteratorV2::next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) {
2589
3.54k
    auto* string_column = assert_cast<ColumnString*, TypeCheckOnRelease::DISABLE>(dst.get());
2590
2591
10.8M
    for (uint32_t i = 0; i < *n; ++i) {
2592
10.8M
        uint32_t row_id = _current_rowid + i;
2593
10.8M
        GlobalRowLoacationV2 location(_version, _backend_id, _file_id, row_id);
2594
10.8M
        string_column->insert_data(reinterpret_cast<const char*>(&location),
2595
10.8M
                                   sizeof(GlobalRowLoacationV2));
2596
10.8M
    }
2597
3.54k
    _current_rowid += *n;
2598
3.54k
    return Status::OK();
2599
3.54k
}
2600
2601
Status RowIdColumnIteratorV2::read_by_rowids(const rowid_t* rowids, const size_t count,
2602
46.9k
                                             MutableColumnPtr& dst) {
2603
46.9k
    auto* string_column = assert_cast<ColumnString*>(dst.get());
2604
2605
31.2M
    for (size_t i = 0; i < count; ++i) {
2606
31.2M
        uint32_t row_id = rowids[i];
2607
31.2M
        GlobalRowLoacationV2 location(_version, _backend_id, _file_id, row_id);
2608
31.2M
        string_column->insert_data(reinterpret_cast<const char*>(&location),
2609
31.2M
                                   sizeof(GlobalRowLoacationV2));
2610
31.2M
    }
2611
46.9k
    return Status::OK();
2612
46.9k
}
2613
2614
} // namespace doris::segment_v2