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

#include <memory>

#include <set>

#include "agent/be_exec_version_manager.h"

#include "common/status.h"

#include "core/block/block.h"

#include "exprs/score_runtime.h"

#include "exprs/vexpr.h"

#include "io/io_common.h"

#include "runtime/runtime_state.h"

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

#include "storage/olap_common.h"

#include "storage/predicate/block_column_predicate.h"

#include "storage/predicate/column_predicate.h"

#include "storage/row_cursor.h"

#include "storage/segment/row_ranges.h"

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

#include "storage/tablet/tablet_schema.h"

namespace doris {

class ReadSchema;

class ColumnPredicate;

struct IteratorRowRef;

class StorageReadOptions {

public:

    // Every storage read needs a statistics sink. Deleting the default constructor keeps copied

    // options valid from creation; for example, SegmentIterator stores a copy and later

    // dereferences stats while reading pages.

    StorageReadOptions() = delete;

    explicit StorageReadOptions(OlapReaderStatistics& stats_) noexcept : stats(&stats_) {}

    // Query reads use their RuntimeState; background reads without one use the newest version.

    int get_be_exec_version() const {

        return runtime_state != nullptr ? runtime_state->be_exec_version()

                                        : BeExecVersionManager::get_newest_version();

    }

    struct KeyRange {

        KeyRange()

                : lower_key(nullptr),

                  include_lower(false),

                  upper_key(nullptr),

                  include_upper(false) {}

        KeyRange(const RowCursor* lower_key_, bool include_lower_, const RowCursor* upper_key_,

                 bool include_upper_)

                : lower_key(lower_key_),

                  include_lower(include_lower_),

                  upper_key(upper_key_),

                  include_upper(include_upper_) {}

        // the lower bound of the range, nullptr if not existed

        const RowCursor* lower_key = nullptr;

        // whether `lower_key` is included in the range

        bool include_lower;

        // the upper bound of the range, nullptr if not existed

        const RowCursor* upper_key = nullptr;

        // whether `upper_key` is included in the range

        bool include_upper;

        uint64_t get_digest(uint64_t seed) const {

            if (lower_key != nullptr) {

                auto key_str = lower_key->to_string();

                seed = HashUtil::hash64(key_str.c_str(), key_str.size(), seed);

                seed = HashUtil::hash64(&include_lower, sizeof(include_lower), seed);

            }

            if (upper_key != nullptr) {

                auto key_str = upper_key->to_string();

                seed = HashUtil::hash64(key_str.c_str(), key_str.size(), seed);

                seed = HashUtil::hash64(&include_upper, sizeof(include_upper), seed);

            }

            return seed;

        }

    };

    // reader's key ranges, empty if not existed.

    // used by short key index to filter row blocks

    std::vector<KeyRange> key_ranges;

    // For unique-key merge-on-write, the effect is similar to delete_conditions

    // that filters out rows that are deleted in realtime.

    // For a particular row, if delete_bitmap.contains(rowid) means that row is

    // marked deleted and invisible to user anymore.

    // segment_id -> roaring::Roaring*

    std::unordered_map<uint32_t, std::shared_ptr<roaring::Roaring>> delete_bitmap;

    std::shared_ptr<AndBlockColumnPredicate> delete_condition_predicates =

            AndBlockColumnPredicate::create_shared();

    // reader's column predicate, nullptr if not existed

    // used to fiter rows in row block

    std::vector<std::shared_ptr<ColumnPredicate>> column_predicates;

    std::unordered_map<int32_t, std::shared_ptr<AndBlockColumnPredicate>> col_id_to_predicates;

    std::unordered_map<int32_t, std::vector<std::shared_ptr<const ColumnPredicate>>>

            del_predicates_for_zone_map;

    TPushAggOp::type push_down_agg_type_opt = TPushAggOp::NONE;

    // Non-key columns whose predicates were proven always true by the segment zone map and then

    // removed from column_predicates. SegmentIterator can skip reading these non-output columns,

    // or COUNT_ON_INDEX columns, because the column values no longer affect filtering or counting.

    std::set<uint32_t> zonemap_always_true_pred_cols;

    // Non-owning and initialized to a valid object by the constructor.

    OlapReaderStatistics* stats;

    bool use_page_cache = false;

    uint32_t block_row_max = 4096 - 32; // see https://github.com/apache/doris/pull/11816

    // Effective adaptive batch size byte budget.

    size_t preferred_block_size_bytes = 8388608UL;

    // Whether the tablet schema this read targets materializes variant subcolumns as extracted

    // columns. A compaction read then takes them as flat leaves.

    bool tablet_has_extracted_variant_columns = false;

    // The compaction output schema's variant path layout; null outside a variant compaction.

    VariantCompactionPathsSPtr variant_compaction_paths;

    bool enable_unique_key_merge_on_write = false;

    std::vector<int> topn_filter_source_node_ids;

    // used for special optimization for query : ORDER BY key DESC LIMIT n

    bool read_orderby_key_reverse = false;

    // For rows with the same key, use ascending order (small-to-large) for tie-breakers.

    // For example, use lower rowset version / segment id first.

    bool use_insert_order_when_same = false;

    bool read_changelog = false;

    int changelog_tso_idx = -1;

    // columns for orderby keys

    std::vector<uint32_t>* read_orderby_key_columns = nullptr;

    io::IOContext io_ctx;

    VExprContextSPtrs common_expr_ctxs_push_down;

    const std::set<int32_t>* output_columns = nullptr;

    // Extra storage key columns that are included only to keep the scan schema

    // aligned with the storage key prefix. SegmentIterator can synthesize

    // placeholders only after proving predicates, delete conditions, and

    // expressions do not need their real values.

    std::set<ColumnId> extra_columns;

    // runtime state

    RuntimeState* runtime_state = nullptr;

    RowsetId rowset_id;

    Version version;

    TsoRange commit_tso;

    int64_t tablet_id = 0;

    // slots that cast may be eliminated in storage layer

    std::map<std::string, DataTypePtr> target_cast_type_for_variants;

    RowRanges row_ranges;

    // Per-segment row budget pushed down from the scanner (topn or general

    // limit). SegmentIterator applies it after predicate/common-expr filtering;

    // _can_opt_limit_reads() only decides whether the pre-filter read can also

    // be capped. 0 disables the optimization.

    size_t read_limit = 0;

    std::map<ColumnId, VExprContextSPtr> virtual_column_exprs;

    std::shared_ptr<segment_v2::AnnTopNRuntime> ann_topn_runtime;

    std::map<int32_t, TColumnAccessPaths> all_access_paths;

    std::map<int32_t, TColumnAccessPaths> predicate_access_paths;

    std::shared_ptr<ScoreRuntime> score_runtime;

    CollectionStatisticsPtr collection_statistics;

    uint64_t condition_cache_digest = 0;

};

struct CompactionSampleInfo {

    int64_t bytes = 0;

    int64_t rows = 0;

    int64_t group_data_size = 0;

    int64_t null_count = 0; // Number of NULL cells in this column group

};

struct BlockWithSameBit {

    Block* block;

    std::vector<bool>& same_bit;

    bool empty() const { return block->rows() == 0; }

};

class RowwiseIterator;

using RowwiseIteratorUPtr = std::unique_ptr<RowwiseIterator>;

class RowwiseIterator {

public:

    RowwiseIterator() = default;

    virtual ~RowwiseIterator() = default;

    // Initialize this iterator and make it ready to read with

    // input options.

    // Input options may contain scan range in which this scan.

    // Return Status::OK() if init successfully,

    // Return other error otherwise

    virtual Status init(const StorageReadOptions& opts) {

        return Status::InternalError("to be implemented, current class: " +

                                     demangle(typeid(*this).name()));

    }

    virtual Status init(const StorageReadOptions& opts, CompactionSampleInfo* sample_info) {

        return Status::InternalError("should not reach here, current class: " +

                                     demangle(typeid(*this).name()));

    }

    // If there is any valid data, this function will load data

    // into input batch with Status::OK() returned

    // If there is no data to read, will return Status::EndOfFile.

    // If other error happens, other error code will be returned.

    virtual Status next_batch(Block* block) {

        return Status::InternalError("should not reach here, current class: " +

                                     demangle(typeid(*this).name()));

    }

    virtual Status next_batch(BlockWithSameBit* block_with_same_bit) {

        return Status::InternalError("should not reach here, current class: " +

                                     demangle(typeid(*this).name()));

    }

    virtual Status next_batch(BlockView* block_view) {

        return Status::InternalError("should not reach here, current class: " +

                                     demangle(typeid(*this).name()));

    }

    virtual Status next_row(IteratorRowRef* ref) {

        return Status::InternalError("should not reach here, current class: " +

                                     demangle(typeid(*this).name()));

    }

    virtual Status unique_key_next_row(IteratorRowRef* ref) {

        return Status::InternalError("should not reach here, current class: " +

                                     demangle(typeid(*this).name()));

    }

    virtual bool is_merge_iterator() const { return false; }

    // return schema for this Iterator

    virtual const ReadSchema& schema() const = 0;

    // Return the data id such as segment id, used for keep the insert order when do

    // merge sort in priority queue

    virtual uint64_t data_id() const { return 0; }

    virtual void update_profile(RuntimeProfile* profile) {}

    // return rows merged count by iterator

    virtual uint64_t merged_rows() const { return 0; }

    // return if it's an empty iterator

    virtual bool empty() const { return false; }

};

} // namespace doris