// 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 <assert.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 <algorithm>

#include <boost/iterator/iterator_facade.hpp>

#include <cstdint>

#include <memory>

#include <numeric>

#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/column/predicate_column.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/function/array/function_array_index.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/descriptors.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/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/json/path_in_data.h"

#include "util/simd/bits.h"

namespace doris {

using namespace ErrorCode;

namespace segment_v2 {

namespace {

Status tablet_column_id_by_slot(const TabletSchemaSPtr& tablet_schema, const SlotDescriptor* slot,

                                ColumnId* cid) {

    int32_t field_index = -1;

    if (slot->type()->get_primitive_type() == PrimitiveType::TYPE_VARIANT) {

        field_index = tablet_schema->field_index(

                PathInData(tablet_schema->column_by_uid(slot->col_unique_id()).name_lower_case(),

                           slot->column_paths()));

    } else {

        field_index = slot->col_unique_id() >= 0 ? tablet_schema->field_index(slot->col_unique_id())

                                                 : tablet_schema->field_index(slot->col_name());

    }

    if (field_index < 0) {

        return Status::InternalError(

                "field name is invalid. field={}, field_name_to_index={}, col_unique_id={}",

                slot->col_name(), tablet_schema->get_all_field_names(), slot->col_unique_id());

    }

    *cid = field_index;

    return Status::OK();

}

Status rebind_storage_expr_to_reader_schema(

        const StorageReadOptions& opts, const VExprSPtr& expr,

        const std::unordered_map<ColumnId, size_t>& cid_to_pos) {

    DORIS_CHECK(expr != nullptr);

    if (expr->is_slot_ref()) {

        auto slot_ref = std::static_pointer_cast<VSlotRef>(expr);

        auto* slot = opts.runtime_state->desc_tbl().get_slot_descriptor(slot_ref->slot_id());

        if (slot == nullptr) {

            return Status::InternalError("slot {} is not found in descriptor table",

                                         slot_ref->slot_id());

        }

        ColumnId cid = 0;

        RETURN_IF_ERROR(tablet_column_id_by_slot(opts.tablet_schema, slot, &cid));

        auto pos_it = cid_to_pos.find(cid);

        if (pos_it == cid_to_pos.end()) {

            return Status::InternalError("slot {} column {} with cid {} is not in reader schema",

                                         slot_ref->slot_id(), slot->col_name(), cid);

        }

        slot_ref->set_column_id(cast_set<int>(pos_it->second));

    } else if (expr->is_virtual_slot_ref()) {

        auto virtual_slot_ref = std::static_pointer_cast<VirtualSlotRef>(expr);

        auto* slot =

                opts.runtime_state->desc_tbl().get_slot_descriptor(virtual_slot_ref->slot_id());

        if (slot == nullptr) {

            return Status::InternalError("slot {} is not found in descriptor table",

                                         virtual_slot_ref->slot_id());

        }

        ColumnId cid = 0;

        RETURN_IF_ERROR(tablet_column_id_by_slot(opts.tablet_schema, slot, &cid));

        auto pos_it = cid_to_pos.find(cid);

        if (pos_it == cid_to_pos.end()) {

            return Status::InternalError(

                    "virtual slot {} column {} with cid {} is not in reader schema",

                    virtual_slot_ref->slot_id(), slot->col_name(), cid);

        }

        virtual_slot_ref->set_column_id(cast_set<int>(pos_it->second));

        // A virtual slot has its own output position in the reader block, and its

        // materialization expression may also contain real slot refs. Rebind both

        // sides so evaluating the virtual expression reads from the same block

        // layout used by SegmentIterator.

        RETURN_IF_ERROR(rebind_storage_expr_to_reader_schema(

                opts, virtual_slot_ref->get_virtual_column_expr(), cid_to_pos));

    }

    for (const auto& child : expr->children()) {

        RETURN_IF_ERROR(rebind_storage_expr_to_reader_schema(opts, child, cid_to_pos));

    }

    return Status::OK();

}

Status rebind_storage_exprs_to_reader_schema(const StorageReadOptions& opts, const Schema& schema,

                                             const VExprContextSPtrs& common_exprs,

                                             std::map<ColumnId, VExprContextSPtr>& virtual_exprs) {

    if (common_exprs.empty() && virtual_exprs.empty()) {

        return Status::OK();

    }

    DORIS_CHECK(opts.runtime_state != nullptr);

    DORIS_CHECK(opts.tablet_schema != nullptr);

    const auto keys_type = opts.tablet_schema->keys_type();

    if (keys_type == KeysType::DUP_KEYS ||

        (keys_type == KeysType::UNIQUE_KEYS && opts.enable_unique_key_merge_on_write)) {

        return Status::OK();

    }

    // Storage exprs are prepared with RowDescriptor, so VSlotRef/VirtualSlotRef column_id points to

    // the scan tuple column ordinal. SegmentIterator evaluates cloned exprs on a block built from

    // the reader schema instead. AGG_KEYS and non-MOW UNIQUE_KEYS readers may expand the reader

    // schema, for example by filling all key columns before merging/aggregating rows, so the scan

    // tuple ordinal is not always the same as the runtime block ordinal.

    //

    // DUP_KEYS and UNIQUE_KEYS MOW use direct readers for query scans, so their reader block keeps

    // the scan tuple layout and can skip this per-segment expression-tree traversal. For merge/agg

    // readers, the reader schema is the source of truth: map tablet column id to reader-block

    // position and rebind every storage expr slot to that position.

    std::unordered_map<ColumnId, size_t> cid_to_pos;

    for (size_t pos = 0; pos < schema.num_column_ids(); ++pos) {

        cid_to_pos.emplace(schema.column_id(cast_set<int>(pos)), pos);

    }

    for (const auto& ctx : common_exprs) {

        RETURN_IF_ERROR(rebind_storage_expr_to_reader_schema(opts, ctx->root(), cid_to_pos));

    }

    for (const auto& [_, ctx] : virtual_exprs) {

        RETURN_IF_ERROR(rebind_storage_expr_to_reader_schema(opts, ctx->root(), cid_to_pos));

    }

    return Status::OK();

}

} // namespace

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;

    _vir_cid_to_idx_in_block = _opts.vir_cid_to_idx_in_block;

    _score_runtime = _opts.score_runtime;

    _ann_topn_runtime = _opts.ann_topn_runtime;

    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: {}, "

            "_vir_cid_to_idx_in_block size: {}, common_expr_pushdown size: {}",

            _opts.virtual_column_exprs.size(), _opts.vir_cid_to_idx_in_block.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());

    _vec_init_char_column_id();

    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;

            // Char type is special , since char type's computational datatype is same with string,

            // both are DataTypeString, but DataTypeString only return FieldType::OLAP_FIELD_TYPE_STRING

            // in get_storage_field_type.

            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(

                            _is_char_type[cid] ? FieldType::OLAP_FIELD_TYPE_CHAR

                                               : storage_column_type->get_storage_field_type(),

                            storage_column_type->is_nullable(), _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 (const auto& column_iter : _column_iterators) {

                if (column_iter != nullptr) {

                    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()) {