// 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 <glog/logging.h>

#include <algorithm>

#include <cstddef>

#include <memory>

#include <string>

#include <unordered_map>

#include <utility>

#include <vector>

#include "common/factory_creator.h"

#include "common/logging.h"

#include "common/status.h"

#include "core/block/block.h"

#include "core/block/column_with_type_and_name.h"

#include "core/column/column.h"

#include "exec/runtime_filter/runtime_filter_selectivity.h"

#include "exprs/expr_zonemap_filter.h"

#include "exprs/function_context.h"

#include "exprs/vexpr_fwd.h"

#include "runtime/runtime_state.h"

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

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

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

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

#include "storage/segment/column_reader.h"

namespace doris {

class RowDescriptor;

class ReadSchema;

using ReadSchemaSPtr = std::shared_ptr<ReadSchema>;

class RuntimeState;

class ZoneMapEvalContext;

} // namespace doris

namespace doris::segment_v2 {

class Segment;

class ColumnIterator;

} // namespace doris::segment_v2

namespace doris {

class ScoreRuntime;

class LambdaExecutionContext;

using ScoreRuntimeSPtr = std::shared_ptr<ScoreRuntime>;

class IndexExecContext {

public:

    IndexExecContext(const std::vector<std::unique_ptr<segment_v2::IndexIterator>>& index_iterators,

                     const std::vector<IndexFieldNameAndTypePair>& storage_name_and_type_vec,

                     std::unordered_map<ColumnId, std::unordered_map<const VExpr*, bool>>&

                             common_expr_index_status,

                     ScoreRuntimeSPtr score_runtime, segment_v2::Segment* segment,

                     const segment_v2::ColumnIteratorOptions& column_iter_opts,

                     ReadSchemaSPtr read_schema)

            : _index_iterators(index_iterators),

              _storage_name_and_type(storage_name_and_type_vec),

              _expr_index_status(common_expr_index_status),

              _score_runtime(std::move(score_runtime)),

              _segment(segment),

              _column_iter_opts(column_iter_opts),

              _read_schema(std::move(read_schema)) {

        DORIS_CHECK(_read_schema != nullptr);

    }

    segment_v2::IndexIterator* get_inverted_index_iterator(int32_t read_ordinal) const {

        if (read_ordinal < 0 || static_cast<size_t>(read_ordinal) >= _index_iterators.size()) {

            return nullptr;

        }

        const auto& iterator = _index_iterators[read_ordinal];

        if (!iterator) {

            return nullptr;

        }

        return iterator.get();

    }

    const IndexFieldNameAndTypePair* get_storage_name_and_type(int32_t read_ordinal) const {

        if (read_ordinal < 0 ||

            static_cast<size_t>(read_ordinal) >= _storage_name_and_type.size()) {

            return nullptr;

        }

        return &_storage_name_and_type[read_ordinal];

    }

    segment_v2::Segment* segment() const { return _segment; }

    const ReadSchemaSPtr& read_schema() const { return _read_schema; }

    const segment_v2::ColumnIteratorOptions& column_iter_opts() const { return _column_iter_opts; }

    bool has_index_result_for_expr(const VExpr* expr) const {

        return _index_result_bitmap.contains(expr);

    }

    void set_index_result_for_expr(const VExpr* expr,

                                   segment_v2::InvertedIndexResultBitmap bitmap) {

        _index_result_bitmap[expr] = std::move(bitmap);

    }

    std::unordered_map<const VExpr*, segment_v2::InvertedIndexResultBitmap>&

    get_index_result_bitmap() {

        return _index_result_bitmap;

    }

    std::unordered_map<const VExpr*, ColumnPtr>& get_index_result_column() {

        return _index_result_column;

    }

    const segment_v2::InvertedIndexResultBitmap* get_index_result_for_expr(const VExpr* expr) {

        auto iter = _index_result_bitmap.find(expr);

        if (iter == _index_result_bitmap.end()) {

            return nullptr;

        }

        return &iter->second;

    }

    // Dedicated entry point for approximate (superset) index results, kept strictly apart from

    // the exact result map above.

    void set_approx_index_result_for_expr(const VExpr* expr,

                                          segment_v2::InvertedIndexResultBitmap bitmap) {

        _approx_index_result_bitmap[expr] = std::move(bitmap);

    }

    const segment_v2::InvertedIndexResultBitmap* get_approx_index_result_for_expr(

            const VExpr* expr) const {

        auto iter = _approx_index_result_bitmap.find(expr);

        if (iter == _approx_index_result_bitmap.end()) {

            return nullptr;

        }

        return &iter->second;

    }

    void set_index_result_column_for_expr(const VExpr* expr, ColumnPtr column) {

        _index_result_column[expr] = std::move(column);

    }

    void set_true_for_index_status(const VExpr* expr, int32_t read_ordinal) {

        if (_expr_index_status.contains(read_ordinal)) {

            if (_expr_index_status[read_ordinal].contains(expr)) {

                _expr_index_status[read_ordinal][expr] = true;

            }

        }

    }

    ScoreRuntimeSPtr get_score_runtime() const { return _score_runtime; }

    void set_analyzer_ctx_for_expr(const VExpr* expr, InvertedIndexAnalyzerCtxSPtr analyzer_ctx) {

        if (expr == nullptr || analyzer_ctx == nullptr) {

            return;

        }

        _expr_analyzer_ctx[expr] = std::move(analyzer_ctx);

    }

    const InvertedIndexAnalyzerCtx* get_analyzer_ctx_for_expr(const VExpr* expr) const {

        auto iter = _expr_analyzer_ctx.find(expr);

        if (iter == _expr_analyzer_ctx.end()) {

            return nullptr;

        }

        return iter->second.get();

    }

    void set_index_query_context(segment_v2::IndexQueryContextPtr index_query_context) {

        _index_query_context = index_query_context;

    }

    const segment_v2::IndexQueryContextPtr& get_index_query_context() const {

        return _index_query_context;

    }

private:

    // A reference to a vector of unique pointers to index iterators.

    const std::vector<std::unique_ptr<segment_v2::IndexIterator>>& _index_iterators;

    // A reference to a vector of storage name and type pairs related to schema.

    const std::vector<IndexFieldNameAndTypePair>& _storage_name_and_type;

    // A map of expressions to their corresponding inverted index result bitmaps.

    std::unordered_map<const VExpr*, segment_v2::InvertedIndexResultBitmap> _index_result_bitmap;

    // A map of expressions to their corresponding result columns.

    std::unordered_map<const VExpr*, ColumnPtr> _index_result_column;

    // Approximate (superset) index results: rows outside the bitmap certainly do not match, but

    // rows inside it may not match either, so it may only be used to prune candidate rows and

    // the expression must stay in the push-down list to re-verify them. Three invariants:

    // (a) never write into _index_result_bitmap / _index_result_column -- VExpr::fast_execute

    //     would then pass the candidate bitmap off as the function result, and

    //     _output_index_result_column would materialize it as a result column;

    // (b) never call set_true_for_index_status -- the column would then be judged

    //     need_read_data=false and there would be no column left to read during re-verification;

    // (c) intersecting with _row_bitmap is allowed only when the expression happens to be the

    //     root of the VExprContext (a top-level AND context); wrapped in NOT/OR, VCompoundPred

    //     never sees this map, so it simply does not apply.

    std::unordered_map<const VExpr*, segment_v2::InvertedIndexResultBitmap>

            _approx_index_result_bitmap;

    // Per-expression analyzer context for inverted index evaluation.

    std::unordered_map<const VExpr*, InvertedIndexAnalyzerCtxSPtr> _expr_analyzer_ctx;

    // A reference to a map of common expressions to their inverted index evaluation status.

    std::unordered_map<ColumnId, std::unordered_map<const VExpr*, bool>>& _expr_index_status;

    ScoreRuntimeSPtr _score_runtime;

    segment_v2::Segment* _segment = nullptr; // Ref

    segment_v2::ColumnIteratorOptions _column_iter_opts;

    // Keep the scan columns alive while index expressions use their TabletColumn pointers.

    ReadSchemaSPtr _read_schema;

    segment_v2::IndexQueryContextPtr _index_query_context;

};

class VExprContext {

    ENABLE_FACTORY_CREATOR(VExprContext);

public:

    VExprContext(VExprSPtr expr);

    ~VExprContext();

    [[nodiscard]] Status prepare(RuntimeState* state, const RowDescriptor& row_desc);

    [[nodiscard]] Status open(RuntimeState* state);

    [[nodiscard]] Status clone(RuntimeState* state, VExprContextSPtr& new_ctx);

    [[nodiscard]] Status execute(Block* block, int* result_column_id);

    [[nodiscard]] Status execute(const Block* block, ColumnPtr& result_column);

    [[nodiscard]] Status execute(const Block* block, ColumnWithTypeAndName& result_data);

    [[nodiscard]] DataTypePtr execute_type(const Block* block);

    [[nodiscard]] const std::string& expr_name() const;

    [[nodiscard]] bool is_blockable() const;

    [[nodiscard]] Status execute_const_expr(ColumnWithTypeAndName& result);

    double execute_cost() const;

    VExprSPtr root() { return _root; }

    void set_root(const VExprSPtr& expr) { _root = expr; }

    void set_index_context(std::shared_ptr<IndexExecContext> index_context) {

        _index_context = std::move(index_context);

    }

    std::shared_ptr<IndexExecContext> get_index_context() const { return _index_context; }

    LambdaExecutionContext& lambda_execution_context();

    /// Creates a FunctionContext, and returns the index that's passed to fn_context() to

    /// retrieve the created context. Exprs that need a FunctionContext should call this in

    /// Prepare() and save the returned index. 'varargs_buffer_size', if specified, is the

    /// size of the varargs buffer in the created FunctionContext (see udf-internal.h).

    int register_function_context(RuntimeState* state, const DataTypePtr& return_type,

                                  const std::vector<DataTypePtr>& arg_types);

    /// Retrieves a registered FunctionContext. 'i' is the index returned by the call to

    /// register_function_context(). This should only be called by VExprs.

    FunctionContext* fn_context(int i) {

        if (i < 0 || i >= _fn_contexts.size()) {

            throw Exception(ErrorCode::INTERNAL_ERROR,

                            "fn_context index invalid, index={}, _fn_contexts.size()={}", i,

                            _fn_contexts.size());

        }

        return _fn_contexts[i].get();

    }

    void set_auto_partition_boundary_context() {

        for (auto& fn_context : _fn_contexts) {

            fn_context->set_auto_partition_boundary_context();

        }

    }

    // execute expr with inverted index which column a, b has inverted indexes

    //  but some situation although column b has indexes, but apply index is not useful, we should

    //  skip this expr, just do not apply index anymore.

    [[nodiscard]] Status evaluate_inverted_index(uint32_t segment_num_rows);

    // The one expression whose approximate (superset) index result this caller will read back,

    // or null when it reads none. SegmentIterator names the root of a pushed-down conjunct,

    // because _apply_approx_index_result looks the result up by exactly that pointer; its

    // virtual column loop names nothing, because it only materializes exact results. For any

    // other expression -- an operand of a compound predicate, a nested child push-down -- the

    // result would be stored under its own pointer and never read again, so a function that can

    // only answer approximately skips the evaluation instead of paying for the index reads.

    void set_approx_index_result_consumer(const VExpr* expr) {

        _approx_index_result_consumer = expr;

    }

    const VExpr* approx_index_result_consumer() const { return _approx_index_result_consumer; }

    [[nodiscard]] static ZoneMapFilterResult evaluate_zonemap_filter(

            const VExprContextSPtrs& conjuncts, const ZoneMapEvalContext& ctx);

    [[nodiscard]] static ZoneMapFilterResult evaluate_dictionary_filter(

            const VExprContextSPtrs& conjuncts, const DictionaryEvalContext& ctx);

    [[nodiscard]] static ZoneMapFilterResult evaluate_bloom_filter(

            const VExprContextSPtrs& conjuncts, const BloomFilterEvalContext& ctx);

    bool all_expr_inverted_index_evaluated();

    Status execute_filter(const Block* block, uint8_t* __restrict result_filter_data, size_t rows,

                          bool accept_null, bool* can_filter_all);

    [[nodiscard]] static Status filter_block(VExprContext* vexpr_ctx, Block* block);

    [[nodiscard]] static Status filter_block(const VExprContextSPtrs& expr_contexts, Block* block,

                                             size_t column_to_keep);

    [[nodiscard]] static Status execute_conjuncts(const VExprContextSPtrs& ctxs,

                                                  const std::vector<IColumn::Filter*>* filters,

                                                  bool accept_null, const Block* block,

                                                  IColumn::Filter* result_filter,

                                                  bool* can_filter_all);

    [[nodiscard]] static Status execute_conjuncts(const VExprContextSPtrs& conjuncts,

                                                  const Block* block, ColumnUInt8& null_map,

                                                  IColumn::Filter& result_filter);

    static Status execute_conjuncts(const VExprContextSPtrs& ctxs,

                                    const std::vector<IColumn::Filter*>* filters, Block* block,

                                    IColumn::Filter* result_filter, bool* can_filter_all);

    [[nodiscard]] static Status execute_conjuncts_and_filter_block(

            const VExprContextSPtrs& ctxs, Block* block, std::vector<uint32_t>& columns_to_filter,

            int column_to_keep);

    static Status execute_conjuncts_and_filter_block(const VExprContextSPtrs& ctxs, Block* block,

                                                     std::vector<uint32_t>& columns_to_filter,

                                                     int column_to_keep, IColumn::Filter& filter);

    [[nodiscard]] static Status get_output_block_after_execute_exprs(const VExprContextSPtrs&,

                                                                     const Block&, Block*,

                                                                     bool do_projection = false);

    int get_last_result_column_id() const {

        DCHECK(_last_result_column_id != -1);

        return _last_result_column_id;

    }

    RuntimeFilterSelectivity& get_runtime_filter_selectivity() {

        if (!_rf_selectivity) {

            throw Exception(ErrorCode::INTERNAL_ERROR, "RuntimeFilterSelectivity is null");

        }

        return *_rf_selectivity;

    }

    FunctionContext::FunctionStateScope get_function_state_scope() const {

        return _is_clone ? FunctionContext::THREAD_LOCAL : FunctionContext::FRAGMENT_LOCAL;

    }

    void clone_fn_contexts(VExprContext* other);

    VExprContext& operator=(const VExprContext& other) = delete;

    VExprContext& operator=(VExprContext&& other) = delete;

    [[nodiscard]] static size_t get_memory_usage(const VExprContextSPtrs& contexts) {

        size_t usage = 0;

        std::for_each(contexts.cbegin(), contexts.cend(),

                      [&usage](auto&& context) { usage += context->_memory_usage; });

        return usage;

    }

    [[nodiscard]] size_t get_memory_usage() const { return _memory_usage; }

    void prepare_ann_range_search(const doris::VectorSearchUserParams& params);

    Status evaluate_ann_range_search(

            const std::vector<std::unique_ptr<segment_v2::IndexIterator>>& index_iterators,

            const std::vector<std::unique_ptr<segment_v2::ColumnIterator>>& column_iterators,

            const std::unordered_map<VExprContext*, std::unordered_map<ColumnId, VExpr*>>&

                    common_expr_to_slotref_map,

            size_t rows_of_segment, roaring::Roaring& row_bitmap,

            segment_v2::AnnIndexStats& ann_index_stats, bool enable_result_cache,

            bool* ann_range_search_executed);

    uint64_t get_digest(uint64_t seed) const;

private:

    // Close method is called in vexpr context dector, not need call expicility

    void close();

    static void _reset_memory_usage(const VExprContextSPtrs& contexts);

    friend class VExpr;

    /// The expr tree this context is for.

    VExprSPtr _root;

    /// True if this context came from a Clone() call. Used to manage FunctionStateScope.

    bool _is_clone = false;

    /// Variables keeping track of current state.

    bool _prepared = false;

    bool _opened = false;

    /// See set_approx_index_result_consumer. Null by default, so a caller that does not read the

    /// approximate map never pays for the index reads behind one.

    const VExpr* _approx_index_result_consumer = nullptr;

    /// FunctionContexts for each registered expression. The FunctionContexts are created

    /// and owned by this VExprContext.

    std::vector<std::unique_ptr<FunctionContext>> _fn_contexts;

    int _last_result_column_id = -1;

    /// The depth of expression-tree.

    int _depth_num = 0;

    std::shared_ptr<IndexExecContext> _index_context;

    size_t _memory_usage = 0;

    segment_v2::AnnRangeSearchRuntime _ann_range_search_runtime;

    bool _suitable_for_ann_index = true;

    std::unique_ptr<LambdaExecutionContext> _lambda_execution_context;

    std::unique_ptr<RuntimeFilterSelectivity> _rf_selectivity =

            std::make_unique<RuntimeFilterSelectivity>();

};

} // namespace doris