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

#include <map>

#include <memory>

#include <optional>

#include <string>

#include <unordered_map>

#include <vector>

#include "common/factory_creator.h"

#include "common/status.h"

#include "core/block/adaptive_block_size_predictor.h"

#include "core/block/block.h"

#include "exec/operator/file_scan_operator.h"

#include "exec/scan/scanner.h"

#include "exec/scan/split_source_connector.h"

#include "exprs/vexpr_fwd.h"

#include "format_v2/column_mapper.h"

#include "format_v2/table_reader.h"

#include "gen_cpp/Descriptors_types.h"

#include "gen_cpp/PlanNodes_types.h"

#include "io/io_common.h"

#include "runtime/runtime_profile.h"

namespace doris {

class RuntimeState;

class SlotDescriptor;

class TFileRangeDesc;

class TFileScanRangeParams;

class ShardedKVCache;

class FileScannerV2 final : public Scanner {

    ENABLE_FACTORY_CREATOR(FileScannerV2);

public:

    static constexpr const char* NAME = "FileScannerV2";

    static constexpr size_t ADAPTIVE_BATCH_INITIAL_PROBE_ROWS = 32;

    static const std::string FileReadBytesProfile;

    static const std::string FileReadTimeProfile;

    struct RealtimeCounterDeltas {

        int64_t scan_rows = 0;

        int64_t scan_bytes = 0;

        int64_t scan_bytes_from_local_storage = 0;

        int64_t scan_bytes_from_remote_storage = 0;

    };

    enum class UncachedReaderBytesStorage { LOCAL, REMOTE, NONE };

    static bool is_supported(const TFileScanRangeParams& params, const TFileRangeDesc& range);

#ifdef BE_TEST

    FileScannerV2(RuntimeState* state, RuntimeProfile* profile,

                  std::unique_ptr<format::TableReader> table_reader);

    static Status TEST_validate_scan_range(const TFileScanRangeParams& params,

                                           const TFileRangeDesc& range);

    static Status TEST_to_file_format(TFileFormatType::type format_type,

                                      format::FileFormat* file_format);

    static bool TEST_is_partition_slot(const TFileScanSlotInfo& slot_info,

                                       const std::string& column_name);

    static bool TEST_is_data_file_slot(const TFileScanSlotInfo& slot_info,

                                       const std::string& column_name);

    static Status TEST_rewrite_slot_refs_to_global_index(

            VExprSPtr* expr,

            const std::unordered_map<int32_t, format::GlobalIndex>& slot_id_to_global_index);

    static RealtimeCounterDeltas TEST_collect_realtime_counter_deltas(

            const io::FileReaderStats& file_reader_stats,

            const io::FileCacheStatistics& file_cache_statistics,

            UncachedReaderBytesStorage uncached_reader_bytes_storage, int64_t* last_read_bytes,

            int64_t* last_read_rows, int64_t* last_bytes_read_from_local,

            int64_t* last_bytes_read_from_remote);

    static void TEST_report_file_cache_profile(

            RuntimeProfile* profile, const io::FileCacheStatistics& file_cache_statistics);

    static bool TEST_should_skip_not_found(const Status& status, bool ignore_not_found);

    static bool TEST_should_skip_empty(const Status& status, bool stopped);

    static Status TEST_validate_variant_projection(TFileFormatType::type format_type,

                                                   bool has_variant_projection) {

        return _validate_variant_projection(format_type, has_variant_projection);

    }

    static Status TEST_contextualize_output_filter_status(Status status,

                                                          TFileFormatType::type format_type) {

        return _contextualize_output_filter_status(std::move(status), format_type);

    }

    static bool TEST_should_run_adaptive_batch_size(bool predictor_initialized,

                                                    bool current_split_uses_metadata_count) {

        return _should_run_adaptive_batch_size(predictor_initialized,

                                               current_split_uses_metadata_count);

    }

#endif

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

    // Standalone scanner used by TopN two-phase materialization.

    FileScannerV2(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) {

        (void)colname_to_slot_id;

        _output_tuple_desc = tuple_desc;

    }

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

                        Block* result_block, int64_t* init_reader_ms, int64_t* get_block_ms);

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

    Status _open_impl(RuntimeState* state) override;

    Status close(RuntimeState* state) override;

    void try_stop() override;

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

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

    void update_realtime_counters() override;

protected:

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

    bool _can_merge_padding_blocks(const Block& left, const Block& right) const override;

    Status _filter_output_block(Block* block) override;

    void _collect_profile_before_close() override;

    bool _should_update_load_counters() const override;

private:

    static Status _validate_scan_range(const TFileScanRangeParams& params,

                                       const TFileRangeDesc& range);

    static Status _validate_variant_projection(TFileFormatType::type format_type,

                                               bool has_variant_projection);

    Status _get_next_scan_range(bool* has_next);

    TFileFormatType::type _get_current_format_type() const;

    Status _init_io_ctx();

    Status _init_expr_ctxes();

    Status _prepare_next_split(bool* eos);

    Status _init_table_reader(const TFileRangeDesc& range);

    Status _create_table_reader_for_format(const TFileRangeDesc& range,

                                           std::unique_ptr<format::TableReader>* reader) const;

    Status _prepare_table_reader_split(const TFileRangeDesc& range,

                                       std::map<std::string, Field> partition_values);

    static bool _should_skip_not_found(const Status& status, bool ignore_not_found);

    static bool _should_skip_empty(const Status& status, bool stopped);

    static Status _contextualize_output_filter_status(Status status,

                                                      TFileFormatType::type format_type);

    bool _should_enable_file_meta_cache() const;

    std::optional<format::GlobalRowIdContext> _create_global_rowid_context(

            const TFileRangeDesc& range) const;

    Status _generate_partition_values(const TFileRangeDesc& range,

                                      std::map<std::string, Field>* partition_values) const;

    Status _parse_partition_value(const SlotDescriptor* slot_desc, const std::string& value,

                                  bool is_null, Field* field) const;

    Status _build_projected_columns(const format::TableReader& table_reader);

    Status _build_default_expr(const TFileScanSlotInfo& slot_info, VExprContextSPtr* ctx) const;

    static format::ColumnDefinition _build_table_column(const SlotDescriptor* slot_desc);

    Status _build_table_conjuncts(VExprContextSPtrs* conjuncts) const;

    static Status _to_file_format(TFileFormatType::type format_type,

                                  format::FileFormat* file_format);

    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;

    static bool _should_run_adaptive_batch_size(bool predictor_initialized,

                                                bool current_split_uses_metadata_count);

    size_t _predict_reader_batch_rows();

    void _update_adaptive_batch_size(const Block& block);

    static RealtimeCounterDeltas _collect_realtime_counter_deltas(

            const io::FileReaderStats& file_reader_stats,

            const io::FileCacheStatistics& file_cache_statistics,

            UncachedReaderBytesStorage uncached_reader_bytes_storage, int64_t* last_read_bytes,

            int64_t* last_read_rows, int64_t* last_bytes_read_from_local,

            int64_t* last_bytes_read_from_remote);

    static UncachedReaderBytesStorage _uncached_reader_bytes_storage(TFileType::type file_type);

    static void _report_file_cache_profile(RuntimeProfile* profile,

                                           const io::FileCacheStatistics& file_cache_statistics);

    void _report_file_reader_predicate_filtered_rows();

    void _report_condition_cache_profile();

    struct PartitionSlotInfo {

        const SlotDescriptor* slot_desc = nullptr;

        std::string canonical_name;

    };

    const TFileScanRangeParams* _params = nullptr;

    std::shared_ptr<SplitSourceConnector> _split_source;

    bool _first_scan_range = false;

    bool _has_prepared_split = false;

    int _table_reader_rf_num = 0;

    TFileRangeDesc _current_range;

    std::string _current_range_path;

    std::unique_ptr<format::TableReader> _table_reader;

    std::vector<format::ColumnDefinition> _projected_columns;

    bool _has_variant_projection = false;

    // File formats without embedded schema, such as CSV, still need the FE slot descriptors in

    // file-column order. This mirrors old FileScanner::_file_slot_descs and is passed only to

    // readers that cannot derive their schema from file metadata.

    std::vector<SlotDescriptor*> _file_slot_descs;

    bool _need_global_rowid_column = false;

    std::unordered_map<int32_t, const SlotDescriptor*> _slot_id_to_desc;

    std::unordered_map<int32_t, format::GlobalIndex> _slot_id_to_global_index;

    std::unordered_map<std::string, PartitionSlotInfo> _partition_slot_descs;

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

    io::FileCacheStatistics _reported_file_cache_statistics;

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

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

    ShardedKVCache* _kv_cache = nullptr;

    RuntimeProfile::Counter* _scanner_total_timer = nullptr;

    RuntimeProfile::Counter* _init_timer = nullptr;

    RuntimeProfile::Counter* _open_timer = nullptr;

    RuntimeProfile::Counter* _get_block_timer = nullptr;

    RuntimeProfile::Counter* _empty_file_counter = nullptr;

    RuntimeProfile::Counter* _prepare_split_timer = nullptr;

    RuntimeProfile::Counter* _get_next_range_timer = nullptr;

    RuntimeProfile::Counter* _close_timer = nullptr;

    RuntimeProfile::Counter* _io_timer = nullptr;

    RuntimeProfile::Counter* _not_found_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* _adaptive_batch_predicted_rows_counter = nullptr;

    RuntimeProfile::Counter* _adaptive_batch_actual_bytes_counter = nullptr;

    RuntimeProfile::Counter* _adaptive_batch_probe_count_counter = nullptr;

    std::unique_ptr<AdaptiveBlockSizePredictor> _block_size_predictor;

    int64_t _reported_predicate_filtered_rows = 0;

    int64_t _reported_condition_cache_hit_count = 0;

    int64_t _reported_condition_cache_filtered_rows = 0;

    int64_t _last_read_bytes = 0;

    int64_t _last_read_rows = 0;

    int64_t _last_bytes_read_from_local = 0;

    int64_t _last_bytes_read_from_remote = 0;

    int64_t _reported_io_read_time = 0;

};

} // namespace doris