// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

#include "storage/segment/segment_iterator.h"

#include <gen_cpp/Exprs_types.h>

#include <gen_cpp/Opcodes_types.h>

#include <gen_cpp/Types_types.h>

#include <gen_cpp/olap_file.pb.h>

#include <glog/logging.h>

#include <algorithm>

#include <boost/iterator/iterator_facade.hpp>

#include <cassert>

#include <cstdint>

#include <memory>

#include <numeric>

#include <optional>

#include <set>

#include <unordered_map>

#include <utility>

#include <vector>

#include "cloud/config.h"

#include "common/compiler_util.h" // IWYU pragma: keep

#include "common/config.h"

#include "common/consts.h"

#include "common/exception.h"

#include "common/logging.h"

#include "common/metrics/doris_metrics.h"

#include "common/object_pool.h"

#include "common/status.h"

#include "core/assert_cast.h"

#include "core/block/column_with_type_and_name.h"

#include "core/column/column.h"

#include "core/column/column_const.h"

#include "core/column/column_nothing.h"

#include "core/column/column_nullable.h"

#include "core/column/column_string.h"

#include "core/column/column_variant.h"

#include "core/column/column_vector.h"

#include "core/data_type/data_type.h"

#include "core/data_type/data_type_factory.hpp"

#include "core/data_type/data_type_number.h"

#include "core/data_type/define_primitive_type.h"

#include "core/field.h"

#include "core/string_ref.h"

#include "core/typeid_cast.h"

#include "core/types.h"

#include "exprs/expr_zonemap_filter.h"

#include "exprs/function/array/function_array_index.h"

#include "exprs/runtime_filter_expr.h"

#include "exprs/vexpr.h"

#include "exprs/vexpr_context.h"

#include "exprs/virtual_slot_ref.h"

#include "exprs/vliteral.h"

#include "exprs/vslot_ref.h"

#include "io/cache/cached_remote_file_reader.h"

#include "io/fs/file_reader.h"

#include "io/io_common.h"

#include "runtime/query_context.h"

#include "runtime/runtime_predicate.h"

#include "runtime/runtime_state.h"

#include "runtime/thread_context.h"

#include "storage/binlog.h"

#include "storage/compaction/collection_similarity.h"

#include "storage/id_manager.h"

#include "storage/index/ann/ann_index.h"

#include "storage/index/ann/ann_index_iterator.h"

#include "storage/index/ann/ann_index_reader.h"

#include "storage/index/ann/ann_topn_runtime.h"

#include "storage/index/index_file_reader.h"

#include "storage/index/index_iterator.h"

#include "storage/index/index_query_context.h"

#include "storage/index/index_reader_helper.h"

#include "storage/index/indexed_column_reader.h"

#include "storage/index/inverted/inverted_index_reader.h"

#include "storage/index/ordinal_page_index.h"

#include "storage/index/primary_key_index.h"

#include "storage/index/short_key_index.h"

#include "storage/index/zone_map/zone_map_index.h"

#include "storage/index/zone_map/zonemap_eval_context.h"

#include "storage/iterators.h"

#include "storage/olap_common.h"

#include "storage/predicate/bloom_filter_predicate.h"

#include "storage/predicate/column_predicate.h"

#include "storage/predicate/like_column_predicate.h"

#include "storage/schema.h"

#include "storage/segment/column_reader.h"

#include "storage/segment/column_reader_cache.h"

#include "storage/segment/condition_cache.h"

#include "storage/segment/row_ranges.h"

#include "storage/segment/segment.h"

#include "storage/segment/segment_prefetcher.h"

#include "storage/segment/variant/variant_column_reader.h"

#include "storage/segment/virtual_column_iterator.h"

#include "storage/tablet/tablet_schema.h"

#include "storage/types.h"

#include "storage/utils.h"

#include "util/concurrency_stats.h"

#include "util/defer_op.h"

#include "util/simd/bits.h"

namespace doris {

using namespace ErrorCode;

namespace segment_v2 {

class ScopedColumnIteratorReadPhase {

public:

    ScopedColumnIteratorReadPhase(ColumnIterator* column_iter, ColumnIterator::ReadPhase mode)

            : _column_iter(column_iter) {

        DORIS_CHECK(_column_iter != nullptr);

        _column_iter->set_read_phase(mode);

    }

    ScopedColumnIteratorReadPhase(const ScopedColumnIteratorReadPhase&) = delete;

    ScopedColumnIteratorReadPhase& operator=(const ScopedColumnIteratorReadPhase&) = delete;

    ~ScopedColumnIteratorReadPhase() {

        // ReadPhase is a per-read phase knob. SegmentIterator only needs a

        // temporary PREDICATE/LAZY mode while reading one column in one phase; it

        // must be restored before the next column or later normal reads reuse the

        // same ColumnIterator. Keep the restoration in one scoped helper instead

        // of open-coding the same Defer block at every call site.

        _column_iter->set_read_phase(ColumnIterator::ReadPhase::NORMAL);

    }

private:

    ColumnIterator* _column_iter = nullptr;

};

SegmentIterator::~SegmentIterator() = default;

void SegmentIterator::_init_row_bitmap_by_condition_cache() {

    // Only dispose need column predicate and expr cal in condition cache

    if (!_col_predicates.empty() || !_common_expr_ctxs_push_down.empty()) {

        if (_opts.condition_cache_digest) {

            auto* condition_cache = ConditionCache::instance();

            ConditionCache::CacheKey cache_key(_opts.rowset_id, _segment->id(),

                                               _opts.condition_cache_digest);

            // Increment search count when digest != 0

            DorisMetrics::instance()->condition_cache_search_count->increment(1);

            ConditionCacheHandle handle;

            _find_condition_cache = condition_cache->lookup(cache_key, &handle);

            // Increment hit count if cache lookup is successful

            if (_find_condition_cache) {

                DorisMetrics::instance()->condition_cache_hit_count->increment(1);

                if (_opts.runtime_state) {

                    VLOG_DEBUG << "Condition cache hit, query id: "

                               << print_id(_opts.runtime_state->query_id())

                               << ", segment id: " << _segment->id()

                               << ", cache digest: " << _opts.condition_cache_digest

                               << ", rowset id: " << _opts.rowset_id.to_string();

                }

            }

            auto num_rows = _segment->num_rows();

            if (_find_condition_cache) {

                const auto& filter_result = *(handle.get_filter_result());

                int64_t filtered_blocks = 0;

                for (int i = 0; i < filter_result.size(); i++) {

                    if (!filter_result[i]) {

                        _row_bitmap.removeRange(

                                i * CONDITION_CACHE_OFFSET,

                                i * CONDITION_CACHE_OFFSET + CONDITION_CACHE_OFFSET);

                        filtered_blocks++;

                    }

                }

                // Record condition_cache hit segment number

                _opts.stats->condition_cache_hit_seg_nums++;

                // Record rows filtered by condition cache hit

                _opts.stats->condition_cache_filtered_rows +=

                        filtered_blocks * SegmentIterator::CONDITION_CACHE_OFFSET;

            } else {

                _condition_cache = std::make_shared<std::vector<bool>>(

                        num_rows / CONDITION_CACHE_OFFSET + 1, false);

            }

        }

    } else {

        _opts.condition_cache_digest = 0;

    }

}

// A fast range iterator for roaring bitmap. Output ranges use closed-open form, like [from, to).

// Example:

//   input bitmap:  [0 1 4 5 6 7 10 15 16 17 18 19]

//   output ranges: [0,2), [4,8), [10,11), [15,20) (when max_range_size=10)

//   output ranges: [0,2), [4,7), [7,8), [10,11), [15,18), [18,20) (when max_range_size=3)

class SegmentIterator::BitmapRangeIterator {

public:

    BitmapRangeIterator() = default;

    virtual ~BitmapRangeIterator() = default;

    explicit BitmapRangeIterator(const roaring::Roaring& bitmap) {

        roaring_init_iterator(&bitmap.roaring, &_iter);

    }

    bool has_more_range() const { return !_eof; }

    [[nodiscard]] static uint32_t get_batch_size() { return kBatchSize; }

    // read next range into [*from, *to) whose size <= max_range_size.

    // return false when there is no more range.

    virtual bool next_range(const uint32_t max_range_size, uint32_t* from, uint32_t* to) {

        if (_eof) {

            return false;

        }

        *from = _buf[_buf_pos];

        uint32_t range_size = 0;

        uint32_t expect_val = _buf[_buf_pos]; // this initial value just make first batch valid

        // if array is contiguous sequence then the following conditions need to be met :

        // a_0: x

        // a_1: x+1

        // a_2: x+2

        // ...

        // a_p: x+p

        // so we can just use (a_p-a_0)-p to check conditions

        // and should notice the previous batch needs to be continuous with the current batch

        while (!_eof && range_size + _buf_size - _buf_pos <= max_range_size &&

               expect_val == _buf[_buf_pos] &&

               _buf[_buf_size - 1] - _buf[_buf_pos] == _buf_size - 1 - _buf_pos) {

            range_size += _buf_size - _buf_pos;

            expect_val = _buf[_buf_size - 1] + 1;

            _read_next_batch();

        }

        // promise remain range not will reach next batch

        if (!_eof && range_size < max_range_size && expect_val == _buf[_buf_pos]) {

            do {

                _buf_pos++;

                range_size++;

            } while (range_size < max_range_size && _buf[_buf_pos] == _buf[_buf_pos - 1] + 1);

        }

        *to = *from + range_size;

        return true;

    }

    // read batch_size of rowids from roaring bitmap into buf array

    virtual uint32_t read_batch_rowids(rowid_t* buf, uint32_t batch_size) {

        return roaring::api::roaring_read_uint32_iterator(&_iter, buf, batch_size);

    }

private:

    void _read_next_batch() {

        _buf_pos = 0;

        _buf_size = roaring::api::roaring_read_uint32_iterator(&_iter, _buf, kBatchSize);

        _eof = (_buf_size == 0);

    }

    static const uint32_t kBatchSize = 256;

    roaring::api::roaring_uint32_iterator_t _iter;

    uint32_t _buf[kBatchSize];

    uint32_t _buf_pos = 0;

    uint32_t _buf_size = 0;

    bool _eof = false;

};

// A backward range iterator for roaring bitmap. Output ranges use closed-open form, like [from, to).

// Example:

//   input bitmap:  [0 1 4 5 6 7 10 15 16 17 18 19]

//   output ranges: , [15,20), [10,11), [4,8), [0,2) (when max_range_size=10)

//   output ranges: [17,20), [15,17), [10,11), [5,8), [4, 5), [0,2) (when max_range_size=3)

class SegmentIterator::BackwardBitmapRangeIterator : public SegmentIterator::BitmapRangeIterator {

public:

    explicit BackwardBitmapRangeIterator(const roaring::Roaring& bitmap) {

        roaring_init_iterator_last(&bitmap.roaring, &_riter);

        _rowid_count = cast_set<uint32_t>(roaring_bitmap_get_cardinality(&bitmap.roaring));

        _rowid_left = _rowid_count;

    }

    bool has_more_range() const { return !_riter.has_value; }

    // read next range into [*from, *to) whose size <= max_range_size.

    // return false when there is no more range.

    bool next_range(const uint32_t max_range_size, uint32_t* from, uint32_t* to) override {

        if (!_riter.has_value) {

            return false;

        }

        uint32_t range_size = 0;

        *to = _riter.current_value + 1;

        do {

            *from = _riter.current_value;

            range_size++;

            roaring_previous_uint32_iterator(&_riter);

        } while (range_size < max_range_size && _riter.has_value &&

                 _riter.current_value + 1 == *from);

        return true;

    }

    /**

     * Reads a batch of row IDs from a roaring bitmap, starting from the end and moving backwards.

     * This function retrieves the last `batch_size` row IDs from the bitmap and stores them in the provided buffer.

     * It updates the internal state to track how many row IDs are left to read in subsequent calls.

     *

     * The row IDs are read in reverse order, but stored in the buffer maintaining their original order in the bitmap.

     *

     * Example:

     *   input bitmap: [0 1 4 5 6 7 10 15 16 17 18 19]

     *   If the bitmap has 12 elements and batch_size is set to 5, the function will first read [15, 16, 17, 18, 19]

     *   into the buffer, leaving 7 elements left. In the next call with batch_size 5, it will read [4, 5, 6, 7, 10].

     *

     */

    uint32_t read_batch_rowids(rowid_t* buf, uint32_t batch_size) override {

        if (!_riter.has_value || _rowid_left == 0) {

            return 0;

        }

        if (_rowid_count <= batch_size) {

            roaring_bitmap_to_uint32_array(_riter.parent,

                                           buf); // Fill 'buf' with '_rowid_count' elements.

            uint32_t num_read = _rowid_left;     // Save the number of row IDs read.

            _rowid_left = 0;                     // No row IDs left after this operation.

            return num_read;                     // Return the number of row IDs read.

        }

        uint32_t read_size = std::min(batch_size, _rowid_left);

        uint32_t num_read = 0; // Counter for the number of row IDs read.

        // Read row IDs into the buffer in reverse order.

        while (num_read < read_size && _riter.has_value) {

            buf[read_size - num_read - 1] = _riter.current_value;

            num_read++;

            _rowid_left--; // Decrement the count of remaining row IDs.

            roaring_previous_uint32_iterator(&_riter);

        }

        // Return the actual number of row IDs read.

        return num_read;

    }

private:

    roaring::api::roaring_uint32_iterator_t _riter;

    uint32_t _rowid_count;

    uint32_t _rowid_left;

};

SegmentIterator::SegmentIterator(std::shared_ptr<Segment> segment, SchemaSPtr schema)

        : _segment(std::move(segment)),

          _schema(schema),

          _column_iterators(_schema->num_columns()),

          _index_iterators(_schema->num_columns()),

          _cur_rowid(0),

          _lazy_materialization_read(false),

          _lazy_inited(false),

          _inited(false),

          _pool(new ObjectPool) {}

Status SegmentIterator::init(const StorageReadOptions& opts) {

    auto status = _init_impl(opts);

    if (!status.ok()) {

        _segment->update_healthy_status(status);

    }

    return status;

}

std::unique_ptr<AdaptiveBlockSizePredictor> SegmentIterator::_make_block_size_predictor() const {

    if (!config::enable_adaptive_batch_size || _opts.preferred_block_size_bytes == 0) {

        return nullptr;

    }

    // Collect per-column raw byte metadata from the segment footer for the columns

    // this iterator will actually output (defined by _schema, which is built from

    // _opts.return_columns).

    std::vector<AdaptiveBlockSizePredictor::ColumnMetadata> col_metadata;

    uint32_t seg_rows = _segment->num_rows();

    uint64_t total_raw_bytes = 0;

    double metadata_hint_bytes_per_row = 0.0;

    if (seg_rows > 0) {

        const auto& ts = _segment->tablet_schema();

        if (ts) {

            for (ColumnId cid : _schema->column_ids()) {

                if (static_cast<size_t>(cid) < ts->num_columns()) {

                    int32_t uid = ts->column(cid).unique_id();

                    uint64_t raw_bytes = _segment->column_raw_data_bytes(uid);

                    if (uid >= 0 && raw_bytes > 0) {

                        total_raw_bytes += raw_bytes;

                    }

                }

            }

            metadata_hint_bytes_per_row = total_raw_bytes / static_cast<double>(seg_rows);

        }

    }

    return std::make_unique<AdaptiveBlockSizePredictor>(

            _opts.preferred_block_size_bytes, metadata_hint_bytes_per_row,

            AdaptiveBlockSizePredictor::kDefaultProbeRows, _opts.block_row_max);

}

Status SegmentIterator::_init_impl(const StorageReadOptions& opts) {

    // get file handle from file descriptor of segment

    if (_inited) {

        return Status::OK();

    }

    _opts = opts;

    SCOPED_RAW_TIMER(&_opts.stats->segment_iterator_init_timer_ns);

    _inited = true;

    _file_reader = _segment->_file_reader;

    _col_predicates.clear();

    for (const auto& predicate : opts.column_predicates) {

        if (!_segment->can_apply_predicate_safely(predicate->column_id(), *_schema,

                                                  _opts.target_cast_type_for_variants, _opts)) {

            continue;

        }

        _col_predicates.emplace_back(predicate);

    }

    _tablet_id = opts.tablet_id;

    // Read options will not change, so that just resize here

    _block_rowids.resize(_opts.block_row_max);

    // Adaptive batch size: snapshot the initial row limit and create predictor if enabled.

    _initial_block_row_max = _opts.block_row_max;

    _block_size_predictor = _make_block_size_predictor();

    if (_schema->rowid_col_idx() > 0) {

        _record_rowids = true;

    }

    _virtual_column_exprs = _opts.virtual_column_exprs;

    _score_runtime = _opts.score_runtime;

    _ann_topn_runtime = _opts.ann_topn_runtime;

    _enable_prune_nested_column = _opts.io_ctx.reader_type == ReaderType::READER_QUERY &&

                                  _opts.runtime_state &&

                                  _opts.runtime_state->enable_prune_nested_column();

    if (opts.output_columns != nullptr) {

        _output_columns = *(opts.output_columns);

    }

    _storage_name_and_type.resize(_schema->columns().size());

    auto storage_format = _opts.tablet_schema->get_inverted_index_storage_format();

    for (int i = 0; i < _schema->columns().size(); ++i) {

        const TabletColumn* col = _schema->column(i);

        if (col) {

            auto storage_type = _segment->get_data_type_of(*col, _opts);

            if (storage_type == nullptr) {

                storage_type =

                        DataTypeFactory::instance().create_data_type(*col, col->is_nullable());

            }

            // Currently, when writing a lucene index, the field of the document is column_name, and the column name is

            // bound to the index field. Since version 1.2, the data file storage has been changed from column_name to

            // column_unique_id, allowing the column name to be changed. Due to current limitations, previous inverted

            // index data cannot be used after Doris changes the column name. Column names also support Unicode

            // characters, which may cause other problems with indexing in non-ASCII characters.

            // After consideration, it was decided to change the field name from column_name to column_unique_id in

            // format V2, while format V1 continues to use column_name.

            std::string field_name;

            if (storage_format == InvertedIndexStorageFormatPB::V1) {

                field_name = col->name();

            } else {

                if (col->is_extracted_column()) {

                    // variant sub col

                    // field_name format: parent_unique_id.sub_col_name

                    field_name = std::to_string(col->parent_unique_id()) + "." + col->name();

                } else {

                    field_name = std::to_string(col->unique_id());

                }

            }

            _storage_name_and_type[i] = std::make_pair(field_name, storage_type);

            if (int32_t uid =

                        col->is_extracted_column() ? col->parent_unique_id() : col->unique_id();

                !_variant_sparse_column_cache.contains(uid)) {

                DCHECK(uid >= 0);

                _variant_sparse_column_cache.emplace(uid,

                                                     std::make_unique<PathToBinaryColumnCache>());

            }

        }

    }

    RETURN_IF_ERROR(init_iterators());

    RETURN_IF_ERROR(_construct_compound_expr_context());

    VLOG_DEBUG << fmt::format(

            "Segment iterator init, virtual_column_exprs size: {}, common_expr_pushdown size: {}",

            _opts.virtual_column_exprs.size(), _common_expr_ctxs_push_down.size());

    _initialize_predicate_results();

    return Status::OK();

}

void SegmentIterator::_initialize_predicate_results() {

    // Initialize from _col_predicates

    for (auto pred : _col_predicates) {

        int cid = pred->column_id();

        _column_predicate_index_exec_status[cid][pred] = false;

    }

    _calculate_common_expr_index_exec_status();

}

Status SegmentIterator::init_iterators() {

    RETURN_IF_ERROR(_init_return_column_iterators());

    RETURN_IF_ERROR(_init_index_iterators());

    return Status::OK();

}

Status SegmentIterator::_lazy_init(Block* block) {

    if (_lazy_inited) {

        return Status::OK();

    }

    SCOPED_RAW_TIMER(&_opts.stats->block_init_ns);

    DorisMetrics::instance()->segment_read_total->increment(1);

    _row_bitmap.addRange(0, _segment->num_rows());

    _init_row_bitmap_by_condition_cache();

    // z-order can not use prefix index

    if (_segment->_tablet_schema->sort_type() != SortType::ZORDER &&

        _segment->_tablet_schema->cluster_key_uids().empty()) {

        RETURN_IF_ERROR(_get_row_ranges_by_keys());

    }

    RETURN_IF_ERROR(_get_row_ranges_by_column_conditions());

    RETURN_IF_ERROR(_vec_init_lazy_materialization());

    // Remove rows that have been marked deleted

    if (_opts.delete_bitmap.count(segment_id()) > 0 &&

        _opts.delete_bitmap.at(segment_id()) != nullptr) {

        size_t pre_size = _row_bitmap.cardinality();

        _row_bitmap -= *(_opts.delete_bitmap.at(segment_id()));

        _opts.stats->rows_del_by_bitmap += (pre_size - _row_bitmap.cardinality());

        VLOG_DEBUG << "read on segment: " << segment_id() << ", delete bitmap cardinality: "

                   << _opts.delete_bitmap.at(segment_id())->cardinality() << ", "

                   << _opts.stats->rows_del_by_bitmap << " rows deleted by bitmap";

    }

    if (!_opts.row_ranges.is_empty()) {

        _row_bitmap &= RowRanges::ranges_to_roaring(_opts.row_ranges);

    }

    _prepare_score_column_materialization();

    RETURN_IF_ERROR(_apply_ann_topn_predicate());

    if (_opts.read_orderby_key_reverse) {

        _range_iter.reset(new BackwardBitmapRangeIterator(_row_bitmap));

    } else {

        _range_iter.reset(new BitmapRangeIterator(_row_bitmap));

    }

    // Reserve columns for _initial_block_row_max (the original max before any adaptive

    // prediction) because the predictor may increase block_row_max on subsequent batches

    // up to this ceiling. Using the current (possibly reduced) _opts.block_row_max would

    // cause heap-buffer-overflow if a later prediction is larger.

    auto nrows_reserve_limit =

            std::min(_row_bitmap.cardinality(), uint64_t(_initial_block_row_max));

    if (_lazy_materialization_read || _opts.record_rowids || _is_need_expr_eval) {

        _block_rowids.resize(_initial_block_row_max);

    }

    _current_return_columns.resize(_schema->columns().size());

    for (size_t i = 0; i < _schema->column_ids().size(); i++) {

        ColumnId cid = _schema->column_ids()[i];

        const auto* column_desc = _schema->column(cid);

        if (_is_pred_column[cid]) {

            auto storage_column_type = _storage_name_and_type[cid].second;

            RETURN_IF_CATCH_EXCEPTION(

                    // Here, cid will not go out of bounds

                    // because the size of _current_return_columns equals _schema->tablet_columns().size()

                    _current_return_columns[cid] = Schema::get_predicate_column_ptr(

                            storage_column_type, _opts.io_ctx.reader_type));

            _current_return_columns[cid]->set_rowset_segment_id(

                    {_segment->rowset_id(), _segment->id()});

            _current_return_columns[cid]->reserve(nrows_reserve_limit);

        } else if (i >= block->columns()) {

            // This column needs to be scanned, but doesn't need to be returned upward. (delete sign)

            // if i >= block->columns means the column and not the pred_column means `column i` is

            // a delete condition column. but the column is not effective in the segment. so we just

            // create a column to hold the data.

            // a. origin data -> b. delete condition -> c. new load data

            // the segment of c do not effective delete condition, but it still need read the column

            // to match the schema.

            // TODO: skip read the not effective delete column to speed up segment read.

            _current_return_columns[cid] = Schema::get_data_type_ptr(*column_desc)->create_column();

            _current_return_columns[cid]->reserve(nrows_reserve_limit);

        }

    }

    // Additional deleted filter condition will be materialized column be at the end of the block,

    // after _output_column_by_sel_idx  will be erase, we not need to filter it,

    // so erase it from _columns_to_filter in the first next_batch.

    // Eg:

    //      `delete from table where a = 10;`

    //      `select b from table;`

    // a column only effective in segment iterator, the block from query engine only contain the b column,

    // so no need to filter a column by expr.

    for (auto it = _columns_to_filter.begin(); it != _columns_to_filter.end();) {

        if (*it >= block->columns()) {

            it = _columns_to_filter.erase(it);

        } else {

            ++it;

        }

    }

    _lazy_inited = true;

    _init_segment_prefetchers();

    return Status::OK();

}

void SegmentIterator::_init_segment_prefetchers() {

    SCOPED_RAW_TIMER(&_opts.stats->segment_iterator_init_segment_prefetchers_timer_ns);

    if (!config::is_cloud_mode()) {

        return;

    }

    static std::vector<ReaderType> supported_reader_types {

            ReaderType::READER_QUERY, ReaderType::READER_BASE_COMPACTION,

            ReaderType::READER_CUMULATIVE_COMPACTION, ReaderType::READER_FULL_COMPACTION};

    if (std::ranges::none_of(supported_reader_types,

                             [&](ReaderType t) { return _opts.io_ctx.reader_type == t; })) {

        return;

    }

    // Initialize segment prefetcher for predicate and non-predicate columns

    bool is_query = (_opts.io_ctx.reader_type == ReaderType::READER_QUERY);

    bool enable_prefetch = is_query ? config::enable_query_segment_file_cache_prefetch

                                    : config::enable_compaction_segment_file_cache_prefetch;

    LOG_IF(INFO, config::enable_segment_prefetch_verbose_log) << fmt::format(

            "[verbose] SegmentIterator _init_segment_prefetchers, is_query={}, "

            "enable_prefetch={}, "

            "_row_bitmap.isEmpty()={}, row_bitmap.cardinality()={}, tablet={}, rowset={}, "

            "segment={}, predicate_column_ids={}, common_expr_column_ids={}",

            is_query, enable_prefetch, _row_bitmap.isEmpty(), _row_bitmap.cardinality(),

            _opts.tablet_id, _opts.rowset_id.to_string(), segment_id(),

            fmt::join(_predicate_column_ids, ","), fmt::join(_common_expr_column_ids, ","));

    if (enable_prefetch && !_row_bitmap.isEmpty()) {

        int window_size =

                1 + (is_query ? config::query_segment_file_cache_prefetch_block_size

                              : config::compaction_segment_file_cache_prefetch_block_size);

        LOG_IF(INFO, config::enable_segment_prefetch_verbose_log) << fmt::format(

                "[verbose] SegmentIterator prefetch config: window_size={}", window_size);

        if (window_size > 0 &&

            !_column_iterators.empty()) { // ensure init_iterators has been called

            SegmentPrefetcherConfig prefetch_config(window_size,

                                                    config::file_cache_each_block_size);

            for (auto cid : _schema->column_ids()) {

                auto& column_iter = _column_iterators[cid];

                if (column_iter == nullptr) {

                    continue;

                }

                const auto* tablet_column = _schema->column(cid);

                SegmentPrefetchParams params {

                        .config = prefetch_config,

                        .read_options = _opts,

                };

                LOG_IF(INFO, config::enable_segment_prefetch_verbose_log) << fmt::format(

                        "[verbose] SegmentIterator init_segment_prefetchers, "

                        "tablet={}, rowset={}, segment={}, column_id={}, col_name={}, type={}",

                        _opts.tablet_id, _opts.rowset_id.to_string(), segment_id(), cid,

                        tablet_column->name(), tablet_column->type());

                Status st = column_iter->init_prefetcher(params);

                if (!st.ok()) {

                    LOG_IF(WARNING, config::enable_segment_prefetch_verbose_log) << fmt::format(

                            "[verbose] failed to init prefetcher for column_id={}, "

                            "tablet={}, rowset={}, segment={}, error={}",

                            cid, _opts.tablet_id, _opts.rowset_id.to_string(), segment_id(),

                            st.to_string());

                }

            }

            // for compaction, it's guaranteed that all rows are read, so we can prefetch all data blocks

            PrefetcherInitMethod init_method = (is_query && _row_bitmap.cardinality() < num_rows())

                                                       ? PrefetcherInitMethod::FROM_ROWIDS

                                                       : PrefetcherInitMethod::ALL_DATA_BLOCKS;

            std::map<PrefetcherInitMethod, std::vector<SegmentPrefetcher*>> prefetchers;

            for (size_t idx = 0; idx < _column_iterators.size(); ++idx) {

                auto cid = cast_set<ColumnId>(idx);

                auto* column_iter = _column_iterators[cid].get();

                if (column_iter != nullptr) {

                    ScopedColumnIteratorReadPhase scoped_read_phase {

                            column_iter, _support_lazy_read_pruned_columns.contains(cid)

                                                 ? ColumnIterator::ReadPhase::PREDICATE

                                                 : ColumnIterator::ReadPhase::NORMAL};

                    column_iter->collect_prefetchers(prefetchers, init_method);

                }

            }

            for (auto& [method, prefetcher_vec] : prefetchers) {

                if (method == PrefetcherInitMethod::ALL_DATA_BLOCKS) {

                    for (auto* prefetcher : prefetcher_vec) {

                        prefetcher->build_all_data_blocks();

                    }

                } else if (method == PrefetcherInitMethod::FROM_ROWIDS && !prefetcher_vec.empty()) {

                    SegmentPrefetcher::build_blocks_by_rowids(_row_bitmap, prefetcher_vec);

                }

            }

        }

    }

}

Status SegmentIterator::_get_row_ranges_by_keys() {

    SCOPED_RAW_TIMER(&_opts.stats->generate_row_ranges_by_keys_ns);

    DorisMetrics::instance()->segment_row_total->increment(num_rows());

    // fast path for empty segment or empty key ranges

    if (_row_bitmap.isEmpty() || _opts.key_ranges.empty()) {

        return Status::OK();

    }

    // Read & seek key columns is a waste of time when no key column in _schema

    if (std::none_of(_schema->columns().begin(), _schema->columns().end(),

                     [&](const TabletColumnPtr& col) {

                         return col &&

                                _opts.tablet_schema->column_by_uid(col->unique_id()).is_key();

                     })) {

        return Status::OK();

    }

    RowRanges result_ranges;

    for (auto& key_range : _opts.key_ranges) {

        rowid_t lower_rowid = 0;

        rowid_t upper_rowid = num_rows();

        RETURN_IF_ERROR(_prepare_seek(key_range));

        if (key_range.upper_key != nullptr) {

            // If client want to read upper_bound, the include_upper is true. So we

            // should get the first ordinal at which key is larger than upper_bound.

            // So we call _lookup_ordinal with include_upper's negate

            RETURN_IF_ERROR(_lookup_ordinal(*key_range.upper_key, !key_range.include_upper,

                                            num_rows(), &upper_rowid));

        }

        if (upper_rowid > 0 && key_range.lower_key != nullptr) {

            RETURN_IF_ERROR(_lookup_ordinal(*key_range.lower_key, key_range.include_lower,

                                            upper_rowid, &lower_rowid));

        }

        auto row_range = RowRanges::create_single(lower_rowid, upper_rowid);

        RowRanges::ranges_union(result_ranges, row_range, &result_ranges);

    }

    size_t pre_size = _row_bitmap.cardinality();

    _row_bitmap &= RowRanges::ranges_to_roaring(result_ranges);

    _opts.stats->rows_key_range_filtered += (pre_size - _row_bitmap.cardinality());

    return Status::OK();

}

// Set up environment for the following seek.

Status SegmentIterator::_prepare_seek(const StorageReadOptions::KeyRange& key_range) {

    std::vector<const TabletColumn*> key_columns;

    std::set<uint32_t> column_set;

    if (key_range.lower_key != nullptr) {

        for (auto cid : key_range.lower_key->schema()->column_ids()) {

            column_set.emplace(cid);

            key_columns.emplace_back(key_range.lower_key->column(cid));

        }

    }

    if (key_range.upper_key != nullptr) {

        for (auto cid : key_range.upper_key->schema()->column_ids()) {

            if (column_set.count(cid) == 0) {

                key_columns.emplace_back(key_range.upper_key->column(cid));

                column_set.emplace(cid);

            }

        }

    }

    if (!_seek_schema) {

        std::vector<TabletColumnPtr> cols;

        cols.reserve(key_columns.size());

        for (const TabletColumn* col : key_columns) {

            cols.emplace_back(std::make_shared<TabletColumn>(*col));

        }

        std::vector<uint32_t> column_ids(cols.size());

        std::iota(column_ids.begin(), column_ids.end(), 0);

        _seek_schema = std::make_unique<Schema>(cols, column_ids);

    }

    // todo(wb) need refactor here, when using pk to search, _seek_block is useless

    if (_seek_block.size() == 0) {

        _seek_block.resize(_seek_schema->num_column_ids());

        int i = 0;

        for (auto cid : _seek_schema->column_ids()) {

            auto column_desc = _seek_schema->column(cid);

            _seek_block[i] = Schema::get_data_type_ptr(*column_desc)->create_column();

            i++;

        }

    }

    // create used column iterator

    for (auto cid : _seek_schema->column_ids()) {

        if (_column_iterators[cid] == nullptr) {

            // TODO: Do we need this?

            if (_virtual_column_exprs.contains(cid)) {

                _column_iterators[cid] = std::make_unique<VirtualColumnIterator>();

                continue;

            }

            RETURN_IF_ERROR(_segment->new_column_iterator(_opts.tablet_schema->column(cid),

                                                          &_column_iterators[cid], &_opts,

                                                          &_variant_sparse_column_cache));

            ColumnIteratorOptions iter_opts {

                    .use_page_cache = _opts.use_page_cache,

                    .file_reader = _file_reader.get(),

                    .stats = _opts.stats,

                    .io_ctx = _opts.io_ctx,

            };

            RETURN_IF_ERROR(_column_iterators[cid]->init(iter_opts));

        }

    }

    return Status::OK();

}

Status SegmentIterator::_get_row_ranges_by_column_conditions() {

    SCOPED_RAW_TIMER(&_opts.stats->generate_row_ranges_by_column_conditions_ns);

    if (_row_bitmap.isEmpty()) {

        return Status::OK();

    }

    {

        if (_opts.runtime_state &&

            _opts.runtime_state->query_options().enable_inverted_index_query &&

            (has_index_in_iterators() || !_common_expr_ctxs_push_down.empty())) {

            SCOPED_RAW_TIMER(&_opts.stats->inverted_index_filter_timer);

            size_t input_rows = _row_bitmap.cardinality();

            // Only apply column-level inverted index if we have iterators

            if (has_index_in_iterators()) {

                RETURN_IF_ERROR(_apply_inverted_index());

            }

            // Always apply expr-level index (e.g., search expressions) if we have common_expr_pushdown

            // This allows search expressions with variant subcolumns to be evaluated even when

            // the segment doesn't have all subcolumns

            RETURN_IF_ERROR(_apply_index_expr());

            for (auto it = _common_expr_ctxs_push_down.begin();

                 it != _common_expr_ctxs_push_down.end();) {

                if ((*it)->all_expr_inverted_index_evaluated()) {

                    const auto* result = (*it)->get_index_context()->get_index_result_for_expr(

                            (*it)->root().get());

                    if (result != nullptr) {

                        _row_bitmap &= *result->get_data_bitmap();

                        it = _common_expr_ctxs_push_down.erase(it);

                    }

                } else {

                    ++it;

                }

            }

            _opts.condition_cache_digest =

                    _common_expr_ctxs_push_down.empty() ? 0 : _opts.condition_cache_digest;

            _opts.stats->rows_inverted_index_filtered += (input_rows - _row_bitmap.cardinality());

            for (auto cid : _schema->column_ids()) {

                bool result_true = _check_all_conditions_passed_inverted_index_for_column(cid);

                if (result_true) {

                    _need_read_data_indices[cid] = false;

                }

            }

        }

    }

    DBUG_EXECUTE_IF("segment_iterator.inverted_index.filtered_rows", {

        LOG(INFO) << "Debug Point: segment_iterator.inverted_index.filtered_rows: "

                  << _opts.stats->rows_inverted_index_filtered;

        auto filtered_rows = DebugPoints::instance()->get_debug_param_or_default<int32_t>(

                "segment_iterator.inverted_index.filtered_rows", "filtered_rows", -1);

        if (filtered_rows != _opts.stats->rows_inverted_index_filtered) {

            return Status::Error<ErrorCode::INTERNAL_ERROR>(

                    "filtered_rows: {} not equal to expected: {}",

                    _opts.stats->rows_inverted_index_filtered, filtered_rows);

        }

    })

    DBUG_EXECUTE_IF("segment_iterator.apply_inverted_index", {

        LOG(INFO) << "Debug Point: segment_iterator.apply_inverted_index";

        if (!_common_expr_ctxs_push_down.empty() || !_col_predicates.empty()) {

            return Status::Error<ErrorCode::INTERNAL_ERROR>(

                    "it is failed to apply inverted index, common_expr_ctxs_push_down: {}, "

                    "col_predicates: {}",

                    _common_expr_ctxs_push_down.size(), _col_predicates.size());

        }

    })

    if (!_row_bitmap.isEmpty() &&

        (!_opts.topn_filter_source_node_ids.empty() || !_opts.col_id_to_predicates.empty() ||

         _opts.delete_condition_predicates->num_of_column_predicate() > 0 ||

         !_common_expr_ctxs_push_down.empty())) {

        RowRanges condition_row_ranges = RowRanges::create_single(_segment->num_rows());

        RETURN_IF_ERROR(_get_row_ranges_from_conditions(&condition_row_ranges));

        size_t pre_size = _row_bitmap.cardinality();

        _row_bitmap &= RowRanges::ranges_to_roaring(condition_row_ranges);

        _opts.stats->rows_conditions_filtered += (pre_size - _row_bitmap.cardinality());

    }

    DBUG_EXECUTE_IF("bloom_filter_must_filter_data", {

        if (_opts.stats->rows_bf_filtered == 0) {

            return Status::Error<ErrorCode::INTERNAL_ERROR>(

                    "Bloom filter did not filter the data.");

        }

    })

    // TODO(hkp): calculate filter rate to decide whether to

    // use zone map/bloom filter/secondary index or not.

    return Status::OK();

}

bool SegmentIterator::_column_has_ann_index(int32_t cid) {

    bool has_ann_index = _index_iterators[cid] != nullptr &&

                         _index_iterators[cid]->get_reader(AnnIndexReaderType::ANN);

    return has_ann_index;

}

Status SegmentIterator::_apply_ann_topn_predicate() {

    if (_ann_topn_runtime == nullptr) {

        return Status::OK();

    }

    VLOG_DEBUG << fmt::format("Try apply ann topn: {}", _ann_topn_runtime->debug_string());

    size_t src_col_idx = _ann_topn_runtime->get_src_column_idx();

    // AnnTopNRuntime keeps VSlotRef::column_id(), which is the scan schema ordinal.

    ColumnId src_cid = _schema->column_id(src_col_idx);

    IndexIterator* ann_index_iterator = _index_iterators[src_cid].get();

    bool has_ann_index = _column_has_ann_index(src_cid);

    bool has_common_expr_push_down = !_common_expr_ctxs_push_down.empty();

    bool has_column_predicate = std::any_of(_is_pred_column.begin(), _is_pred_column.end(),

                                            [](bool is_pred) { return is_pred; });

    if (!has_ann_index || has_common_expr_push_down || has_column_predicate) {

        VLOG_DEBUG << fmt::format(

                "Ann topn can not be evaluated by ann index, has_ann_index: {}, "

                "has_common_expr_push_down: {}, has_column_predicate: {}",

                has_ann_index, has_common_expr_push_down, has_column_predicate);

        // Disable index-only scan on ann indexed column.

        _need_read_data_indices[src_cid] = true;

        _opts.stats->ann_fall_back_brute_force_cnt += 1;

        return Status::OK();

    }

    // Process asc & desc according to the type of metric

    auto index_reader = ann_index_iterator->get_reader(AnnIndexReaderType::ANN);

    auto ann_index_reader = dynamic_cast<AnnIndexReader*>(index_reader.get());

    DCHECK(ann_index_reader != nullptr);

    if (ann_index_reader->get_metric_type() == AnnIndexMetric::IP) {

        if (_ann_topn_runtime->is_asc()) {

            VLOG_DEBUG << fmt::format(

                    "Asc topn for inner product can not be evaluated by ann index");

            // Disable index-only scan on ann indexed column.

            _need_read_data_indices[src_cid] = true;

            _opts.stats->ann_fall_back_brute_force_cnt += 1;

            return Status::OK();

        }

    } else {

        if (!_ann_topn_runtime->is_asc()) {

            VLOG_DEBUG << fmt::format("Desc topn for l2/cosine can not be evaluated by ann index");

            // Disable index-only scan on ann indexed column.

            _need_read_data_indices[src_cid] = true;

            _opts.stats->ann_fall_back_brute_force_cnt += 1;

            return Status::OK();

        }

    }