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

#include <stdint.h>

#include <memory>

#include <optional>

#include <string>

#include <unordered_map>

#include <unordered_set>

#include <vector>

#include "common/factory_creator.h"

#include "common/global_types.h"

#include "common/status.h"

#include "core/block/block.h"

#include "exec/operator/file_scan_operator.h"

#include "exec/scan/file_scan_io_context.h"

#include "exprs/vexpr_fwd.h"

#include "format/generic_reader.h"

#include "format/orc/vorc_reader.h"

#include "format/parquet/vparquet_reader.h"

#include "format/table/iceberg_reader.h"

#include "io/io_common.h"

#include "runtime/descriptors.h"

#include "runtime/runtime_profile.h"

#include "storage/olap_common.h"

#include "storage/olap_scan_common.h"

#include "storage/segment/adaptive_block_size_predictor.h"

#include "storage/segment/condition_cache.h"

namespace doris {

class RuntimeState;

class TFileRangeDesc;

class TFileScanRange;

class TFileScanRangeParams;

class ShardedKVCache;

class VExpr;

class VExprContext;

} // namespace doris

namespace doris {

class FileScanner : public Scanner {

    ENABLE_FACTORY_CREATOR(FileScanner);

public:

    static constexpr const char* NAME = "FileScanner";

    static constexpr size_t ADAPTIVE_BATCH_INITIAL_PROBE_ROWS = 32;

    // sub profile name (for parquet/orc)

    static const std::string FileReadBytesProfile;

    static const std::string FileReadTimeProfile;

#ifdef BE_TEST

    void TEST_init_runtime_filter_partition_prune_ctxs(

            const VExprContextSPtrs& conjuncts,

            const std::unordered_map<SlotId, int>& partition_slot_index_map) {

        _conjuncts = conjuncts;

        _partition_slot_index_map = partition_slot_index_map;

        _init_runtime_filter_partition_prune_ctxs();

    }

    const VExprContextSPtrs& TEST_runtime_filter_partition_prune_ctxs() const {

        return _runtime_filter_partition_prune_ctxs;

    }

    static TPushAggOp::type TEST_effective_push_down_agg_type(

            TPushAggOp::type agg_type, const std::optional<std::vector<int32_t>>& count_slot_ids) {

        return _effective_push_down_agg_type(agg_type, count_slot_ids);

    }

#endif

    FileScanner(RuntimeState* state, FileScanLocalState* parent, int64_t limit,

                std::shared_ptr<SplitSourceConnector> split_source, RuntimeProfile* profile,

                ShardedKVCache* kv_cache,

                const std::unordered_map<std::string, int>* colname_to_slot_id);

    Status _open_impl(RuntimeState* state) override;

    Status close(RuntimeState* state) override;

    void try_stop() override;

    Status init(RuntimeState* state, const VExprContextSPtrs& conjuncts) override;

    std::string get_name() override { return FileScanner::NAME; }

    std::string get_current_scan_range_name() override { return _current_range_path; }

    //only used for read one line.

    FileScanner(RuntimeState* state, RuntimeProfile* profile, const TFileScanRangeParams* params,

                const std::unordered_map<std::string, int>* colname_to_slot_id,

                TupleDescriptor* tuple_desc)

            : Scanner(state, profile),

              _params(params),

              _col_name_to_slot_id(colname_to_slot_id),

              _real_tuple_desc(tuple_desc) {

        _configure_file_scan_handlers();

    };

    Status read_lines_from_range(const TFileRangeDesc& range, const std::list<int64_t>& row_ids,

                                 Block* result_block, const ExternalFileMappingInfo& external_info,

                                 int64_t* init_reader_ms, int64_t* get_block_ms);

    Status prepare_for_read_lines(const TFileRangeDesc& range);

    void update_realtime_counters() override;

protected:

    Status _get_block_impl(RuntimeState* state, Block* block, bool* eof) override;

    Status _get_block_wrapped(RuntimeState* state, Block* block, bool* eof);

    Status _get_next_reader();

    // Build a ReaderInitContext with shared fields from FileScanner members.

    void _fill_base_init_context(ReaderInitContext* ctx);

    // TODO: cast input block columns type to string.

    Status _cast_src_block(Block* block) { return Status::OK(); }

    void _collect_profile_before_close() override;

    bool _should_update_load_counters() const override;

    // fe will add skip_bitmap_col to _input_tuple_desc iff the target olaptable has skip_bitmap_col

    // and the current load is a flexible partial update

    bool _should_process_skip_bitmap_col() const { return _skip_bitmap_col_idx != -1; }

protected:

    const TFileScanRangeParams* _params = nullptr;

    std::shared_ptr<SplitSourceConnector> _split_source;

    bool _first_scan_range = false;

    TFileRangeDesc _current_range;

    std::unique_ptr<GenericReader> _cur_reader;

    bool _cur_reader_eof = false;

    // File source slot descriptors

    std::vector<SlotDescriptor*> _file_slot_descs;

    // Unified column descriptors for init_reader (includes file, partition, missing, synthesized cols)

    std::vector<ColumnDescriptor> _column_descs;

    // Partition slot id to partition key index (for matching columns_from_path)

    std::unordered_map<SlotId, int> _partition_slot_index_map;

    // created from param.expr_of_dest_slot

    // For query, it saves default value expr of all dest columns, or nullptr for NULL.

    // For load, it saves conversion expr/default value of all dest columns.

    VExprContextSPtrs _dest_vexpr_ctx;

    // dest slot name to index in _dest_vexpr_ctx;

    std::unordered_map<std::string, int> _dest_slot_name_to_idx;

    // col name to default value expr

    // TODO: only used by json reader. Could we delete this?

    std::unordered_map<std::string, VExprContextSPtr> _col_default_value_ctx;

    // the map values of dest slot id to src slot desc

    // if there is not key of dest slot id in dest_sid_to_src_sid_without_trans, it will be set to nullptr

    std::vector<SlotDescriptor*> _src_slot_descs_order_by_dest;

    // dest slot desc index to src slot desc index

    std::unordered_map<int, int> _dest_slot_to_src_slot_index;

    std::unordered_map<std::string, uint32_t> _src_block_name_to_idx;

    // Get from GenericReader, save the existing columns in file to their type.

    std::unordered_map<std::string, DataTypePtr> _slot_lower_name_to_col_type;

    // Get from GenericReader, save columns that required by scan but not exist in file.

    // The col lowercase name of source file to type of source file.

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

    // For load task

    VExprContextSPtrs _pre_conjunct_ctxs;

    std::unique_ptr<RowDescriptor> _src_row_desc;

    std::unique_ptr<RowDescriptor> _dest_row_desc;

    // row desc for default exprs

    std::unique_ptr<RowDescriptor> _default_val_row_desc;

    // owned by scan node

    ShardedKVCache* _kv_cache = nullptr;

    std::set<TSlotId> _is_file_slot;

    bool _scanner_eof = false;

    int _rows = 0;

    int _num_of_columns_from_file;

    bool _src_block_mem_reuse = false;

    bool _strict_mode;

    bool _src_block_init = false;

    Block* _src_block_ptr = nullptr;

    Block _src_block;

    VExprContextSPtrs _push_down_conjuncts;

    VExprContextSPtrs _runtime_filter_partition_prune_ctxs;

    Block _runtime_filter_partition_prune_block;

    std::unique_ptr<io::FileCacheStatistics> _file_cache_statistics;

    std::unique_ptr<io::FileReaderStats> _file_reader_stats;

    std::shared_ptr<io::IOContext> _io_ctx;

    // Whether to fill partition columns from path, default is true.

    std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>

            _partition_col_descs;

    std::unordered_map<std::string, bool> _partition_value_is_null;

    // idx of skip_bitmap_col in _input_tuple_desc

    int32_t _skip_bitmap_col_idx {-1};

    int32_t _sequence_map_col_uid {-1};

    int32_t _sequence_col_uid {-1};

private:

    RuntimeProfile::Counter* _get_block_timer = nullptr;

    RuntimeProfile::Counter* _cast_to_input_block_timer = nullptr;

    RuntimeProfile::Counter* _fill_missing_columns_timer = nullptr;

    RuntimeProfile::Counter* _pre_filter_timer = nullptr;

    RuntimeProfile::Counter* _convert_to_output_block_timer = nullptr;

    RuntimeProfile::Counter* _runtime_filter_partition_prune_timer = nullptr;

    RuntimeProfile::Counter* _empty_file_counter = nullptr;

    RuntimeProfile::Counter* _not_found_file_counter = nullptr;

    RuntimeProfile::Counter* _fully_skipped_file_counter = nullptr;

    RuntimeProfile::Counter* _file_counter = nullptr;

    RuntimeProfile::Counter* _file_read_bytes_counter = nullptr;

    RuntimeProfile::Counter* _file_read_calls_counter = nullptr;

    RuntimeProfile::Counter* _file_read_time_counter = nullptr;

    RuntimeProfile::Counter* _runtime_filter_partition_pruned_range_counter = nullptr;

    RuntimeProfile::Counter* _adaptive_batch_predicted_rows_counter = nullptr;

    RuntimeProfile::Counter* _adaptive_batch_actual_bytes_before_truncate_counter = nullptr;

    RuntimeProfile::Counter* _adaptive_batch_actual_bytes_after_truncate_counter = nullptr;

    RuntimeProfile::Counter* _adaptive_batch_probe_count_counter = nullptr;

    const std::unordered_map<std::string, int>* _col_name_to_slot_id = nullptr;

    // single slot filter conjuncts

    std::unordered_map<int, VExprContextSPtrs> _slot_id_to_filter_conjuncts;

    // not single(zero or multi) slot filter conjuncts

    VExprContextSPtrs _not_single_slot_filter_conjuncts;

    // save the path of current scan range

    std::string _current_range_path = "";

    // Only for load scan node.

    const TupleDescriptor* _input_tuple_desc = nullptr;

    // If _input_tuple_desc is set,

    // the _real_tuple_desc will point to _input_tuple_desc,

    // otherwise, point to _output_tuple_desc

    const TupleDescriptor* _real_tuple_desc = nullptr;

    int64_t _last_bytes_read_from_local = 0;

    int64_t _last_bytes_read_from_remote = 0;

    Status (FileScanner::*_init_src_block_handler)(Block* block) = nullptr;

    Status (FileScanner::*_process_src_block_after_read_handler)(Block* block) = nullptr;

    bool (FileScanner::*_should_push_down_predicates_handler)(

            TFileFormatType::type format_type) const = nullptr;

    bool (FileScanner::*_should_enable_condition_cache_handler)() const = nullptr;

    // Condition cache for external tables

    uint64_t _condition_cache_digest = 0;

    segment_v2::ConditionCache::ExternalCacheKey _condition_cache_key;

    std::shared_ptr<std::vector<bool>> _condition_cache;

    std::shared_ptr<ConditionCacheContext> _condition_cache_ctx;

    int64_t _condition_cache_hit_count = 0;

    std::unique_ptr<AdaptiveBlockSizePredictor> _block_size_predictor;

    void _configure_file_scan_handlers();

    Status _init_expr_ctxes();

    Status _init_src_block(Block* block);

    Status _init_src_block_for_load(Block* block);

    Status _init_src_block_for_query(Block* block);

    Status _process_src_block_after_read(Block* block);

    Status _process_src_block_after_read_for_load(Block* block);

    Status _process_src_block_after_read_for_query(Block* block);

    Status _check_output_block_types();

    Status _cast_to_input_block(Block* block);

    Status _pre_filter_src_block();

    Status _convert_to_output_block(Block* block);

    Status _truncate_char_or_varchar_columns(Block* block);

    void _truncate_char_or_varchar_column(Block* block, int idx, int len);

    Status _generate_partition_columns();

    bool _check_partition_prune_expr(const VExprSPtr& expr);

    bool _contains_runtime_filter(const VExprContextSPtrs& conjuncts) const;

    void _init_runtime_filter_partition_prune_ctxs();

    void _init_runtime_filter_partition_prune_block();

    Status _process_runtime_filters_partition_prune(bool& is_partition_pruned);

    Status _process_conjuncts();

    Status _process_late_arrival_conjuncts();

    void _get_slot_ids(VExpr* expr, std::vector<int>* slot_ids);

    Status _generate_truncate_columns(bool need_to_get_parsed_schema);

    Status _set_fill_or_truncate_columns(bool need_to_get_parsed_schema);

    Status _init_orc_reader(FileMetaCache* file_meta_cache_ptr,

                            std::unique_ptr<OrcReader> orc_reader = nullptr);

    Status _init_parquet_reader(FileMetaCache* file_meta_cache_ptr,

                                std::unique_ptr<ParquetReader> parquet_reader = nullptr);

    std::shared_ptr<segment_v2::RowIdColumnIteratorV2> _create_row_id_column_iterator();

    TFileFormatType::type _get_current_format_type() {

        // for compatibility, if format_type is not set in range, use the format type of params

        const TFileRangeDesc& range = _current_range;

        return range.__isset.format_type ? range.format_type : _params->format_type;

    };

    Status _init_io_ctx() {

        _io_ctx = create_file_scan_io_context(_state);

        return Status::OK();

    };

    void _reset_counter() {

        _counter.num_rows_unselected = 0;

        _counter.num_rows_filtered = 0;

    }

    bool _should_enable_condition_cache();

    bool _should_enable_condition_cache_for_load() const;

    bool _should_enable_condition_cache_for_query() const;

    bool _should_push_down_predicates(TFileFormatType::type format_type) const;

    bool _should_push_down_predicates_for_load(TFileFormatType::type format_type) const;

    bool _should_push_down_predicates_for_query(TFileFormatType::type format_type) const;

    void _init_reader_condition_cache();

    void _finalize_reader_condition_cache();

    void _reset_adaptive_batch_size_state();

    void _init_adaptive_batch_size_state(TFileFormatType::type format_type);

    bool _should_enable_adaptive_batch_size(TFileFormatType::type format_type) const;

    bool _should_run_adaptive_batch_size() const;

    size_t _predict_reader_batch_rows();

    void _update_adaptive_batch_size_before_truncate(const Block& block);

    void _update_adaptive_batch_size_after_truncate(const Block& block);

    static TPushAggOp::type _effective_push_down_agg_type(

            TPushAggOp::type agg_type, const std::optional<std::vector<int32_t>>& count_slot_ids) {

        if (agg_type != TPushAggOp::type::COUNT) {

            return agg_type;

        }

        // V1's CountReader receives only the file's total row count and emits that many synthetic

        // rows. This is exact for COUNT(*)/COUNT(1), but it has no column metadata for NULL or CAST

        // semantics. For example, a 10,000-row file with 9,015 non-null values must return 10,000

        // for COUNT(*) and 9,015 for COUNT(nullable_col); CountReader can produce only the former.

        // Therefore a non-empty argument list must use the normal reader. nullopt is an old FE plan

        // that predates the argument field; treating it as empty would silently reinterpret unknown

        // semantics as COUNT(*).

        return count_slot_ids.has_value() && count_slot_ids->empty() ? TPushAggOp::type::COUNT

                                                                     : TPushAggOp::type::NONE;

    }

    TPushAggOp::type _get_push_down_agg_type() const {

        if (_local_state == nullptr) {

            return TPushAggOp::type::NONE;

        }

        return _effective_push_down_agg_type(_local_state->get_push_down_agg_type(),

                                             _local_state->get_push_down_count_slot_ids());

    }

    // enable the file meta cache only when

    // 1. max_external_file_meta_cache_num is > 0

    // 2. the file number is less than 1/3 of cache's capacibility

    // Otherwise, the cache miss rate will be high

    bool _should_enable_file_meta_cache() {

        return ExecEnv::GetInstance()->file_meta_cache()->enabled() &&

               _split_source->num_scan_ranges() < config::max_external_file_meta_cache_num / 3;

    }

};

} // namespace doris