// 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 "exec/scan/olap_scanner.h"

#include <gen_cpp/Descriptors_types.h>

#include <gen_cpp/PlanNodes_types.h>

#include <gen_cpp/Types_types.h>

#include <glog/logging.h>

#include <stdlib.h>

#include <thrift/protocol/TDebugProtocol.h>

#include <algorithm>

#include <atomic>

#include <iterator>

#include <ostream>

#include <set>

#include "cloud/cloud_storage_engine.h"

#include "cloud/cloud_tablet_hotspot.h"

#include "cloud/config.h"

#include "common/config.h"

#include "common/consts.h"

#include "common/logging.h"

#include "common/metrics/doris_metrics.h"

#include "core/block/block.h"

#include "core/data_type/data_type_number.h"

#include "exec/common/variant_util.h"

#include "exec/operator/olap_scan_operator.h"

#include "exec/scan/scan_node.h"

#include "exprs/function_filter.h"

#include "exprs/vexpr.h"

#include "exprs/vexpr_context.h"

#include "io/cache/block_file_cache_profile.h"

#include "io/io_common.h"

#include "runtime/descriptors.h"

#include "runtime/exec_env.h"

#include "runtime/query_context.h"

#include "runtime/runtime_profile.h"

#include "runtime/runtime_state.h"

#include "service/backend_options.h"

#include "storage/binlog.h"

#include "storage/id_manager.h"

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

#include "storage/iterator/block_reader.h"

#include "storage/olap_common.h"

#include "storage/olap_tuple.h"

#include "storage/olap_utils.h"

#include "storage/predicate/predicate_creator.h"

#include "storage/storage_engine.h"

#include "storage/tablet/tablet_schema.h"

#ifndef NDEBUG

#include "util/debug_points.h"

#endif

#include "util/json/path_in_data.h"

namespace doris {

#include "common/compile_check_avoid_begin.h"

using ReadSource = TabletReadSource;

OlapScanner::OlapScanner(ScanLocalStateBase* parent, OlapScanner::Params&& params)

        : Scanner(params.state, parent, params.limit, params.profile),

          _key_ranges(std::move(params.key_ranges)),

          _tablet_reader_params({.tablet = std::move(params.tablet),

                                 .tablet_schema {},

                                 .reader_type = params.read_row_binlog ? ReaderType::READER_BINLOG

                                                                       : ReaderType::READER_QUERY,

                                 .aggregation = params.aggregation,

                                 .version = {0, params.version},

                                 .start_key {},

                                 .end_key {},

                                 .predicates {},

                                 .function_filters {},

                                 .delete_predicates {},

                                 .target_cast_type_for_variants {},

                                 .all_access_paths {},

                                 .predicate_access_paths {},

                                 .rs_splits {},

                                 .return_columns {},

                                 .tso_predicate_column_id {},

                                 .output_columns {},

                                 .extra_columns {},

                                 .common_expr_ctxs_push_down {},

                                 .topn_filter_source_node_ids {},

                                 .key_group_cluster_key_idxes {},

                                 .virtual_column_exprs {},

                                 .score_runtime {},

                                 .collection_statistics {},

                                 .ann_topn_runtime {},

                                 .condition_cache_digest = parent->get_condition_cache_digest(),

                                 .binlog_scan_type = params.binlog_scan_type}),

          _start_tso(params.start_tso),

          _end_tso(params.end_tso),

          _initial_file_cache_stats(std::move(params.initial_file_cache_stats)) {

    _tablet_reader_params.set_read_source(std::move(params.read_source),

                                          _state->skip_delete_bitmap());

    _has_prepared = false;

    _vector_search_params = params.state->get_vector_search_params();

}

static std::string read_columns_to_string(TabletSchemaSPtr tablet_schema,

                                          const std::vector<uint32_t>& read_columns) {

    // avoid too long for one line,

    // it is hard to display in `show profile` stmt if one line is too long.

    const int col_per_line = 10;

    int i = 0;

    std::string read_columns_string;

    read_columns_string += "[";

    for (auto it = read_columns.cbegin(); it != read_columns.cend(); it++) {

        if (it != read_columns.cbegin()) {

            read_columns_string += ", ";

        }

        read_columns_string += tablet_schema->columns().at(*it)->name();

        if (i >= col_per_line) {

            read_columns_string += "\n";

            i = 0;

        } else {

            ++i;

        }

    }

    read_columns_string += "]";

    return read_columns_string;

}

static bool has_file_cache_statistics(const io::FileCacheStatistics& stats) {

    return stats.num_local_io_total != 0 || stats.num_remote_io_total != 0 ||

           stats.num_peer_io_total != 0 || stats.local_io_timer != 0 ||

           stats.bytes_read_from_local != 0 || stats.bytes_read_from_remote != 0 ||

           stats.bytes_read_from_peer != 0 || stats.remote_io_timer != 0 ||

           stats.peer_io_timer != 0 || stats.remote_wait_timer != 0 ||

           stats.write_cache_io_timer != 0 || stats.bytes_write_into_cache != 0 ||

           stats.num_skip_cache_io_total != 0 || stats.read_cache_file_directly_timer != 0 ||

           stats.cache_get_or_set_timer != 0 || stats.lock_wait_timer != 0 ||

           stats.get_timer != 0 || stats.set_timer != 0 ||

           stats.inverted_index_num_local_io_total != 0 ||

           stats.inverted_index_num_remote_io_total != 0 ||

           stats.inverted_index_num_peer_io_total != 0 ||

           stats.inverted_index_bytes_read_from_local != 0 ||

           stats.inverted_index_bytes_read_from_remote != 0 ||

           stats.inverted_index_bytes_read_from_peer != 0 ||

           stats.inverted_index_local_io_timer != 0 || stats.inverted_index_remote_io_timer != 0 ||

           stats.inverted_index_peer_io_timer != 0 || stats.inverted_index_io_timer != 0;

}

io::IOContext build_score_runtime_collection_io_context(RuntimeState* state, ReaderType reader_type,

                                                        int64_t expiration_time,

                                                        io::FileCacheStatistics* file_cache_stats) {

    io::IOContext io_ctx {

            .reader_type = reader_type,

            .expiration_time = expiration_time,

            .query_id = &state->query_id(),

            .file_cache_stats = file_cache_stats,

            .is_inverted_index = true,

    };

    if (auto* query_ctx = state->get_query_ctx(); query_ctx != nullptr) {

        io_ctx.remote_scan_cache_write_limiter = query_ctx->remote_scan_cache_write_limiter();

    }

    return io_ctx;

}

Status OlapScanner::_prepare_impl() {

    auto* local_state = static_cast<OlapScanLocalState*>(_local_state);

    auto& tablet = _tablet_reader_params.tablet;

    auto& tablet_schema = _tablet_reader_params.tablet_schema;

    DBUG_EXECUTE_IF("CloudTablet.capture_rs_readers.return.e-230", {

        LOG_WARNING("CloudTablet.capture_rs_readers.return e-230 init")

                .tag("tablet_id", tablet->tablet_id());

        return Status::Error<false>(-230, "injected error");

    });

    for (auto& ctx : local_state->_common_expr_ctxs_push_down) {

        VExprContextSPtr context;

        RETURN_IF_ERROR(ctx->clone(_state, context));

        _common_expr_ctxs_push_down.emplace_back(context);

        context->prepare_ann_range_search(_vector_search_params);

    }

    for (auto pair : local_state->_slot_id_to_virtual_column_expr) {

        // Scanner will be executed in a different thread, so we need to clone the context.

        VExprContextSPtr context;

        RETURN_IF_ERROR(pair.second->clone(_state, context));

        _slot_id_to_virtual_column_expr[pair.first] = context;

    }

    _score_runtime = local_state->_score_runtime;

    // All scanners share the same ann_topn_runtime.

    _ann_topn_runtime = local_state->_ann_topn_runtime;

    // set limit to reduce end of rowset and segment mem use

    _tablet_reader = std::make_unique<BlockReader>();

    // batch size is passed down to segment iterator, use _state->batch_size()

    // instead of _parent->limit(), because if _parent->limit() is a very small

    // value (e.g. select a from t where a .. and b ... limit 1),

    // it will be very slow when reading data in segment iterator

    _tablet_reader->set_batch_size(_state->batch_size());

    // Adaptive batch size: pass byte-budget settings to the storage reader.

    // The reader still uses batch_size() as the row ceiling.

    _tablet_reader->set_preferred_block_size_bytes(_state->preferred_block_size_bytes());

    {

        TOlapScanNode& olap_scan_node = local_state->olap_scan_node();

        TabletSchemaSPtr source_tablet_schema =

                _tablet_reader_params.reader_type == ReaderType::READER_BINLOG

                        ? tablet->row_binlog_tablet_schema()

                        : tablet->tablet_schema();

        tablet_schema = std::make_shared<TabletSchema>();

        tablet_schema->copy_from(*source_tablet_schema);

        if (olap_scan_node.__isset.columns_desc && !olap_scan_node.columns_desc.empty() &&

            olap_scan_node.columns_desc[0].col_unique_id >= 0) {

            tablet_schema->clear_columns();

            for (const auto& column_desc : olap_scan_node.columns_desc) {

                tablet_schema->append_column(TabletColumn(column_desc));

            }

            if (olap_scan_node.__isset.schema_version) {

                tablet_schema->set_schema_version(olap_scan_node.schema_version);

            }

        }

        if (olap_scan_node.__isset.indexes_desc) {

            tablet_schema->update_indexes_from_thrift(olap_scan_node.indexes_desc);

        }

        if (_tablet_reader_params.rs_splits.empty()) {

            // Non-pipeline mode, Tablet : Scanner = 1 : 1

            // acquire tablet rowset readers at the beginning of the scan node

            // to prevent this case: when there are lots of olap scanners to run for example 10000

            // the rowsets maybe compacted when the last olap scanner starts

            ReadSource read_source;

            if (config::is_cloud_mode()) {

                // FIXME(plat1ko): Avoid pointer cast

                ExecEnv::GetInstance()->storage_engine().to_cloud().tablet_hotspot().count(*tablet);

            }

            auto maybe_read_source = tablet->capture_read_source(

                    _tablet_reader_params.version,

                    {

                            .skip_missing_versions = _state->skip_missing_version(),

                            .enable_fetch_rowsets_from_peers =

                                    config::enable_fetch_rowsets_from_peer_replicas,

                            .capture_row_binlog =

                                    _tablet_reader_params.reader_type == ReaderType::READER_BINLOG,

                            .enable_prefer_cached_rowset =

                                    config::is_cloud_mode() ? _state->enable_prefer_cached_rowset()

                                                            : false,

                            .query_freshness_tolerance_ms =

                                    config::is_cloud_mode() ? _state->query_freshness_tolerance_ms()

                                                            : -1,

                    });

            if (!maybe_read_source) {

                LOG(WARNING) << "fail to init reader. res=" << maybe_read_source.error();

                return maybe_read_source.error();

            }

            read_source = std::move(maybe_read_source.value());

            if (config::enable_mow_verbose_log && tablet->enable_unique_key_merge_on_write()) {

                LOG_INFO("finish capture_rs_readers for tablet={}, query_id={}",

                         tablet->tablet_id(), print_id(_state->query_id()));

            }

            if (!_state->skip_delete_predicate()) {

                read_source.fill_delete_predicates();

            }

            _tablet_reader_params.set_read_source(std::move(read_source));

        }

        // Initialize tablet_reader_params

        RETURN_IF_ERROR(_init_tablet_reader_params(

                local_state->_parent->cast<OlapScanOperatorX>()._slot_id_to_slot_desc, _key_ranges,

                local_state->_slot_id_to_predicates, local_state->_push_down_functions));

    }

    // add read columns in profile

    if (_state->enable_profile()) {

        _profile->add_info_string("ReadColumns",

                                  read_columns_to_string(tablet_schema, _return_columns));

    }

    if (_tablet_reader_params.score_runtime) {

        SCOPED_TIMER(local_state->_statistics_collect_timer);

        _tablet_reader_params.collection_statistics = std::make_shared<CollectionStatistics>();

        auto io_ctx = build_score_runtime_collection_io_context(

                _state, _tablet_reader_params.reader_type, tablet->ttl_seconds(),

                &_tablet_reader->mutable_stats()->file_cache_stats);

        RETURN_IF_ERROR(_tablet_reader_params.collection_statistics->collect(

                _state, _tablet_reader_params.rs_splits, _tablet_reader_params.tablet_schema,

                _tablet_reader_params.common_expr_ctxs_push_down, &io_ctx));

    }

    _has_prepared = true;

    return Status::OK();

}

Status OlapScanner::_open_impl(RuntimeState* state) {

    RETURN_IF_ERROR(Scanner::_open_impl(state));

    SCOPED_TIMER(_local_state->cast<OlapScanLocalState>()._reader_init_timer);

    auto res = _tablet_reader->init(_tablet_reader_params);

    if (!res.ok()) {

        // init() also runs the eager first-row read that evaluates pushed-down expressions,

        // so res may be a data/expression error rather than a storage failure. Keep its own

        // message and only append the tablet/backend, without a misleading storage wording.

        res.append(". tablet=" + std::to_string(_tablet_reader_params.tablet->tablet_id()) +

                   ", backend=" + BackendOptions::get_localhost());

        return res;

    }

    _tablet_reader->mutable_stats()->file_cache_stats.merge_from(_initial_file_cache_stats);

    // Do not hold rs_splits any more to release memory.

    _tablet_reader_params.rs_splits.clear();

    return Status::OK();

}

// For binlog partition-based incremental read. Pushes down [start_tso, end_tso] range

// predicates onto the TSO column. If the TSO column is not already returned, pass it as

// a storage-only predicate column instead of widening scan output.

Status OlapScanner::_init_tso_predicates() {

    if (!_start_tso.has_value() && !_end_tso.has_value()) {

        return Status::OK();

    }

    auto& tablet_schema = _tablet_reader_params.tablet_schema;

    int32_t tso_index = _tablet_reader_params.reader_type == ReaderType::READER_BINLOG

                                ? tablet_schema->binlog_tso_col_idx()

                                : tablet_schema->commit_tso_col_idx();

    const std::string& column_name = _tablet_reader_params.reader_type == ReaderType::READER_BINLOG

                                             ? BINLOG_TSO_COL

                                             : COMMIT_TSO_COL;

    if (tso_index < 0) {

        return Status::InternalError("Column {} not found in tablet schema after append",

                                     column_name);

    }

    auto data_type = std::make_shared<DataTypeInt64>();

    if (_start_tso.has_value()) {

        Field start_value = Field::create_field<TYPE_BIGINT>(*_start_tso);

        _tablet_reader_params.predicates.push_back(create_comparison_predicate<PredicateType::GT>(

                tso_index, column_name, data_type, start_value, false));

    }

    if (_end_tso.has_value()) {

        Field end_value = Field::create_field<TYPE_BIGINT>(*_end_tso);

        _tablet_reader_params.predicates.push_back(create_comparison_predicate<PredicateType::LE>(

                tso_index, column_name, data_type, end_value, false));

    }

    // The storage-layer statistics fast path (VStatisticsIterator, picked when

    // push_down_agg_type is COUNT/MINMAX) bypasses SegmentIterator and returns raw

    // segment row counts without applying any column predicate. The commit-tso

    // predicate injected above is row-level, so the fast path would both miscount

    // (ignoring commit_tso <= snapshot_tso) and crash on a column-count DCHECK when

    // the tso predicate column is not in return_columns. Disable it here, matching

    // the binlog DETAIL/MIN_DELTA handling.

    _tablet_reader_params.push_down_agg_type_opt = TPushAggOp::NONE;

    if (std::find(_tablet_reader_params.return_columns.begin(),

                  _tablet_reader_params.return_columns.end(),

                  tso_index) == _tablet_reader_params.return_columns.end()) {

        _tablet_reader_params.tso_predicate_column_id = static_cast<ColumnId>(tso_index);

    }

    return Status::OK();

}

// it will be called under tablet read lock because capture rs readers need

Status OlapScanner::_init_tablet_reader_params(

        const phmap::flat_hash_map<int, SlotDescriptor*>& slot_id_to_slot_desc,

        const std::vector<OlapScanRange*>& key_ranges,

        const phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&

                slot_to_predicates,

        const std::vector<FunctionFilter>& function_filters) {

    // if the table with rowset [0-x] or [0-1] [2-y], and [0-1] is empty

    const bool single_version = _tablet_reader_params.has_single_version();

    auto* olap_local_state = static_cast<OlapScanLocalState*>(_local_state);

    bool read_mor_as_dup = olap_local_state->olap_scan_node().__isset.read_mor_as_dup &&

                           olap_local_state->olap_scan_node().read_mor_as_dup;

    if (_state->skip_storage_engine_merge() || read_mor_as_dup) {

        _tablet_reader_params.direct_mode = true;

        _tablet_reader_params.aggregation = true;

    } else {

        auto push_down_agg_type = _local_state->get_push_down_agg_type();

        _tablet_reader_params.direct_mode = _tablet_reader_params.aggregation || single_version ||

                                            (push_down_agg_type != TPushAggOp::NONE &&

                                             push_down_agg_type != TPushAggOp::COUNT_ON_INDEX);

    }

    RETURN_IF_ERROR(_init_variant_columns());

    RETURN_IF_ERROR(_init_return_columns());

    _tablet_reader_params.push_down_agg_type_opt = _local_state->get_push_down_agg_type();

    // Binlog DETAIL/MIN_DELTA scans widen `return_columns` with key/tso/op/before

    // columns to drive the row-level merge in BlockReader. The storage-layer

    // statistics fast path (VStatisticsIterator, picked when push_down_agg_type

    // is COUNT/MINMAX) bypasses SegmentIterator entirely, returning raw segment

    // row counts without binlog op filtering and with a schema that does not

    // match the widened read schema. The result is both wrong (raw segment

    // count != binlog row count) and unsafe (column-count DCHECK fires inside

    // VStatisticsIterator::next_batch). Disable the fast path for these scans.

    if (_tablet_reader_params.binlog_scan_type == TBinlogScanType::DETAIL ||

        _tablet_reader_params.binlog_scan_type == TBinlogScanType::MIN_DELTA) {

        _tablet_reader_params.push_down_agg_type_opt = TPushAggOp::NONE;

    }

    _tablet_reader_params.common_expr_ctxs_push_down = _common_expr_ctxs_push_down;

    _tablet_reader_params.virtual_column_exprs = _virtual_column_exprs;

    _tablet_reader_params.score_runtime = _score_runtime;

    _tablet_reader_params.output_columns = ((OlapScanLocalState*)_local_state)->_output_column_ids;

    _tablet_reader_params.ann_topn_runtime = _ann_topn_runtime;

    for (const auto& ele : ((OlapScanLocalState*)_local_state)->_cast_types_for_variants) {

        _tablet_reader_params.target_cast_type_for_variants[ele.first] = ele.second;

    };

    auto& tablet_schema = _tablet_reader_params.tablet_schema;

    for (auto& predicates : slot_to_predicates) {

        const int sid = predicates.first;

        DCHECK(slot_id_to_slot_desc.contains(sid));

        int32_t index =

                tablet_schema->field_index(slot_id_to_slot_desc.find(sid)->second->col_name());

        if (index < 0) {

            throw Exception(

                    Status::InternalError("Column {} not found in tablet schema",

                                          slot_id_to_slot_desc.find(sid)->second->col_name()));

        }

        for (auto& predicate : predicates.second) {

            _tablet_reader_params.predicates.push_back(predicate->clone(index));

        }

    }

    std::copy(function_filters.cbegin(), function_filters.cend(),

              std::inserter(_tablet_reader_params.function_filters,

                            _tablet_reader_params.function_filters.begin()));

    // Merge the columns in delete predicate that not in latest schema in to current tablet schema

    for (auto& del_pred : _tablet_reader_params.delete_predicates) {

        tablet_schema->merge_dropped_columns(*del_pred->tablet_schema());

    }

    // Push key ranges to the tablet reader.

    // Skip the "full scan" placeholder (has_lower_bound == false) — when no key

    // predicates exist, start_key/end_key remain empty and the reader does a full scan.

    for (auto* key_range : key_ranges) {

        if (!key_range->has_lower_bound) {

            continue;

        }

        _tablet_reader_params.start_key_include = key_range->begin_include;

        _tablet_reader_params.end_key_include = key_range->end_include;

        _tablet_reader_params.start_key.push_back(key_range->begin_scan_range);

        _tablet_reader_params.end_key.push_back(key_range->end_scan_range);

    }

    _tablet_reader_params.profile = _local_state->custom_profile();

    _tablet_reader_params.runtime_state = _state;

    _tablet_reader_params.origin_return_columns = &_return_columns;

    _tablet_reader_params.tablet_columns_convert_to_null_set = &_tablet_columns_convert_to_null_set;

    auto add_return_column_if_absent = [&](uint32_t cid) {

        if (std::find(_tablet_reader_params.return_columns.begin(),

                      _tablet_reader_params.return_columns.end(),

                      cid) == _tablet_reader_params.return_columns.end()) {

            _tablet_reader_params.return_columns.push_back(cid);

        }

    };

    // For row-binlog scans that emit BEFORE/AFTER pairs (MIN_DELTA / DETAIL), we must read

    // every key column, every requested value column, the binlog meta columns (tso / op)

    // and their __BEFORE__ mirrors, so the BlockReader can reconstruct change rows.

    const bool need_before_columns =

            _tablet_reader_params.binlog_scan_type == TBinlogScanType::MIN_DELTA ||

            _tablet_reader_params.binlog_scan_type == TBinlogScanType::DETAIL;

    if (need_before_columns) {

        for (size_t i = 0; i < tablet_schema->num_key_columns(); ++i) {

            add_return_column_if_absent(static_cast<uint32_t>(i));

        }

        for (auto cid : _return_columns) {

            add_return_column_if_absent(cid);

        }

        if (int32_t tso_idx = tablet_schema->binlog_tso_col_idx(); tso_idx >= 0) {

            add_return_column_if_absent(static_cast<uint32_t>(tso_idx));

        }

        if (int32_t op_idx = tablet_schema->binlog_op_col_idx(); op_idx >= 0) {

            add_return_column_if_absent(static_cast<uint32_t>(op_idx));

        }

        for (auto cid : _return_columns) {

            if (cid >= tablet_schema->num_key_columns()) {

                const auto& col_name = tablet_schema->column(cid).name();

                std::string before_col_name;

                before_col_name.append("__BEFORE__");

                before_col_name.append(col_name);

                before_col_name.append("__");

                if (int32_t before_idx = tablet_schema->field_index(before_col_name);

                    before_idx >= 0) {

                    add_return_column_if_absent(static_cast<uint32_t>(before_idx));

                }

            }

        }

    } else if (_tablet_reader_params.direct_mode) {

        _tablet_reader_params.return_columns = _return_columns;

    } else {

        // we need to fetch all key columns to do the right aggregation on storage engine side.

        for (size_t i = 0; i < tablet_schema->num_key_columns(); ++i) {

            _tablet_reader_params.return_columns.push_back(i);

        }

        for (auto index : _return_columns) {

            if (tablet_schema->column(index).is_key()) {

                continue;

            }

            _tablet_reader_params.return_columns.push_back(index);

        }

        // expand the sequence column

        if (tablet_schema->has_sequence_col() || tablet_schema->has_seq_map()) {

            bool has_replace_col = false;

            for (auto col : _return_columns) {

                if (tablet_schema->column(col).aggregation() ==

                    FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE) {

                    has_replace_col = true;

                    break;

                }

            }

            if (auto sequence_col_idx = tablet_schema->sequence_col_idx();

                has_replace_col && tablet_schema->has_sequence_col() &&

                std::find(_return_columns.begin(), _return_columns.end(), sequence_col_idx) ==

                        _return_columns.end()) {

                _tablet_reader_params.return_columns.push_back(sequence_col_idx);

            }

            if (has_replace_col) {

                const auto& val_to_seq = tablet_schema->value_col_idx_to_seq_col_idx();

                std::set<uint32_t> return_seq_columns;

                for (auto col : _tablet_reader_params.return_columns) {

                    // we need to add the necessary sequence column in _return_columns, and

                    // Avoid adding the same seq column twice

                    const auto val_iter = val_to_seq.find(col);

                    if (val_iter != val_to_seq.end()) {

                        auto seq = val_iter->second;

                        if (std::find(_tablet_reader_params.return_columns.begin(),

                                      _tablet_reader_params.return_columns.end(),

                                      seq) == _tablet_reader_params.return_columns.end()) {

                            return_seq_columns.insert(seq);

                        }

                    }

                }

                _tablet_reader_params.return_columns.insert(

                        std::end(_tablet_reader_params.return_columns),

                        std::begin(return_seq_columns), std::end(return_seq_columns));

            }

        }

    }

    RETURN_IF_ERROR(_init_tso_predicates());

    // For any row-binlog scan, force the storage layer to deliver rows strictly in primary-key

    // order so the BlockReader can group consecutive same-key changes (MIN_DELTA) or emit

    // BEFORE/AFTER pairs in deterministic order (DETAIL). Disable ORDER BY / TopN pushdowns

    // and reset their related params, since they would otherwise re-order the stream.

    if (_tablet_reader_params.binlog_scan_type != TBinlogScanType::NONE) {

        _tablet_reader_params.read_orderby_key = true;

        _tablet_reader_params.read_orderby_key_reverse = false;

        _tablet_reader_params.read_orderby_key_num_prefix_columns = 0;

        _tablet_reader_params.read_orderby_key_limit = 0;

        _tablet_reader_params.force_key_ordered_read = true;

        _tablet_reader_params.topn_filter_source_node_ids.clear();

    }

    _tablet_reader_params.use_page_cache = _state->enable_page_cache();

    DBUG_EXECUTE_IF("NewOlapScanner::_init_tablet_reader_params.block", DBUG_BLOCK);

    if (!_state->skip_storage_engine_merge()) {

        auto* olap_scan_local_state = (OlapScanLocalState*)_local_state;

        TOlapScanNode& olap_scan_node = olap_scan_local_state->olap_scan_node();

        // Set MOR value predicate pushdown flag

        if (olap_scan_node.__isset.enable_mor_value_predicate_pushdown &&

            olap_scan_node.enable_mor_value_predicate_pushdown) {

            _tablet_reader_params.enable_mor_value_predicate_pushdown = true;

        }

        const bool has_key_topn =

                olap_scan_node.__isset.sort_info && !olap_scan_node.sort_info.is_asc_order.empty();

        if (has_key_topn) {

            _limit = _local_state->limit_per_scanner();

        }

        const bool no_runtime_filters = _total_rf_num == 0;

        const bool segment_limit_enabled = _state->enable_segment_limit_pushdown();

        const bool storage_no_merge = olap_scan_local_state->_storage_no_merge();

        if (_limit > 0 && no_runtime_filters && segment_limit_enabled && storage_no_merge) {

            for (const auto& conjunct : _conjuncts) {

                DORIS_CHECK(!olap_scan_local_state->_check_expr_storage_filter(

                        conjunct->root(), OlapScanLocalState::ExprStorageFilterCheckMode::

                                                  HAS_SEGMENT_EVALUABLE_EXPR));

            }

        }

        // Segment LIMIT has only two legal states: completely disabled, or enabled after every

        // row-filtering conjunct has become a storage predicate or SegmentIterator common expr.

        const bool can_push_down_segment_limit = _limit > 0 && no_runtime_filters &&

                                                 _conjuncts.empty() && segment_limit_enabled &&

                                                 storage_no_merge;

        if (can_push_down_segment_limit) {

            if (has_key_topn) {

                _tablet_reader_params.read_orderby_key = true;

                if (!olap_scan_node.sort_info.is_asc_order[0]) {

                    _tablet_reader_params.read_orderby_key_reverse = true;

                }

                _tablet_reader_params.read_orderby_key_num_prefix_columns =

                        olap_scan_node.sort_info.is_asc_order.size();

                _tablet_reader_params.read_orderby_key_limit = _limit;

            } else {

                _tablet_reader_params.general_read_limit = _limit;

            }

        }

        if (_tablet_reader_params.read_orderby_key_limit > 0 ||

            _tablet_reader_params.general_read_limit > 0) {

            DORIS_CHECK(can_push_down_segment_limit);

            DORIS_CHECK(_conjuncts.empty());

        }

        // A key TopN scan cannot share the plain LIMIT early-stop counter. If

        // storage TopN is pushed down, each scanner must produce its full local

        // candidates. If it is not pushed down for any reason, the upper TopN

        // still needs all rows from the scan.

        if (has_key_topn) {

            _shared_scan_limit = nullptr;

            if (_tablet_reader_params.read_orderby_key_limit == 0) {

                _limit = -1;

            }

        }

        // Note: _shared_scan_limit is intentionally not pushed into the

        // storage layer. SegmentIterator's _process_eof() is irreversible,

        // so a concurrently-decremented atomic could reach 0 while a segment

        // still has data needed by other scanners.

        // set push down topn filter

        _tablet_reader_params.topn_filter_source_node_ids =

                olap_scan_local_state->get_topn_filter_source_node_ids(_state, true);

        if (!_tablet_reader_params.topn_filter_source_node_ids.empty()) {

            _tablet_reader_params.topn_filter_target_node_id =

                    olap_scan_local_state->parent()->node_id();

        }

    }

    // If this is a Two-Phase read query, and we need to delay the release of Rowset

    // by rowset->update_delayed_expired_timestamp().This could expand the lifespan of Rowset

    if (tablet_schema->field_index(BeConsts::ROWID_COL) >= 0) {

        constexpr static int delayed_s = 60;

        for (auto rs_reader : _tablet_reader_params.rs_splits) {

            uint64_t delayed_expired_timestamp =

                    UnixSeconds() + _tablet_reader_params.runtime_state->execution_timeout() +

                    delayed_s;

            rs_reader.rs_reader->rowset()->update_delayed_expired_timestamp(

                    delayed_expired_timestamp);

            ExecEnv::GetInstance()->storage_engine().add_quering_rowset(

                    rs_reader.rs_reader->rowset());

        }

    }

    if (tablet_schema->has_global_row_id()) {

        auto& id_file_map = _state->get_id_file_map();

        for (auto rs_reader : _tablet_reader_params.rs_splits) {

            id_file_map->add_temp_rowset(rs_reader.rs_reader->rowset());

        }

    }

    return Status::OK();

}

Status OlapScanner::_init_variant_columns() {

    auto& tablet_schema = _tablet_reader_params.tablet_schema;

    if (tablet_schema->num_variant_columns() == 0) {

        return Status::OK();

    }

    // Parent column has path info to distinction from each other

    for (auto* slot : _output_tuple_desc->slots()) {

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

            // Such columns are not exist in frontend schema info, so we need to

            // add them into tablet_schema for later column indexing.

            const auto& dt_variant =

                    assert_cast<const DataTypeVariant&>(*remove_nullable(slot->type()));

            TabletColumn subcol = TabletColumn::create_materialized_variant_column(

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

                    slot->column_paths(), slot->col_unique_id(),

                    dt_variant.variant_max_subcolumns_count(), dt_variant.enable_doc_mode());

            if (tablet_schema->field_index(*subcol.path_info_ptr()) < 0) {

                tablet_schema->append_column(subcol, TabletSchema::ColumnType::VARIANT);

            }

        }

    }

    variant_util::inherit_column_attributes(tablet_schema);

    return Status::OK();

}

Status OlapScanner::_init_return_columns() {

    // For OLAP scan, _output_tuple_desc is the storage-aligned scan tuple

    // descriptor. extra_key_column_slot_ids marks extra key slots that are

    // present only for scan-schema alignment. For example, on an AGG table with

    // keys (k1, k2), a query returning only k2 may still scan (k1, k2); k1 is

    // an extra column and can be removed by the projection output tuple.

    for (auto* slot : _output_tuple_desc->slots()) {

        // variant column using path to index a column

        int32_t index = 0;

        auto& tablet_schema = _tablet_reader_params.tablet_schema;

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

            index = tablet_schema->field_index(PathInData(

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

                    slot->column_paths()));

        } else {

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

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

        }

        if (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());

        }

        if (slot->get_virtual_column_expr()) {

            _virtual_column_exprs[index] = _slot_id_to_virtual_column_expr[slot->id()];

            VLOG_DEBUG << fmt::format("Virtual column, slot id: {}, cid {}, type: {}", slot->id(),

                                      index, slot->get_data_type_ptr()->get_name());

        }

        const auto& column = tablet_schema->column(index);

        auto* olap_local_state = static_cast<OlapScanLocalState*>(_local_state);

        const auto& olap_scan_node = olap_local_state->olap_scan_node();

        if (olap_scan_node.__isset.extra_key_column_slot_ids &&

            olap_scan_node.extra_key_column_slot_ids.contains(slot->id())) {

            DORIS_CHECK(column.is_key());

            if (_tablet_reader_params.direct_mode) {

                // Direct readers can synthesize extra storage keys because they are only

                // placeholders before the scan projection removes them. Merge/aggregation

                // readers must still read real key values to preserve storage semantics.

                _tablet_reader_params.extra_columns.insert(index);

            }

        }

        int32_t unique_id =

                column.unique_id() >= 0 ? column.unique_id() : column.parent_unique_id();

        if (!slot->all_access_paths().empty()) {

            _tablet_reader_params.all_access_paths.insert({unique_id, slot->all_access_paths()});

        }

        if (!slot->predicate_access_paths().empty()) {

            _tablet_reader_params.predicate_access_paths.insert(

                    {unique_id, slot->predicate_access_paths()});

        }

        if ((slot->type()->get_primitive_type() == PrimitiveType::TYPE_STRUCT ||

             slot->type()->get_primitive_type() == PrimitiveType::TYPE_MAP ||

             slot->type()->get_primitive_type() == PrimitiveType::TYPE_ARRAY) &&

            !slot->all_access_paths().empty()) {

            tablet_schema->add_pruned_columns_data_type(column.unique_id(), slot->type());

        }

        _return_columns.push_back(index);

        if (slot->is_nullable() && !tablet_schema->column(index).is_nullable()) {

            _tablet_columns_convert_to_null_set.emplace(index);

        } else if (!slot->is_nullable() && tablet_schema->column(index).is_nullable()) {

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

                    "slot(id: {}, name: {})'s nullable does not match "

                    "column(tablet id: {}, index: {}, name: {}) ",

                    slot->id(), slot->col_name(), tablet_schema->table_id(), index,

                    tablet_schema->column(index).name());

        }

    }

    if (_return_columns.empty()) {

        return Status::InternalError("failed to build storage scanner, no materialized slot!");

    }

    return Status::OK();

}

bool OlapScanner::check_partition_pruned() const {

    if (!_local_state) {

        return false;

    }

    return _local_state->is_partition_pruned(_tablet_reader_params.tablet->partition_id());

}

doris::TabletStorageType OlapScanner::get_storage_type() {

    if (config::is_cloud_mode()) {

        // we don't have cold storage in cloud mode, all storage is treated as local

        return doris::TabletStorageType::STORAGE_TYPE_LOCAL;

    }

    int local_reader = 0;

    for (const auto& reader : _tablet_reader_params.rs_splits) {

        local_reader += reader.rs_reader->rowset()->is_local();

    }

    int total_reader = _tablet_reader_params.rs_splits.size();

    if (local_reader == total_reader) {

        return doris::TabletStorageType::STORAGE_TYPE_LOCAL;

    } else if (local_reader == 0) {

        return doris::TabletStorageType::STORAGE_TYPE_REMOTE;

    }

    return doris::TabletStorageType::STORAGE_TYPE_REMOTE_AND_LOCAL;

}

Status OlapScanner::_get_block_impl(RuntimeState* state, Block* block, bool* eof) {

    // Read one block from block reader

    // ATTN: Here we need to let the _get_block_impl method guarantee the semantics of the interface,

    // that is, eof can be set to true only when the returned block is empty.

    RETURN_IF_ERROR(_tablet_reader->next_block_with_aggregation(block, eof));

    if (block->rows() > 0) {

        _tablet_reader_params.tablet->read_block_count.fetch_add(1, std::memory_order_relaxed);

        *eof = false;

    }

#ifndef NDEBUG

    RETURN_IF_ERROR(_check_ann_cache_hit_debug_points(_tablet_reader->stats()));

#endif

    return Status::OK();

}

Status OlapScanner::close(RuntimeState* state) {

    if (!_try_close()) {

        return Status::OK();

    }

    RETURN_IF_ERROR(Scanner::close(state));

    return Status::OK();

}

void OlapScanner::update_realtime_counters() {

    if (!_has_prepared) {

        // Counter update need prepare successfully, or it maybe core. For example, olap scanner

        // will open tablet reader during prepare, if not prepare successfully, tablet reader == nullptr.

        return;

    }

    OlapScanLocalState* local_state = static_cast<OlapScanLocalState*>(_local_state);

    const OlapReaderStatistics& stats = _tablet_reader->stats();

    COUNTER_UPDATE(local_state->_read_compressed_counter, stats.compressed_bytes_read);

    COUNTER_UPDATE(local_state->_read_uncompressed_counter, stats.uncompressed_bytes_read);

    COUNTER_UPDATE(local_state->_scan_bytes, stats.uncompressed_bytes_read);

    COUNTER_UPDATE(local_state->_scan_rows, stats.raw_rows_read);

    // Make sure the scan bytes and scan rows counter in audit log is the same as the counter in

    // doris metrics.

    // ScanBytes is the uncompressed bytes read from local + remote

    // bytes_read_from_local is the compressed bytes read from local

    // bytes_read_from_remote is the compressed bytes read from remote

    // scan bytes > bytes_read_from_local + bytes_read_from_remote

    _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_rows(stats.raw_rows_read);

    _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes(

            stats.uncompressed_bytes_read);

    // In case of no cache, we still need to update the IO stats. uncompressed bytes read == local + remote

    if (stats.file_cache_stats.bytes_read_from_local == 0 &&

        stats.file_cache_stats.bytes_read_from_remote == 0) {

        _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_local_storage(

                stats.compressed_bytes_read);

        DorisMetrics::instance()->query_scan_bytes_from_local->increment(

                stats.compressed_bytes_read);

    } else {

        _state->get_query_ctx()->resource_ctx()->io_context()->update_scan_bytes_from_local_storage(

                stats.file_cache_stats.bytes_read_from_local);