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

#pragma once

#include <gen_cpp/Descriptors_types.h>

#include <gen_cpp/segment_v2.pb.h>

#include <glog/logging.h>

#include <sys/types.h>

#include <cstddef> // for size_t

#include <cstdint> // for uint32_t

#include <functional>

#include <map>

#include <memory> // for unique_ptr

#include <string>

#include <utility>

#include <vector>

#include "common/compiler_util.h"

#include "common/config.h"

#include "common/exception.h"

#include "common/logging.h"

#include "common/status.h"            // for Status

#include "core/column/column_array.h" // ColumnArray

#include "core/data_type/data_type.h"

#include "core/data_type/storage_field_type.h"

#include "io/cache/cached_remote_file_reader.h"

#include "io/fs/file_reader_writer_fwd.h"

#include "io/io_common.h"

#include "storage/index/index_reader.h"

#include "storage/index/ordinal_page_index.h" // for OrdinalPageIndexIterator

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

#include "storage/olap_common.h"

#include "storage/predicate/column_predicate.h"

#include "storage/segment/column_iterator_options.h"

#include "storage/segment/common.h"

#include "storage/segment/page_handle.h" // for PageHandle

#include "storage/segment/page_pointer.h"

#include "storage/segment/parsed_page.h" // for ParsedPage

#include "storage/segment/row_ranges.h"

#include "storage/segment/segment_prefetcher.h"

#include "storage/segment/stream_reader.h"

#include "storage/tablet/tablet_schema.h"

#include "storage/types.h"

#include "storage/utils.h"

#include "util/once.h"

namespace doris {

class BlockCompressionCodec;

class AndBlockColumnPredicate;

class ColumnPredicate;

class TabletIndex;

class StorageReadOptions;

namespace io {

class FileReader;

} // namespace io

struct Slice;

struct StringRef;

using TColumnAccessPaths = std::vector<TColumnAccessPath>;

namespace segment_v2 {

class EncodingInfo;

class ColumnIterator;

class BloomFilterIndexReader;

class InvertedIndexIterator;

class InvertedIndexReader;

class IndexFileReader;

class PageDecoder;

class RowRanges;

class ZoneMapIndexReader;

class IndexIterator;

class ColumnMetaAccessor;

class ColumnReadAhead;

class SegmentReadAhead;

struct ColumnReadAheadContext;

struct ColumnReadAheadRequest;

struct ColumnReadAheadPlan;

struct ColumnReaderOptions {

    // whether verify checksum when read page

    bool verify_checksum = true;

    // for in memory olap table, use DURABLE CachePriority in page cache

    bool kept_in_memory = false;

    int be_exec_version = -1;

    TabletSchemaSPtr tablet_schema = nullptr;

};

class ColumnIterator;

class OffsetFileColumnIterator;

class FileColumnIterator;

using ColumnIteratorUPtr = std::unique_ptr<ColumnIterator>;

using OffsetFileColumnIteratorUPtr = std::unique_ptr<OffsetFileColumnIterator>;

using FileColumnIteratorUPtr = std::unique_ptr<FileColumnIterator>;

using ColumnIteratorSPtr = std::shared_ptr<ColumnIterator>;

// There will be concurrent users to read the same column. So

// we should do our best to reduce resource usage through share

// same information, such as OrdinalPageIndex and Page data.

// This will cache data shared by all reader

class ColumnReader : public MetadataAdder<ColumnReader>,

                     public std::enable_shared_from_this<ColumnReader> {

public:

    ColumnReader();

    // Create an initialized ColumnReader in *reader.

    // This should be a lightweight operation without I/O.

    static Status create(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,

                         uint64_t num_rows, const io::FileReaderSPtr& file_reader,

                         std::shared_ptr<ColumnReader>* reader);

    static Status create_array(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,

                               const io::FileReaderSPtr& file_reader,

                               std::shared_ptr<ColumnReader>* reader);

    static Status create_map(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,

                             const io::FileReaderSPtr& file_reader,

                             std::shared_ptr<ColumnReader>* reader);

    static Status create_struct(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,

                                uint64_t num_rows, const io::FileReaderSPtr& file_reader,

                                std::shared_ptr<ColumnReader>* reader);

    static Status create_agg_state(const ColumnReaderOptions& opts, const ColumnMetaPB& meta,

                                   uint64_t num_rows, const io::FileReaderSPtr& file_reader,

                                   std::shared_ptr<ColumnReader>* reader);

    enum class DictEncodingType { UNKNOWN_DICT_ENCODING, PARTIAL_DICT_ENCODING, ALL_DICT_ENCODING };

    static bool is_compaction_reader_type(ReaderType type);

    ~ColumnReader() override;

    // create a new column iterator. Client should delete returned iterator

    virtual Status new_iterator(ColumnIteratorUPtr* iterator, const TabletColumn* col,

                                const StorageReadOptions*);

    Status new_iterator(ColumnIteratorUPtr* iterator, const TabletColumn* tablet_column);

    Status new_array_iterator(ColumnIteratorUPtr* iterator, const TabletColumn* tablet_column);

    Status new_struct_iterator(ColumnIteratorUPtr* iterator, const TabletColumn* tablet_column);

    Status new_map_iterator(ColumnIteratorUPtr* iterator, const TabletColumn* tablet_column);

    Status new_agg_state_iterator(ColumnIteratorUPtr* iterator);

    virtual Status new_index_iterator(const std::shared_ptr<IndexFileReader>& index_file_reader,

                                      const TabletIndex* index_meta, const std::string& rowset_id,

                                      uint32_t segment_id, size_t rows_of_segment,

                                      std::unique_ptr<IndexIterator>* iterator);

    Status seek_at_or_before(ordinal_t ordinal, OrdinalPageIndexIterator* iter,

                             const ColumnIteratorOptions& iter_opts);

    Status get_ordinal_index_reader(OrdinalIndexReader*& reader,

                                    OlapReaderStatistics* index_load_stats);

    // read a page from file into a page handle

    Status read_page(const ColumnIteratorOptions& iter_opts, const PagePointer& pp,

                     PageHandle* handle, Slice* page_body, PageFooterPB* footer,

                     BlockCompressionCodec* codec) const;

    bool is_nullable() const { return _meta_is_nullable; }

    const EncodingInfo* encoding_info() const { return _encoding_info; }

    virtual bool has_zone_map() const { return _zone_map_index != nullptr; }

    bool has_bloom_filter_index(bool ngram) const;

    // Check if this column could match `cond' using segment zone map.

    // Since segment zone map is stored in metadata, this function is fast without I/O.

    // set matched to true if segment zone map is absent or `cond' could be satisfied, false otherwise.

    virtual Status match_condition(const AndBlockColumnPredicate* col_predicates,

                                   bool* matched) const;

    Status next_batch_of_zone_map(size_t* n, MutableColumnPtr& dst) const;

    // get row ranges with zone map

    // - cond_column is user's query predicate

    // - delete_condition is a delete predicate of one version

    Status get_row_ranges_by_zone_map(

            const AndBlockColumnPredicate* col_predicates,

            const std::vector<std::shared_ptr<const ColumnPredicate>>* delete_predicates,

            RowRanges* row_ranges, const ColumnIteratorOptions& iter_opts);

    // get row ranges with bloom filter index

    Status get_row_ranges_by_bloom_filter(const AndBlockColumnPredicate* col_predicates,

                                          RowRanges* row_ranges,

                                          const ColumnIteratorOptions& iter_opts);

    PagePointer get_dict_page_pointer() const { return _meta_dict_page; }

    bool is_empty() const { return _num_rows == 0; }

    Status prune_predicates_by_zone_map(std::vector<std::shared_ptr<ColumnPredicate>>& predicates,

                                        const int column_id, bool* pruned) const;

    virtual Status get_segment_zone_map(segment_v2::ZoneMap* zone_map) const;

    Status get_page_zone_maps(const ColumnIteratorOptions& iter_opts,

                              const std::vector<ZoneMapPB>** zone_maps);

    Status get_row_range_for_page(uint32_t page_index, const ColumnIteratorOptions& iter_opts,

                                  RowRange* row_range);

    CompressionTypePB get_compression() const { return _meta_compression; }

    uint64_t num_rows() const { return _num_rows; }

    void set_dict_encoding_type(DictEncodingType type) {

        static_cast<void>(_set_dict_encoding_type_once.call([&] {

            _dict_encoding_type = type;

            return Status::OK();

        }));

    }

    DictEncodingType get_dict_encoding_type() { return _dict_encoding_type; }

    void disable_index_meta_cache() { _use_index_page_cache = false; }

    DataTypePtr get_vec_data_type() { return _data_type; }

    virtual FieldType get_meta_type() { return _meta_type; }

    int64_t get_metadata_size() const override;

#ifdef BE_TEST

    void check_data_by_zone_map_for_test(const MutableColumnPtr& dst) const;

#endif

private:

    friend class VariantColumnReader;

    friend class FileColumnIterator;

    friend class SegmentPrefetcher;

    ColumnReader(const ColumnReaderOptions& opts, const ColumnMetaPB& meta, uint64_t num_rows,

                 io::FileReaderSPtr file_reader);

    Status init(const ColumnMetaPB* meta);

    [[nodiscard]] Status _load_zone_map_index(bool use_page_cache, bool kept_in_memory,

                                              const ColumnIteratorOptions& iter_opts);

    [[nodiscard]] Status _load_ordinal_index(bool use_page_cache, bool kept_in_memory,

                                             const ColumnIteratorOptions& iter_opts);

    [[nodiscard]] Status _load_index(const std::shared_ptr<IndexFileReader>& index_file_reader,

                                     const TabletIndex* index_meta, const std::string& rowset_id,

                                     uint32_t segment_id, size_t rows_of_segment);

    [[nodiscard]] Status _load_bloom_filter_index(bool use_page_cache, bool kept_in_memory,

                                                  const ColumnIteratorOptions& iter_opts);

    bool _zone_map_match_condition(const segment_v2::ZoneMap& zone_map,

                                   const AndBlockColumnPredicate* col_predicates) const;

    Status _get_filtered_pages(

            const AndBlockColumnPredicate* col_predicates,

            const std::vector<std::shared_ptr<const ColumnPredicate>>* delete_predicates,

            std::vector<uint32_t>* page_indexes, const ColumnIteratorOptions& iter_opts);

    Status _calculate_row_ranges(const std::vector<uint32_t>& page_indexes, RowRanges* row_ranges,

                                 const ColumnIteratorOptions& iter_opts);

    int64_t _meta_length;

    FieldType _meta_type;

    FieldType _meta_children_column_type;

    bool _meta_is_nullable;

    bool _use_index_page_cache;

    int _be_exec_version = -1;

    PagePointer _meta_dict_page;

    CompressionTypePB _meta_compression;

    ColumnReaderOptions _opts;

    uint64_t _num_rows;

    io::FileReaderSPtr _file_reader;

    DictEncodingType _dict_encoding_type;

    DataTypePtr _data_type;

    FieldType _type =

            FieldType::OLAP_FIELD_TYPE_NONE; // initialized in init(), may changed by subclasses.

    const EncodingInfo* _encoding_info =

            nullptr; // initialized in init(), used for create PageDecoder

    // meta for various column indexes (null if the index is absent)

    std::unique_ptr<ZoneMapPB> _segment_zone_map;

    mutable std::shared_mutex _load_index_lock;

    std::unique_ptr<ZoneMapIndexReader> _zone_map_index;

    std::unique_ptr<OrdinalIndexReader> _ordinal_index;

    std::shared_ptr<BloomFilterIndexReader> _bloom_filter_index;

    std::unordered_map<int64_t, IndexReaderPtr> _index_readers;

    std::vector<std::shared_ptr<ColumnReader>> _sub_readers;

    DorisCallOnce<Status> _set_dict_encoding_type_once;

};

// Base iterator to read one column data

class ColumnIterator {

public:

    ColumnIterator() = default;

    virtual ~ColumnIterator() = default;

    virtual Status init(const ColumnIteratorOptions& opts) {

        _opts = opts;

        return Status::OK();

    }

    // Seek to the given ordinal entry in the column.

    // Entry 0 is the first entry written to the column.

    // If provided seek point is past the end of the file,

    // then returns false.

    virtual Status seek_to_ordinal(ordinal_t ord) = 0;

    Status next_batch(size_t* n, MutableColumnPtr& dst) {

        bool has_null;

        return next_batch(n, dst, &has_null);

    }

    virtual Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) {

        return Status::NotSupported("next_batch not implement");

    }

    virtual Status next_batch_of_zone_map(size_t* n, MutableColumnPtr& dst) {

        return Status::NotSupported("next_batch_of_zone_map not implement");

    }

    virtual Status read_by_rowids(const rowid_t* rowids, const size_t count,

                                  MutableColumnPtr& dst) {

        return Status::NotSupported("read_by_rowids not implement");

    }

    virtual ordinal_t get_current_ordinal() const = 0;

    virtual Status get_row_ranges_by_zone_map(

            const AndBlockColumnPredicate* col_predicates,

            const std::vector<std::shared_ptr<const ColumnPredicate>>* delete_predicates,

            RowRanges* row_ranges) {

        return Status::OK();

    }

    virtual Status get_row_ranges_by_bloom_filter(const AndBlockColumnPredicate* col_predicates,

                                                  RowRanges* row_ranges) {

        return Status::OK();

    }

    virtual Status get_row_ranges_by_dict(const AndBlockColumnPredicate* col_predicates,

                                          RowRanges* row_ranges) {

        return Status::OK();

    }

    virtual bool is_all_dict_encoding() const { return false; }

    virtual Status set_access_paths(const TColumnAccessPaths& all_access_paths,

                                    const TColumnAccessPaths& predicate_access_paths) {

        if (!predicate_access_paths.empty()) {

            set_read_requirement_self(ReadRequirement::PREDICATE);

        }

        return Status::OK();

    }

    void set_column_name(const std::string& column_name) { _column_name = column_name; }

    const std::string& column_name() const { return _column_name; }

    // Per-iterator read requirement derived from nested access paths.

    //

    // The ordering is intentional and used by set_read_requirement_self(): requirements are

    // monotonic and a weaker requirement must not downgrade a stronger one.

    // - NORMAL: no pruning decision has been made yet.

    // - SKIP: this iterator should not be read.

    // - LAZY_OUTPUT: materialize this iterator in the lazy phase after predicate filtering.

    // - PREDICATE: read this iterator in the predicate phase. This must stay stronger than

    //   LAZY_OUTPUT because parents may mark children as LAZY_OUTPUT after child set_access_paths()

    //   has already promoted predicate-only children to PREDICATE.

    enum class ReadRequirement : int { NORMAL, SKIP, LAZY_OUTPUT, PREDICATE };

    // Set the read requirement on this iterator and all nested child iterators.

    virtual void set_read_requirement(ReadRequirement requirement) {

        set_read_requirement_self(requirement);

    }

    ReadRequirement read_requirement() const { return _read_requirement; }

    virtual void set_lazy_output_requirement() {

        set_read_requirement(ReadRequirement::LAZY_OUTPUT);

    }

    virtual void remove_pruned_sub_iterators() {};

    virtual Status init_prefetcher(const SegmentPrefetchParams& params) { return Status::OK(); }

    /// Appends data-page plans for scalar columns. Complex iterators keep the original read path.

    virtual Status prepare_read_ahead(const ColumnReadAheadRequest& request,

                                      std::vector<ColumnReadAheadPlan>* plans) {

        return Status::OK();

    }

    virtual void collect_prefetchers(

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

            PrefetcherInitMethod init_method) {}

    static constexpr const char* ACCESS_OFFSET = "OFFSET";

    static constexpr const char* ACCESS_ALL = "*";

    static constexpr const char* ACCESS_MAP_KEYS = "KEYS";

    static constexpr const char* ACCESS_MAP_VALUES = "VALUES";

    static constexpr const char* ACCESS_NULL = "NULL";

    // Meta-only read modes:

    // - OFFSET_ONLY: read offsets while skipping actual child/string data. For nullable

    //   complex columns, the parent null map is still materialized when needed.

    // - NULL_MAP_ONLY: only read null map (e.g., for IS NULL / IS NOT NULL predicates)

    // When these modes are enabled, actual content data is skipped.

    enum class MetaReadMode : int { DEFAULT, OFFSET_ONLY, NULL_MAP_ONLY };

    bool read_offset_only() const { return _meta_read_mode == MetaReadMode::OFFSET_ONLY; }

    bool read_null_map_only() const { return _meta_read_mode == MetaReadMode::NULL_MAP_ONLY; }

    // The current scanner phase. This is intentionally separate from ReadRequirement

    // (why this iterator is needed) and MetaReadMode (what physical metadata to read).

    enum class ReadPhase : int {

        NORMAL,    // default full materialization without lazy read split

        PREDICATE, // predicate evaluation before row filtering

        LAZY       // post-filter lazy materialization

    };

    virtual void set_read_phase(ReadPhase mode) {

        _read_phase = mode;

        if (mode == ReadPhase::PREDICATE) {

            _has_place_holder_column = false;

        }

    }

    virtual bool need_to_read() const {

        switch (_read_phase) {

        case ReadPhase::NORMAL:

            return _read_requirement != ReadRequirement::SKIP;

        case ReadPhase::PREDICATE:

            return _read_requirement == ReadRequirement::PREDICATE;

        case ReadPhase::LAZY:

            return _read_requirement == ReadRequirement::LAZY_OUTPUT;

        default:

            return false;

        }

    }

    // Whether the current iterator itself should materialize meta columns, such as

    // the null-map column or the offset column, into the destination column.

    //

    // Do not use the virtual need_to_read() here. Complex iterators override

    // need_to_read() in LAZY mode to keep the parent iterator active when only a

    // nested child still has data to materialize. That parent-level control-flow

    // decision is different from materializing the parent's own offsets/null-map:

    // if the parent was already read for predicate evaluation, LAZY mode should

    // only fill the missing children and must not append parent meta again.

    bool need_to_read_meta_columns() const { return ColumnIterator::need_to_read(); }

    virtual void finalize_lazy_phase(MutableColumnPtr& dst) {

        _recovery_from_place_holder_column(dst);

    }

    // Set only this iterator's requirement without modifying requirements of any nested child

    // iterators. Use this when the parent/wrapper state must be updated while child requirements

    // are decided independently.

    virtual void set_read_requirement_self(ReadRequirement requirement) {

        if (static_cast<int>(requirement) > static_cast<int>(_read_requirement)) {

            _read_requirement = requirement;

        }

    }

    // Whether this iterator or any nested iterator has data that must be materialized

    // in lazy mode. Predicate-only branches are read before filtering and must not be

    // re-read in the lazy phase. Meta-only access paths still become lazy targets when

    // they appear only in all_access_paths, because OFFSET/NULL is the requested output.

    virtual bool has_lazy_read_target() const {

        return _read_requirement == ReadRequirement::LAZY_OUTPUT;

    }

protected:

    struct AccessPathSplit {

        TColumnAccessPaths descendant_paths;

        bool reads_current_data = false;

        MetaReadMode current_meta_mode = MetaReadMode::DEFAULT;

        bool has_descendant_paths() const { return !descendant_paths.empty(); }

    };

    // Nested columns share the same current-level access-path planning, while their data-child

    // topology and descendant routing remain container-specific.

    struct NestedAccessPathPlan {

        AccessPathSplit all;

        AccessPathSplit predicate;

        bool skip_data_descendants = false;

    };

    // At their current level, Struct supports null-map metadata. Map and Array additionally

    // support offsets.

    enum class NestedMetaSupport { NULL_MAP, NULL_MAP_AND_OFFSET };

    void _convert_to_place_holder_column(MutableColumnPtr& dst, size_t count);

    void _recovery_from_place_holder_column(MutableColumnPtr& dst);

    // Derive current-level meta-only read mode from an explicit access-path split. Meta-only is

    // valid only when this iterator had no data-read requirement before applying the current paths,

    // no current DATA path exists, and no path must be routed to a descendant iterator.

    Status _check_and_set_meta_read_mode(ReadRequirement requirement_before_access_path,

                                         const AccessPathSplit& all_access_paths);

    // Apply the common current-level access-path state transitions and select a supported

    // parent-owned meta-only mode. When that mode skips data descendants, synchronously invoke the

    // callback once with SKIP before returning the routing plan. The callback is never retained.

    Result<NestedAccessPathPlan> _prepare_nested_access_paths(

            const TColumnAccessPaths& all_access_paths,

            const TColumnAccessPaths& predicate_access_paths, NestedMetaSupport meta_support,

            const std::function<void(ReadRequirement)>& set_all_data_descendants_read_requirement);

    // Normalize the wire encoding, strip this iterator's column name, and explicitly partition

    // paths consumed by this iterator from paths that must be routed to descendants. This helper is

    // intentionally side-effect free; callers apply DATA/predicate read requirements explicitly.

    // owns_offset_meta tells whether this iterator has current-level offsets. When it does not

    // (Struct), a legacy OFFSET tail is a data field named "offset" and stays a descendant path,

    // and a typed META OFFSET request is an FE/BE contract violation.

    Result<AccessPathSplit> _split_access_paths(TColumnAccessPaths access_paths,

                                                bool owns_offset_meta) const;

    ColumnIteratorOptions _opts;

    ReadRequirement _read_requirement {ReadRequirement::NORMAL};

    MetaReadMode _meta_read_mode = MetaReadMode::DEFAULT;

    ReadPhase _read_phase {ReadPhase::NORMAL};

    std::string _column_name;

    bool _has_place_holder_column {false};

};

// This iterator is used to read column data from file

// for scalar type

class FileColumnIterator : public ColumnIterator {

public:

    explicit FileColumnIterator(std::shared_ptr<ColumnReader> reader);

    ~FileColumnIterator() override;

    Status init(const ColumnIteratorOptions& opts) override;

    Status seek_to_ordinal(ordinal_t ord) override;

    Status seek_to_page_start();

    Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) override;

    Status next_batch_of_zone_map(size_t* n, MutableColumnPtr& dst) override;

    Status read_by_rowids(const rowid_t* rowids, const size_t count,

                          MutableColumnPtr& dst) override;

    Status set_access_paths(const TColumnAccessPaths& all_access_paths,

                            const TColumnAccessPaths& predicate_access_paths) override;

    ordinal_t get_current_ordinal() const override { return _current_ordinal; }

    // get row ranges by zone map

    // - cond_column is user's query predicate

    // - delete_condition is delete predicate of one version

    Status get_row_ranges_by_zone_map(

            const AndBlockColumnPredicate* col_predicates,

            const std::vector<std::shared_ptr<const ColumnPredicate>>* delete_predicates,

            RowRanges* row_ranges) override;

    Status get_row_ranges_by_bloom_filter(const AndBlockColumnPredicate* col_predicates,

                                          RowRanges* row_ranges) override;

    Status get_row_ranges_by_dict(const AndBlockColumnPredicate* col_predicates,

                                  RowRanges* row_ranges) override;

    ParsedPage* get_current_page() { return &_page; }

    bool is_nullable() { return _reader->is_nullable(); }

    bool is_all_dict_encoding() const override { return _is_all_dict_encoding; }

    Status init_prefetcher(const SegmentPrefetchParams& params) override;

    Status prepare_read_ahead(const ColumnReadAheadRequest& request,

                              std::vector<ColumnReadAheadPlan>* plans) override;

    void collect_prefetchers(

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

            PrefetcherInitMethod init_method) override;

protected:

    // Exposed to derived iterators (e.g. StringFileColumnIterator) so they can

    // query column metadata such as the storage field type.

    const std::shared_ptr<ColumnReader>& get_reader() const { return _reader; }

private:

    Status _seek_to_pos_in_page(ParsedPage* page, ordinal_t offset_in_page) const;

    Status _load_next_page(bool* eos);

    Status _read_data_page(const OrdinalPageIndexIterator& iter);

    Status _read_dict_data();

    // Materialize compressed data-page ranges from the ordinal index once. Later calls must keep

    // the same window options and scan direction for this iterator.

    Status _init_read_ahead(const ColumnReadAheadRequest& request);

    // Run before PageIO so both buffered reads and Page Cache hits advance the scan window.

    void _advance_read_ahead(int32_t page_index);

    void _trigger_prefetch_if_eligible(ordinal_t ord);

    std::shared_ptr<ColumnReader> _reader = nullptr;

    BlockCompressionCodec* _compress_codec = nullptr;

    // 1. The _page represents current page.

    // 2. We define an operation is one seek and following read,

    //    If new seek is issued, the _page will be reset.

    ParsedPage _page;

    // keep dict page decoder

    std::unique_ptr<PageDecoder> _dict_decoder;

    // keep dict page handle to avoid released

    PageHandle _dict_page_handle;

    // page iterator used to get next page when current page is finished.

    // This value will be reset when a new seek is issued

    OrdinalPageIndexIterator _page_iter;

    // current value ordinal

    ordinal_t _current_ordinal = 0;

    bool _is_all_dict_encoding = false;

    std::unique_ptr<StringRef[]> _dict_word_info;

    bool _enable_prefetch {false};

    std::unique_ptr<SegmentPrefetcher> _prefetcher;

    std::shared_ptr<io::CachedRemoteFileReader> _cached_remote_file_reader {nullptr};

    std::unique_ptr<ColumnReadAhead> _read_ahead;

    // Set only for page-driven scans; owned by the enclosing SegmentIterator.

    SegmentReadAhead* _read_ahead_segment {nullptr};

};

class EmptyFileColumnIterator final : public ColumnIterator {

public:

    Status seek_to_ordinal(ordinal_t ord) override { return Status::OK(); }

    ordinal_t get_current_ordinal() const override { return 0; }

};

// StringFileColumnIterator extends FileColumnIterator's NULL metadata support with OFFSET-only

// reading for string/binary column types. When the OFFSET path is detected in set_access_paths, it

// sets only_read_offsets on the ColumnIteratorOptions so that the BinaryPlainPageDecoder skips

// chars memcpy and only fills offsets.

class StringFileColumnIterator final : public FileColumnIterator {

public:

    explicit StringFileColumnIterator(std::shared_ptr<ColumnReader> reader);

    ~StringFileColumnIterator() override = default;

    Status init(const ColumnIteratorOptions& opts) override;

    Status set_access_paths(const TColumnAccessPaths& all_access_paths,

                            const TColumnAccessPaths& predicate_access_paths) override;

};

// This iterator make offset operation write once for

class OffsetFileColumnIterator final : public ColumnIterator {

public:

    explicit OffsetFileColumnIterator(FileColumnIteratorUPtr offset_reader) {

        _offset_iterator = std::move(offset_reader);

    }

    ~OffsetFileColumnIterator() override = default;

    Status init(const ColumnIteratorOptions& opts) override;

    Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) override;

    Status next_batch(size_t* n, MutableColumnPtr& dst) {

        bool has_null;

        return next_batch(n, dst, &has_null);

    }

    ordinal_t get_current_ordinal() const override {

        return _offset_iterator->get_current_ordinal();

    }

    Status seek_to_ordinal(ordinal_t ord) override {

        RETURN_IF_ERROR(_offset_iterator->seek_to_ordinal(ord));

        return Status::OK();

    }

    Status _peek_one_offset(ordinal_t* offset);

    Status _calculate_offsets(ssize_t start, ColumnArray::ColumnOffsets& column_offsets);

    Status read_by_rowids(const rowid_t* rowids, const size_t count,

                          MutableColumnPtr& dst) override {

        return _offset_iterator->read_by_rowids(rowids, count, dst);

    }

    void set_read_requirement(ReadRequirement requirement) override {

        set_read_requirement_self(requirement);

        _offset_iterator->set_read_requirement(requirement);

    }

    Status init_prefetcher(const SegmentPrefetchParams& params) override;

    void collect_prefetchers(

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

            PrefetcherInitMethod init_method) override;

private:

    std::unique_ptr<FileColumnIterator> _offset_iterator;

    // reuse a tiny column for peek to avoid frequent allocations

    MutableColumnPtr _peek_tmp_col;

};

// This iterator is used to read map value column

class MapFileColumnIterator final : public ColumnIterator {

public:

    explicit MapFileColumnIterator(std::shared_ptr<ColumnReader> reader,

                                   ColumnIteratorUPtr null_iterator,

                                   OffsetFileColumnIteratorUPtr offsets_iterator,

                                   ColumnIteratorUPtr key_iterator,

                                   ColumnIteratorUPtr val_iterator);

    ~MapFileColumnIterator() override = default;

    Status init(const ColumnIteratorOptions& opts) override;

    Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) override;

    Status read_by_rowids(const rowid_t* rowids, const size_t count,

                          MutableColumnPtr& dst) override;

    Status seek_to_ordinal(ordinal_t ord) override;

    ordinal_t get_current_ordinal() const override {

        if (read_null_map_only() && _null_iterator) {

            return _null_iterator->get_current_ordinal();

        }

        return _offsets_iterator->get_current_ordinal();

    }

    Status init_prefetcher(const SegmentPrefetchParams& params) override;

    void collect_prefetchers(

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

            PrefetcherInitMethod init_method) override;

    Status set_access_paths(const TColumnAccessPaths& all_access_paths,

                            const TColumnAccessPaths& predicate_access_paths) override;

    void set_lazy_output_requirement() override;

    void remove_pruned_sub_iterators() override;

    void set_read_phase(ReadPhase mode) override;

    bool need_to_read() const override {

        switch (_read_phase) {

        case ReadPhase::NORMAL:

            return _read_requirement != ReadRequirement::SKIP;

        case ReadPhase::PREDICATE:

            return _read_requirement == ReadRequirement::PREDICATE;

        case ReadPhase::LAZY:

            // In lazy mode, read this map only when at least one key/value branch still

            // has non-predicate data to materialize.

            return has_lazy_read_target();

        default:

            return false;

        }

    }

    void finalize_lazy_phase(MutableColumnPtr& dst) override;

    void set_read_requirement(ReadRequirement requirement) override;

    bool has_lazy_read_target() const override;

private:

    std::shared_ptr<ColumnReader> _map_reader = nullptr;

    ColumnIteratorUPtr _null_iterator;

    OffsetFileColumnIteratorUPtr _offsets_iterator; //OffsetFileIterator

    ColumnIteratorUPtr _key_iterator;

    ColumnIteratorUPtr _val_iterator;

};

class StructFileColumnIterator final : public ColumnIterator {

public:

    explicit StructFileColumnIterator(std::shared_ptr<ColumnReader> reader,

                                      ColumnIteratorUPtr null_iterator,

                                      std::vector<ColumnIteratorUPtr>&& sub_column_iterators);

    ~StructFileColumnIterator() override = default;

    Status init(const ColumnIteratorOptions& opts) override;

    Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) override;

    Status next_batch(size_t* n, MutableColumnPtr& dst) {

        bool has_null;

        return next_batch(n, dst, &has_null);

    }

    Status read_by_rowids(const rowid_t* rowids, const size_t count,

                          MutableColumnPtr& dst) override;

    Status seek_to_ordinal(ordinal_t ord) override;

    ordinal_t get_current_ordinal() const override {

        if (read_null_map_only() && _null_iterator) {

            return _null_iterator->get_current_ordinal();

        }

        return _sub_column_iterators[0]->get_current_ordinal();

    }

    Status set_access_paths(const TColumnAccessPaths& all_access_paths,

                            const TColumnAccessPaths& predicate_access_paths) override;

    void set_lazy_output_requirement() override;

    void remove_pruned_sub_iterators() override;

    Status init_prefetcher(const SegmentPrefetchParams& params) override;

    void collect_prefetchers(

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

            PrefetcherInitMethod init_method) override;

    void set_read_phase(ReadPhase mode) override;

    bool need_to_read() const override {

        switch (_read_phase) {

        case ReadPhase::NORMAL:

            return _read_requirement != ReadRequirement::SKIP;

        case ReadPhase::PREDICATE:

            return _read_requirement == ReadRequirement::PREDICATE;

        case ReadPhase::LAZY:

            // In lazy mode, read this struct only when at least one nested branch still

            // has non-predicate data to materialize.

            return has_lazy_read_target();

        default:

            return false;

        }

    }

    void finalize_lazy_phase(MutableColumnPtr& dst) override;

    void set_read_requirement(ReadRequirement requirement) override;

    bool has_lazy_read_target() const override;

private:

    std::shared_ptr<ColumnReader> _struct_reader = nullptr;

    ColumnIteratorUPtr _null_iterator;

    std::vector<ColumnIteratorUPtr> _sub_column_iterators;

};

class ArrayFileColumnIterator final : public ColumnIterator {

public:

    explicit ArrayFileColumnIterator(std::shared_ptr<ColumnReader> reader,

                                     OffsetFileColumnIteratorUPtr offset_reader,

                                     ColumnIteratorUPtr item_iterator,

                                     ColumnIteratorUPtr null_iterator);

    ~ArrayFileColumnIterator() override = default;

    Status init(const ColumnIteratorOptions& opts) override;

    Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) override;

    Status next_batch(size_t* n, MutableColumnPtr& dst) {

        bool has_null;

        return next_batch(n, dst, &has_null);

    }

    Status read_by_rowids(const rowid_t* rowids, const size_t count,

                          MutableColumnPtr& dst) override;

    Status seek_to_ordinal(ordinal_t ord) override;

    ordinal_t get_current_ordinal() const override {

        if (read_null_map_only() && _null_iterator) {

            return _null_iterator->get_current_ordinal();

        }

        return _offset_iterator->get_current_ordinal();

    }

    Status set_access_paths(const TColumnAccessPaths& all_access_paths,

                            const TColumnAccessPaths& predicate_access_paths) override;

    void set_lazy_output_requirement() override;

    void remove_pruned_sub_iterators() override;

    Status init_prefetcher(const SegmentPrefetchParams& params) override;

    void collect_prefetchers(

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

            PrefetcherInitMethod init_method) override;

    void set_read_phase(ReadPhase mode) override;

    bool need_to_read() const override {

        switch (_read_phase) {

        case ReadPhase::NORMAL:

            return _read_requirement != ReadRequirement::SKIP;

        case ReadPhase::PREDICATE:

            return _read_requirement == ReadRequirement::PREDICATE;

        case ReadPhase::LAZY:

            // In lazy mode, read this array only when its item branch still has

            // non-predicate data to materialize.

            return has_lazy_read_target();

        default:

            return false;

        }

    }

    void finalize_lazy_phase(MutableColumnPtr& dst) override;

    void set_read_requirement(ReadRequirement requirement) override;

    bool has_lazy_read_target() const override;

private:

    std::shared_ptr<ColumnReader> _array_reader = nullptr;

    std::unique_ptr<OffsetFileColumnIterator> _offset_iterator;

    std::unique_ptr<ColumnIterator> _null_iterator;

    std::unique_ptr<ColumnIterator> _item_iterator;

    Status _seek_by_offsets(ordinal_t ord);

};

// Reads COMPACTION_ROWID_COL by synthesizing the physical source RowLocation during compaction.

// Its payload is RowLocation, distinct from the query-facing GlobalRowLoacationV2 format.

class CompactionRowIdColumnIterator : public ColumnIterator {

public:

    CompactionRowIdColumnIterator(RowsetId rowset_id, uint32_t segment_id)

            : _rowset_id(rowset_id), _segment_id(segment_id) {}

    Status seek_to_ordinal(ordinal_t ord_idx) override {

        _current_rowid = cast_set<uint32_t>(ord_idx);

        return Status::OK();

    }

    Status next_batch(size_t* n, MutableColumnPtr& dst) {

        bool has_null;

        return next_batch(n, dst, &has_null);

    }

    Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) override {

        for (size_t i = 0; i < *n; ++i) {

            RowLocation location(_rowset_id, _segment_id, cast_set<uint32_t>(_current_rowid + i));

            dst->insert_data(reinterpret_cast<const char*>(&location), sizeof(RowLocation));

        }

        _current_rowid += *n;

        return Status::OK();

    }

    Status read_by_rowids(const rowid_t* rowids, const size_t count,

                          MutableColumnPtr& dst) override {

        for (size_t i = 0; i < count; ++i) {

            RowLocation location(_rowset_id, _segment_id, cast_set<uint32_t>(rowids[i]));

            dst->insert_data(reinterpret_cast<const char*>(&location), sizeof(RowLocation));

        }

        return Status::OK();

    }

    ordinal_t get_current_ordinal() const override { return _current_rowid; }

private:

    uint32_t _current_rowid = 0;

    RowsetId _rowset_id;

    uint32_t _segment_id = 0;

};

class RowIdColumnIteratorV2 : public ColumnIterator {

public:

    RowIdColumnIteratorV2(uint8_t version, int64_t backend_id, uint32_t file_id)

            : _version(version), _backend_id(backend_id), _file_id(file_id) {}

    Status seek_to_ordinal(ordinal_t ord_idx) override {

        _current_rowid = cast_set<uint32_t>(ord_idx);

        return Status::OK();

    }

    Status next_batch(size_t* n, MutableColumnPtr& dst) {

        bool has_null;

        return next_batch(n, dst, &has_null);

    }

    Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) override;

    Status read_by_rowids(const rowid_t* rowids, const size_t count,

                          MutableColumnPtr& dst) override;

    ordinal_t get_current_ordinal() const override { return _current_rowid; }

private:

    uint32_t _current_rowid = 0;

    uint8_t _version;

    int64_t _backend_id;

    uint32_t _file_id;

};

// Produces a column whose every row is the same request-scoped constant Field value.

// It is created directly by Segment and never stored in ColumnReaderCache.

class ConstantColumnIterator : public ColumnIterator {

public:

    ConstantColumnIterator() = delete;

    explicit ConstantColumnIterator(Field value) : _value(std::move(value)) {}

    Status seek_to_ordinal(ordinal_t ord_idx) override {

        _current_rowid = ord_idx;

        return Status::OK();

    }

    Status next_batch(size_t* n, MutableColumnPtr& dst) {

        bool has_null;

        return next_batch(n, dst, &has_null);

    }

    Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null) override {

        if (!need_to_read()) {

            _convert_to_place_holder_column(dst, *n);

            return Status::OK();

        }

        _recovery_from_place_holder_column(dst);

        *has_null = _value.is_null();

        Status st = _insert_many(dst, *n);

        if (!st.ok()) {

            return st;

        }

        _current_rowid += *n;

        return st;

    }

    Status next_batch_of_zone_map(size_t* n, MutableColumnPtr& dst) override {

        return next_batch(n, dst);

    }

    Status read_by_rowids(const rowid_t* rowids, const size_t count,

                          MutableColumnPtr& dst) override {

        if (!need_to_read()) {

            _convert_to_place_holder_column(dst, count);

            return Status::OK();

        }

        _recovery_from_place_holder_column(dst);

        return _insert_many(dst, count);

    }

    ordinal_t get_current_ordinal() const override { return _current_rowid; }

private:

    Status _insert_many(MutableColumnPtr& dst, size_t n) {

        if (_value.is_null()) {

            if (UNLIKELY(!dst->is_nullable())) {

                return Status::InternalError(

                        "try to apply constant null value to not nullable target column");

            }

            dst->insert_many_defaults(n);

            return Status::OK();

        }

        // Query predicates may allocate ColumnDictI32 for strings. For example, evaluating

        // `city = 'Paris'` on an old segment whose added `city` column defaults to "Paris" must

        // materialize a ColumnString before inserting the default Field.

        dst = dst->convert_to_predicate_column_if_dictionary();

        dst->insert_duplicate_fields(_value, n);

        return Status::OK();

    }

    Field _value;

    ordinal_t _current_rowid = 0;

};

// A request-scoped ColumnReader that represents one constant value for the whole segment instead

// of reading on-disk data. For example, an old segment after `ADD COLUMN city DEFAULT 'Paris'`

// uses [Paris, Paris] as its zonemap and produces "Paris" without caching that request's reader.

class ConstantColumnReader : public ColumnReader {

public:

    explicit ConstantColumnReader(Field value, FieldType meta_type)

            : _value(std::move(value)), _schema_type(meta_type) {

        // A null Field still needs its schema type. For example, an old segment after adding a

        // nullable BIGINT column has value NULL but must report BIGINT to the read pipeline.

        if (_schema_type == FieldType::OLAP_FIELD_TYPE_UNKNOWN ||

            _schema_type == FieldType::OLAP_FIELD_TYPE_NONE) {