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

#include "exec/scan/file_scanner.h"

#include <fmt/format.h>

#include <gen_cpp/Exprs_types.h>

#include <gen_cpp/Metrics_types.h>

#include <gen_cpp/Opcodes_types.h>

#include <gen_cpp/PaloInternalService_types.h>

#include <gen_cpp/PlanNodes_types.h>

#include <glog/logging.h>

#include <algorithm>

#include <boost/iterator/iterator_facade.hpp>

#include <map>

#include <ranges>

#include <tuple>

#include <unordered_map>

#include <utility>

#include "common/compiler_util.h" // IWYU pragma: keep

#include "common/config.h"

#include "common/consts.h"

#include "common/logging.h"

#include "common/status.h"

#include "core/block/column_with_type_and_name.h"

#include "core/block/columns_with_type_and_name.h"

#include "core/column/column.h"

#include "core/column/column_nullable.h"

#include "core/column/column_vector.h"

#include "core/data_type/data_type.h"

#include "core/data_type/data_type_nullable.h"

#include "core/data_type/data_type_string.h"

#include "core/string_ref.h"

#include "exec/common/stringop_substring.h"

#include "exec/rowid_fetcher.h"

#include "exec/scan/scan_node.h"

#include "exprs/aggregate/aggregate_function.h"

#include "exprs/function/function.h"

#include "exprs/function/simple_function_factory.h"

#include "exprs/vexpr.h"

#include "exprs/vexpr_context.h"

#include "exprs/vexpr_fwd.h"

#include "exprs/vslot_ref.h"

#include "format/arrow/arrow_stream_reader.h"

#include "format/count_reader.h"

#include "format/csv/csv_reader.h"

#include "format/json/new_json_reader.h"

#include "format/native/native_reader.h"

#include "format/orc/vorc_reader.h"

#include "format/parquet/vparquet_reader.h"

#include "format/table/es/es_http_reader.h"

#include "format/table/hive_reader.h"

#include "format/table/hudi_jni_reader.h"

#include "format/table/hudi_reader.h"

#include "format/table/iceberg_reader.h"

#include "format/table/iceberg_sys_table_jni_reader.h"

#include "format/table/jdbc_jni_reader.h"

#include "format/table/max_compute_jni_reader.h"

#include "format/table/paimon_cpp_reader.h"

#include "format/table/paimon_jni_reader.h"

#include "format/table/paimon_predicate_converter.h"

#include "format/table/paimon_reader.h"

#include "format/table/remote_doris_reader.h"

#include "format/table/transactional_hive_reader.h"

#include "format/table/trino_connector_jni_reader.h"

#include "format/text/text_reader.h"

#ifdef BUILD_RUST_READERS

#include "format/lance/lance_rust_reader.h"

#endif

#include "io/cache/block_file_cache_profile.h"

#include "load/group_commit/wal/wal_reader.h"

#include "runtime/descriptors.h"

#include "runtime/runtime_profile.h"

#include "runtime/runtime_state.h"

namespace cctz {

class time_zone;

} // namespace cctz

namespace doris {

class ShardedKVCache;

} // namespace doris

namespace doris {

using namespace ErrorCode;

const std::string FileScanner::FileReadBytesProfile = "FileReadBytes";

const std::string FileScanner::FileReadTimeProfile = "FileReadTime";

FileScanner::FileScanner(RuntimeState* state, FileScanLocalState* local_state, 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)

        : Scanner(state, local_state, limit, profile),

          _split_source(split_source),

          _cur_reader(nullptr),

          _cur_reader_eof(false),

          _kv_cache(kv_cache),

          _strict_mode(false),

          _col_name_to_slot_id(colname_to_slot_id) {

    if (state->get_query_ctx() != nullptr &&

        state->get_query_ctx()->file_scan_range_params_map.count(local_state->parent_id()) > 0) {

        _params = &(state->get_query_ctx()->file_scan_range_params_map[local_state->parent_id()]);

    } else {

        // old fe thrift protocol

        _params = _split_source->get_params();

    }

    if (_params->__isset.strict_mode) {

        _strict_mode = _params->strict_mode;

    }

    // For load scanner, there are input and output tuple.

    // For query scanner, there is only output tuple

    _input_tuple_desc = state->desc_tbl().get_tuple_descriptor(_params->src_tuple_id);

    _real_tuple_desc = _input_tuple_desc == nullptr ? _output_tuple_desc : _input_tuple_desc;

    _is_load = (_input_tuple_desc != nullptr);

    _configure_file_scan_handlers();

}

void FileScanner::_configure_file_scan_handlers() {

    if (_is_load) {

        _init_src_block_handler = &FileScanner::_init_src_block_for_load;

        _process_src_block_after_read_handler =

                &FileScanner::_process_src_block_after_read_for_load;

        _should_push_down_predicates_handler = &FileScanner::_should_push_down_predicates_for_load;

        _should_enable_condition_cache_handler =

                &FileScanner::_should_enable_condition_cache_for_load;

    } else {

        _init_src_block_handler = &FileScanner::_init_src_block_for_query;

        _process_src_block_after_read_handler =

                &FileScanner::_process_src_block_after_read_for_query;

        _should_push_down_predicates_handler = &FileScanner::_should_push_down_predicates_for_query;

        _should_enable_condition_cache_handler =

                &FileScanner::_should_enable_condition_cache_for_query;

    }

}

Status FileScanner::init(RuntimeState* state, const VExprContextSPtrs& conjuncts) {

    RETURN_IF_ERROR(Scanner::init(state, conjuncts));

    _get_block_timer =

            ADD_TIMER_WITH_LEVEL(_local_state->scanner_profile(), "FileScannerGetBlockTime", 1);

    _cast_to_input_block_timer = ADD_TIMER_WITH_LEVEL(_local_state->scanner_profile(),

                                                      "FileScannerCastInputBlockTime", 1);

    _fill_missing_columns_timer = ADD_TIMER_WITH_LEVEL(_local_state->scanner_profile(),

                                                       "FileScannerFillMissingColumnTime", 1);

    _pre_filter_timer =

            ADD_TIMER_WITH_LEVEL(_local_state->scanner_profile(), "FileScannerPreFilterTimer", 1);

    _convert_to_output_block_timer = ADD_TIMER_WITH_LEVEL(_local_state->scanner_profile(),

                                                          "FileScannerConvertOuputBlockTime", 1);

    _runtime_filter_partition_prune_timer = ADD_TIMER_WITH_LEVEL(

            _local_state->scanner_profile(), "FileScannerRuntimeFilterPartitionPruningTime", 1);

    _empty_file_counter =

            ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(), "EmptyFileNum", TUnit::UNIT, 1);

    _not_found_file_counter = ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),

                                                     "NotFoundFileNum", TUnit::UNIT, 1);

    _fully_skipped_file_counter = ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),

                                                         "FullySkippedFileNum", TUnit::UNIT, 1);

    _file_counter =

            ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(), "FileNumber", TUnit::UNIT, 1);

    _file_read_bytes_counter = ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),

                                                      FileReadBytesProfile, TUnit::BYTES, 1);

    _file_read_calls_counter = ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),

                                                      "FileReadCalls", TUnit::UNIT, 1);

    _file_read_time_counter =

            ADD_TIMER_WITH_LEVEL(_local_state->scanner_profile(), FileReadTimeProfile, 1);

    _runtime_filter_partition_pruned_range_counter =

            ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),

                                   "RuntimeFilterPartitionPrunedRangeNum", TUnit::UNIT, 1);

    // Keep the current file's adaptive state while also preserving the peak value across all

    // files handled by this scanner instance.

    _adaptive_batch_predicted_rows_counter =

            _local_state->scanner_profile()->AddHighWaterMarkCounter(

                    "AdaptiveBatchPredictedRows", TUnit::UNIT, RuntimeProfile::ROOT_COUNTER, 1);

    _adaptive_batch_actual_bytes_before_truncate_counter =

            _local_state->scanner_profile()->AddHighWaterMarkCounter(

                    "AdaptiveBatchActualBytesBeforeTruncate", TUnit::BYTES,

                    RuntimeProfile::ROOT_COUNTER, 1);

    _adaptive_batch_actual_bytes_after_truncate_counter =

            _local_state->scanner_profile()->AddHighWaterMarkCounter(

                    "AdaptiveBatchActualBytesAfterTruncate", TUnit::BYTES,

                    RuntimeProfile::ROOT_COUNTER, 1);

    _adaptive_batch_probe_count_counter = ADD_COUNTER_WITH_LEVEL(

            _local_state->scanner_profile(), "AdaptiveBatchProbeCount", TUnit::UNIT, 1);

    _file_cache_statistics.reset(new io::FileCacheStatistics());

    _file_reader_stats.reset(new io::FileReaderStats());

    RETURN_IF_ERROR(_init_io_ctx());

    _io_ctx->file_cache_stats = _file_cache_statistics.get();

    _io_ctx->file_reader_stats = _file_reader_stats.get();

    _io_ctx->is_disposable = _state->query_options().disable_file_cache;

    if (_is_load) {

        _src_row_desc.reset(new RowDescriptor(_state->desc_tbl(),

                                              std::vector<TupleId>({_input_tuple_desc->id()})));

        // prepare pre filters

        if (_params->__isset.pre_filter_exprs_list) {

            RETURN_IF_ERROR(doris::VExpr::create_expr_trees(_params->pre_filter_exprs_list,

                                                            _pre_conjunct_ctxs));

        } else if (_params->__isset.pre_filter_exprs) {

            VExprContextSPtr context;

            RETURN_IF_ERROR(doris::VExpr::create_expr_tree(_params->pre_filter_exprs, context));

            _pre_conjunct_ctxs.emplace_back(context);

        }

        for (auto& conjunct : _pre_conjunct_ctxs) {

            RETURN_IF_ERROR(conjunct->prepare(_state, *_src_row_desc));

            RETURN_IF_ERROR(conjunct->open(_state));

        }

        _dest_row_desc.reset(new RowDescriptor(_state->desc_tbl(),

                                               std::vector<TupleId>({_output_tuple_desc->id()})));

    }

    _default_val_row_desc.reset(

            new RowDescriptor(_state->desc_tbl(), std::vector<TupleId>({_real_tuple_desc->id()})));

    return Status::OK();

}

bool FileScanner::_should_enable_adaptive_batch_size(TFileFormatType::type format_type) const {

    // Only enable for readers that support set_batch_size().

    // Table-format wrappers are covered because they delegate to native readers.

    if (!config::enable_adaptive_batch_size) {

        return false;

    }

    switch (format_type) {

    case TFileFormatType::FORMAT_PARQUET:

    case TFileFormatType::FORMAT_ORC:

    case TFileFormatType::FORMAT_CSV_PLAIN:

    case TFileFormatType::FORMAT_CSV_GZ:

    case TFileFormatType::FORMAT_CSV_BZ2:

    case TFileFormatType::FORMAT_CSV_LZ4FRAME:

    case TFileFormatType::FORMAT_CSV_LZ4BLOCK:

    case TFileFormatType::FORMAT_CSV_LZOP:

    case TFileFormatType::FORMAT_CSV_DEFLATE:

    case TFileFormatType::FORMAT_CSV_SNAPPYBLOCK:

    case TFileFormatType::FORMAT_PROTO:

    case TFileFormatType::FORMAT_TEXT:

    case TFileFormatType::FORMAT_JSON:

    case TFileFormatType::FORMAT_JNI:

        return true;

    default:

        return false;

    }

}

bool FileScanner::_should_run_adaptive_batch_size() const {

    // Skip adaptive batch sizing for pushed-down COUNT(*): the reader is wrapped by CountReader

    // and only emits a single aggregated row count instead of materializing real columns, so

    // there is no per-row byte cost to learn from and no benefit in tuning the batch size.

    return _block_size_predictor != nullptr && _get_push_down_agg_type() != TPushAggOp::type::COUNT;

}

void FileScanner::_reset_adaptive_batch_size_state() {

    _block_size_predictor.reset();

    COUNTER_SET(_adaptive_batch_predicted_rows_counter, int64_t(0));

    COUNTER_SET(_adaptive_batch_actual_bytes_before_truncate_counter, int64_t(0));

    COUNTER_SET(_adaptive_batch_actual_bytes_after_truncate_counter, int64_t(0));

}

void FileScanner::_init_adaptive_batch_size_state(TFileFormatType::type format_type) {

    _reset_adaptive_batch_size_state();

    if (!_should_enable_adaptive_batch_size(format_type)) {

        return;

    }

    // External file readers do not provide reliable memory-size metadata hints. Use a small probe

    // batch so the predictor can learn from real FileScanner output quickly.

    _block_size_predictor = std::make_unique<AdaptiveBlockSizePredictor>(

            _state->preferred_block_size_bytes(), 0.0, ADAPTIVE_BATCH_INITIAL_PROBE_ROWS,

            _state->batch_size());

}

size_t FileScanner::_predict_reader_batch_rows() {

    DCHECK(_block_size_predictor != nullptr);

    size_t predicted_rows = _block_size_predictor->predict_next_rows();

    COUNTER_SET(_adaptive_batch_predicted_rows_counter, static_cast<int64_t>(predicted_rows));

    return predicted_rows;

}

void FileScanner::_update_adaptive_batch_size_before_truncate(const Block& block) {

    if (!_should_run_adaptive_batch_size()) {

        return;

    }

    // Learn from the logical bytes before CHAR/VARCHAR truncation. The truncated block can be

    // much smaller than the data the reader and FileScanner have already materialized.

    COUNTER_SET(_adaptive_batch_actual_bytes_before_truncate_counter,

                static_cast<int64_t>(block.bytes()));

    if (block.rows() == 0) {

        return;

    }

    // Count a probe only when we actually obtain the first non-empty sample that seeds history.

    if (!_block_size_predictor->has_history()) {

        COUNTER_UPDATE(_adaptive_batch_probe_count_counter, 1);

    }

    _block_size_predictor->update(block);

}

void FileScanner::_update_adaptive_batch_size_after_truncate(const Block& block) {

    if (!_should_run_adaptive_batch_size()) {

        return;

    }

    // Keep the post-truncate size only for observability. It should not affect the next batch

    // because truncation happens after the upstream memory cost has already been paid.

    COUNTER_SET(_adaptive_batch_actual_bytes_after_truncate_counter,

                static_cast<int64_t>(block.bytes()));

}

// check if the expr is a partition pruning expr

bool FileScanner::_check_partition_prune_expr(const VExprSPtr& expr) {

    if (expr->is_slot_ref()) {

        auto* slot_ref = static_cast<VSlotRef*>(expr.get());

        return _partition_slot_index_map.find(slot_ref->slot_id()) !=

               _partition_slot_index_map.end();

    }

    if (expr->is_literal()) {

        return true;

    }

    return std::ranges::all_of(expr->children(), [this](const auto& child) {

        return _check_partition_prune_expr(child);

    });

}

void FileScanner::_init_runtime_filter_partition_prune_ctxs() {

    _runtime_filter_partition_prune_ctxs.clear();

    for (auto& conjunct : _conjuncts) {

        auto impl = conjunct->root()->get_impl();

        // If impl is not null, which means this a conjuncts from runtime filter.

        auto expr = impl ? impl : conjunct->root();

        if (_check_partition_prune_expr(expr)) {

            _runtime_filter_partition_prune_ctxs.emplace_back(conjunct);

        }

    }

}

void FileScanner::_init_runtime_filter_partition_prune_block() {

    // init block with empty column

    for (auto const* slot_desc : _real_tuple_desc->slots()) {

        _runtime_filter_partition_prune_block.insert(

                ColumnWithTypeAndName(slot_desc->get_empty_mutable_column(),

                                      slot_desc->get_data_type_ptr(), slot_desc->col_name()));

    }

}

Status FileScanner::_process_runtime_filters_partition_prune(bool& can_filter_all) {

    SCOPED_TIMER(_runtime_filter_partition_prune_timer);

    if (_runtime_filter_partition_prune_ctxs.empty() || _partition_col_descs.empty()) {

        return Status::OK();

    }

    size_t partition_value_column_size = 1;

    // 1. Get partition key values to string columns.

    std::unordered_map<SlotId, MutableColumnPtr> partition_slot_id_to_column;

    for (auto const& partition_col_desc : _partition_col_descs) {

        const auto& [partition_value, partition_slot_desc] = partition_col_desc.second;

        auto data_type = partition_slot_desc->get_data_type_ptr();

        auto test_serde = data_type->get_serde();

        auto partition_value_column = data_type->create_column();

        auto* col_ptr = static_cast<IColumn*>(partition_value_column.get());

        Slice slice(partition_value.data(), partition_value.size());

        uint64_t num_deserialized = 0;

        DataTypeSerDe::FormatOptions options {};

        if (_partition_value_is_null.contains(partition_slot_desc->col_name())) {

            // for iceberg/paimon table

            // NOTICE: column is always be nullable for iceberg/paimon table now

            DCHECK(data_type->is_nullable());

            test_serde = test_serde->get_nested_serdes()[0];

            auto* null_column = assert_cast<ColumnNullable*>(col_ptr);

            if (_partition_value_is_null[partition_slot_desc->col_name()]) {

                null_column->insert_many_defaults(partition_value_column_size);

            } else {

                // If the partition value is not null, we set null map to 0 and deserialize it normally.

                null_column->get_null_map_column().insert_many_vals(0, partition_value_column_size);

                RETURN_IF_ERROR(test_serde->deserialize_column_from_fixed_json(

                        null_column->get_nested_column(), slice, partition_value_column_size,

                        &num_deserialized, options));

            }

        } else {

            // for hive/hudi table, the null value is set as "\\N"

            // TODO: this will be unified as iceberg/paimon table in the future

            RETURN_IF_ERROR(test_serde->deserialize_column_from_fixed_json(

                    *col_ptr, slice, partition_value_column_size, &num_deserialized, options));

        }

        partition_slot_id_to_column[partition_slot_desc->id()] = std::move(partition_value_column);

    }

    // 2. Fill _runtime_filter_partition_prune_block from the partition column, then execute conjuncts and filter block.

    // 2.1 Fill _runtime_filter_partition_prune_block from the partition column to match the conjuncts executing.

    size_t index = 0;

    bool first_column_filled = false;

    for (auto const* slot_desc : _real_tuple_desc->slots()) {

        if (partition_slot_id_to_column.find(slot_desc->id()) !=

            partition_slot_id_to_column.end()) {

            auto data_type = slot_desc->get_data_type_ptr();

            auto partition_value_column = std::move(partition_slot_id_to_column[slot_desc->id()]);

            if (data_type->is_nullable()) {

                _runtime_filter_partition_prune_block.insert(

                        index, ColumnWithTypeAndName(

                                       ColumnNullable::create(

                                               std::move(partition_value_column),

                                               ColumnUInt8::create(partition_value_column_size, 0)),

                                       data_type, slot_desc->col_name()));

            } else {

                _runtime_filter_partition_prune_block.insert(

                        index, ColumnWithTypeAndName(std::move(partition_value_column), data_type,

                                                     slot_desc->col_name()));

            }

            if (index == 0) {

                first_column_filled = true;

            }

        }

        index++;

    }

    // 2.2 Execute conjuncts.

    if (!first_column_filled) {

        // VExprContext.execute has an optimization, the filtering is executed when block->rows() > 0

        // The following process may be tricky and time-consuming, but we have no other way.

        auto column = IColumn::mutate(

                std::move(_runtime_filter_partition_prune_block.get_by_position(0).column));

        column->resize(partition_value_column_size);

        _runtime_filter_partition_prune_block.replace_by_position(0, std::move(column));

    }

    IColumn::Filter result_filter(_runtime_filter_partition_prune_block.rows(), 1);

    RETURN_IF_ERROR(VExprContext::execute_conjuncts(_runtime_filter_partition_prune_ctxs, nullptr,

                                                    &_runtime_filter_partition_prune_block,

                                                    &result_filter, &can_filter_all));

    return Status::OK();

}

Status FileScanner::_process_conjuncts() {

    _slot_id_to_filter_conjuncts.clear();

    _not_single_slot_filter_conjuncts.clear();

    for (auto& conjunct : _push_down_conjuncts) {

        auto impl = conjunct->root()->get_impl();

        // If impl is not null, which means this a conjuncts from runtime filter.

        auto cur_expr = impl ? impl : conjunct->root();

        std::vector<int> slot_ids;

        _get_slot_ids(cur_expr.get(), &slot_ids);

        if (slot_ids.empty()) {

            _not_single_slot_filter_conjuncts.emplace_back(conjunct);

            continue;

        }

        bool single_slot = true;

        for (int i = 1; i < slot_ids.size(); i++) {

            if (slot_ids[i] != slot_ids[0]) {

                single_slot = false;

                break;

            }

        }

        if (single_slot) {

            SlotId slot_id = slot_ids[0];

            _slot_id_to_filter_conjuncts[slot_id].emplace_back(conjunct);

        } else {

            _not_single_slot_filter_conjuncts.emplace_back(conjunct);

        }

    }

    return Status::OK();

}

Status FileScanner::_process_late_arrival_conjuncts() {

    if (_push_down_conjuncts.size() < _conjuncts.size()) {

        _push_down_conjuncts = _conjuncts;

        // Do not clear _conjuncts here!

        // We must keep it for fallback filtering, especially when mixing

        // Native readers (which use _push_down_conjuncts) and JNI readers (which rely on _conjuncts).

        // _conjuncts.clear();

        RETURN_IF_ERROR(_process_conjuncts());

    }

    if (_applied_rf_num == _total_rf_num) {

        _local_state->scanner_profile()->add_info_string("ApplyAllRuntimeFilters", "True");

    }

    return Status::OK();

}

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

    for (auto& child_expr : expr->children()) {

        if (child_expr->is_slot_ref()) {

            VSlotRef* slot_ref = reinterpret_cast<VSlotRef*>(child_expr.get());

            SlotDescriptor* slot_desc = _state->desc_tbl().get_slot_descriptor(slot_ref->slot_id());

            slot_desc->set_is_predicate(true);

            slot_ids->emplace_back(slot_ref->slot_id());

        } else {

            _get_slot_ids(child_expr.get(), slot_ids);

        }

    }

}

Status FileScanner::_open_impl(RuntimeState* state) {

    RETURN_IF_CANCELLED(state);

    RETURN_IF_ERROR(Scanner::_open_impl(state));

    if (_local_state) {

        _condition_cache_digest = _local_state->get_condition_cache_digest();

    }

    RETURN_IF_ERROR(_split_source->get_next(&_first_scan_range, &_current_range));

    if (_first_scan_range) {

        RETURN_IF_ERROR(_init_expr_ctxes());

        if (_state->query_options().enable_runtime_filter_partition_prune &&

            !_partition_slot_index_map.empty()) {

            _init_runtime_filter_partition_prune_ctxs();

            _init_runtime_filter_partition_prune_block();

        }

    } else {

        // there's no scan range in split source. stop scanner directly.

        _scanner_eof = true;

    }

    return Status::OK();

}

Status FileScanner::_get_block_impl(RuntimeState* state, Block* block, bool* eof) {

    Status st = _get_block_wrapped(state, block, eof);

    if (!st.ok()) {

        // add cur path in error msg for easy debugging

        return std::move(st.append(". cur path: " + get_current_scan_range_name()));

    }

    return st;

}

// For query:

//                              [exist cols]  [non-exist cols]  [col from path]  input  output

//                              A     B    C  D                 E

// _init_src_block              x     x    x  x                 x                -      x

// get_next_block               x     x    x  -                 -                -      x

// _cast_to_input_block         -     -    -  -                 -                -      -

// _fill_columns_from_path      -     -    -  -                 x                -      x

// _fill_missing_columns        -     -    -  x                 -                -      x

// _convert_to_output_block     -     -    -  -                 -                -      -

//

// For load:

//                              [exist cols]  [non-exist cols]  [col from path]  input  output

//                              A     B    C  D                 E

// _init_src_block              x     x    x  x                 x                x      -

// get_next_block               x     x    x  -                 -                x      -

// _cast_to_input_block         x     x    x  -                 -                x      -

// _fill_columns_from_path      -     -    -  -                 x                x      -

// _fill_missing_columns        -     -    -  x                 -                x      -

// _convert_to_output_block     -     -    -  -                 -                -      x

Status FileScanner::_get_block_wrapped(RuntimeState* state, Block* block, bool* eof) {

    do {

        RETURN_IF_CANCELLED(state);

        if (_cur_reader == nullptr || _cur_reader_eof) {

            _finalize_reader_condition_cache();

            // The file may not exist because the file list is got from meta cache,

            // And the file may already be removed from storage.

            // Just ignore not found files.

            Status st = _get_next_reader();

            if (st.is<ErrorCode::NOT_FOUND>() && config::ignore_not_found_file_in_external_table) {

                _cur_reader_eof = true;

                COUNTER_UPDATE(_not_found_file_counter, 1);

                continue;

            } else if (st.is<ErrorCode::END_OF_FILE>()) {

                _cur_reader_eof = true;

                COUNTER_UPDATE(_fully_skipped_file_counter, 1);

                continue;

            } else if (!st) {

                return st;

            }

            _init_reader_condition_cache();

        }

        if (_scanner_eof) {

            *eof = true;

            return Status::OK();

        }

        // Init src block for load job based on the data file schema (e.g. parquet)

        // For query job, simply set _src_block_ptr to block.

        size_t read_rows = 0;

        RETURN_IF_ERROR(_init_src_block(block));

        if (_should_run_adaptive_batch_size()) {

            _cur_reader->set_batch_size(_predict_reader_batch_rows());

        }

        {

            SCOPED_TIMER(_get_block_timer);

            // Read next block.

            // Some of column in block may not be filled (column not exist in file)

            RETURN_IF_ERROR(

                    _cur_reader->get_next_block(_src_block_ptr, &read_rows, &_cur_reader_eof));

        }

        // use read_rows instead of _src_block_ptr->rows(), because the first column of _src_block_ptr

        // may not be filled after calling `get_next_block()`, so _src_block_ptr->rows() may return wrong result.

        if (read_rows > 0) {

            if ((!_cur_reader->count_read_rows()) && _io_ctx) {

                _io_ctx->file_reader_stats->read_rows += read_rows;

            }

            // If the push_down_agg_type is COUNT, no need to do the rest,

            // because we only save a number in block.

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

                RETURN_IF_ERROR(_process_src_block_after_read(block));

            }

        }

        break;

    } while (true);

    // Update filtered rows and unselected rows for load, reset counter.

    // {

    //     state->update_num_rows_load_filtered(_counter.num_rows_filtered);

    //     state->update_num_rows_load_unselected(_counter.num_rows_unselected);

    //     _reset_counter();

    // }

    return Status::OK();

}

/**

 * Check whether there are complex types in parquet/orc reader in broker/stream load.

 * Broker/stream load will cast any type as string type, and complex types will be casted wrong.

 * This is a temporary method, and will be replaced by tvf.

 */

Status FileScanner::_check_output_block_types() {

    // Only called from _init_src_block_for_load, so _is_load is always true.

    TFileFormatType::type format_type = _params->format_type;

    if (format_type == TFileFormatType::FORMAT_PARQUET ||

        format_type == TFileFormatType::FORMAT_ORC) {

        for (auto slot : _output_tuple_desc->slots()) {

            if (is_complex_type(slot->type()->get_primitive_type())) {

                return Status::InternalError(

                        "Parquet/orc doesn't support complex types in broker/stream load, "

                        "please use tvf(table value function) to insert complex types.");

            }

        }

    }

    return Status::OK();

}

Status FileScanner::_init_src_block(Block* block) {

    DCHECK(_init_src_block_handler != nullptr);

    return (this->*_init_src_block_handler)(block);

}

Status FileScanner::_init_src_block_for_query(Block* block) {

    _src_block_ptr = block;

    // Build name to index map only once on first call.

    if (_src_block_name_to_idx.empty()) {

        _src_block_name_to_idx = block->get_name_to_pos_map();

    }

    return Status::OK();

}

Status FileScanner::_init_src_block_for_load(Block* block) {

    static_cast<void>(block);

    RETURN_IF_ERROR(_check_output_block_types());

    // if (_src_block_init) {

    //     _src_block.clear_column_data();

    //     _src_block_ptr = &_src_block;

    //     return Status::OK();

    // }

    _src_block.clear();

    uint32_t idx = 0;

    // slots in _input_tuple_desc contains all slots describe in load statement, eg:

    // -H "columns: k1, k2, tmp1, k3 = tmp1 + 1"

    // _input_tuple_desc will contains: k1, k2, tmp1

    // and some of them are from file, such as k1 and k2, and some of them may not exist in file, such as tmp1

    // _input_tuple_desc also contains columns from path

    for (auto& slot : _input_tuple_desc->slots()) {

        DataTypePtr data_type;

        auto it = _slot_lower_name_to_col_type.find(slot->col_name());

        if (slot->is_skip_bitmap_col()) {

            _skip_bitmap_col_idx = idx;

        }

        if (_params->__isset.sequence_map_col) {

            if (_params->sequence_map_col == slot->col_name()) {

                _sequence_map_col_uid = slot->col_unique_id();

            }

        }

        data_type =

                it == _slot_lower_name_to_col_type.end() ? slot->type() : make_nullable(it->second);

        MutableColumnPtr data_column = data_type->create_column();

        _src_block.insert(

                ColumnWithTypeAndName(std::move(data_column), data_type, slot->col_name()));

        _src_block_name_to_idx.emplace(slot->col_name(), idx++);

    }

    if (_params->__isset.sequence_map_col) {

        for (const auto& slot : _output_tuple_desc->slots()) {

            // When the target table has seqeunce map column, _input_tuple_desc will not contains __DORIS_SEQUENCE_COL__,

            // so we should get its column unique id from _output_tuple_desc

            if (slot->is_sequence_col()) {

                _sequence_col_uid = slot->col_unique_id();

            }

        }

    }

    _src_block_ptr = &_src_block;

    _src_block_init = true;

    return Status::OK();

}

Status FileScanner::_cast_to_input_block(Block* block) {

    // Only called from _process_src_block_after_read_for_load, so _is_load is always true.

    SCOPED_TIMER(_cast_to_input_block_timer);

    // cast primitive type(PT0) to primitive type(PT1)

    uint32_t idx = 0;

    for (auto& slot_desc : _input_tuple_desc->slots()) {

        if (_slot_lower_name_to_col_type.find(slot_desc->col_name()) ==

            _slot_lower_name_to_col_type.end()) {

            // skip columns which does not exist in file

            continue;

        }

        auto& arg = _src_block_ptr->get_by_position(_src_block_name_to_idx[slot_desc->col_name()]);

        auto return_type = slot_desc->get_data_type_ptr();

        // remove nullable here, let the get_function decide whether nullable

        auto data_type = get_data_type_with_default_argument(remove_nullable(return_type));

        ColumnsWithTypeAndName arguments {

                arg, {data_type->create_column(), data_type, slot_desc->col_name()}};

        auto func_cast =

                SimpleFunctionFactory::instance().get_function("CAST", arguments, return_type, {});

        if (!func_cast) {

            return Status::InternalError("Function CAST[arg={}, col name={}, return={}] not found!",

                                         arg.type->get_name(), slot_desc->col_name(),

                                         return_type->get_name());

        }

        idx = _src_block_name_to_idx[slot_desc->col_name()];

        DCHECK(_state != nullptr);

        auto ctx = FunctionContext::create_context(_state, {}, {});

        RETURN_IF_ERROR(

                func_cast->execute(ctx.get(), *_src_block_ptr, {idx}, idx, arg.column->size()));

        _src_block_ptr->get_by_position(idx).type = std::move(return_type);

    }

    return Status::OK();

}

Status FileScanner::_pre_filter_src_block() {

    // Only called from _process_src_block_after_read_for_load, so _is_load is always true.

    if (!_pre_conjunct_ctxs.empty()) {

        SCOPED_TIMER(_pre_filter_timer);

        auto origin_column_num = _src_block_ptr->columns();

        auto old_rows = _src_block_ptr->rows();

        RETURN_IF_ERROR(

                VExprContext::filter_block(_pre_conjunct_ctxs, _src_block_ptr, origin_column_num));

        _counter.num_rows_unselected += old_rows - _src_block_ptr->rows();

    }

    return Status::OK();

}

Status FileScanner::_convert_to_output_block(Block* block) {

    // Only called from _process_src_block_after_read_for_load, so _is_load is always true.

    SCOPED_TIMER(_convert_to_output_block_timer);

    // The block is passed from scanner context's free blocks,

    // which is initialized by output columns

    // so no need to clear it

    // block->clear();

    int ctx_idx = 0;

    size_t rows = _src_block_ptr->rows();

    auto filter_column = ColumnUInt8::create(rows, 1);

    auto& filter_map = filter_column->get_data();

    // After convert, the column_ptr should be copied into output block.

    // Can not use block->insert() because it may cause use_count() non-zero bug

    auto scoped_mutable_output_block =

            VectorizedUtils::build_scoped_mutable_mem_reuse_block(block, *_dest_row_desc);

    auto& mutable_output_block = scoped_mutable_output_block.mutable_block();

    auto& mutable_output_columns = mutable_output_block.mutable_columns();

    std::vector<BitmapValue>* skip_bitmaps {nullptr};

    MutableColumnPtr skip_bitmap_column;

    if (_should_process_skip_bitmap_col()) {

        skip_bitmap_column = IColumn::mutate(

                std::move(_src_block_ptr->get_by_position(_skip_bitmap_col_idx).column));

        auto* skip_bitmap_nullable_col_ptr = assert_cast<ColumnNullable*>(skip_bitmap_column.get());

        skip_bitmaps = &(assert_cast<ColumnBitmap*>(

                                 skip_bitmap_nullable_col_ptr->get_nested_column_ptr().get())

                                 ->get_data());

        // NOTE:

        // - If the table has sequence type column, __DORIS_SEQUENCE_COL__ will be put in _input_tuple_desc, so whether

        //   __DORIS_SEQUENCE_COL__ will be marked in skip bitmap depends on whether it's specified in that row

        // - If the table has sequence map column, __DORIS_SEQUENCE_COL__ will not be put in _input_tuple_desc,

        //   so __DORIS_SEQUENCE_COL__ will be ommited if it't specified in a row and will not be marked in skip bitmap.

        //   So we should mark __DORIS_SEQUENCE_COL__ in skip bitmap here if the corresponding sequence map column us marked

        if (_sequence_map_col_uid != -1) {

            for (int j = 0; j < rows; ++j) {

                if ((*skip_bitmaps)[j].contains(_sequence_map_col_uid)) {

                    (*skip_bitmaps)[j].add(_sequence_col_uid);

                }

            }

        }

        _src_block_ptr->replace_by_position(_skip_bitmap_col_idx, std::move(skip_bitmap_column));

    }

    // for (auto slot_desc : _output_tuple_desc->slots()) {

    for (int j = 0; j < mutable_output_columns.size(); ++j) {

        auto* slot_desc = _output_tuple_desc->slots()[j];

        int dest_index = ctx_idx;

        ColumnPtr column_ptr;

        auto& ctx = _dest_vexpr_ctx[dest_index];

        // PT1 => dest primitive type

        RETURN_IF_ERROR(ctx->execute(_src_block_ptr, column_ptr));

        // column_ptr maybe a ColumnConst, convert it to a normal column

        column_ptr = column_ptr->convert_to_full_column_if_const();

        DCHECK(column_ptr);

        // because of src_slot_desc is always be nullable, so the column_ptr after do dest_expr

        // is likely to be nullable

        if (LIKELY(column_ptr->is_nullable())) {

            const auto* nullable_column = reinterpret_cast<const ColumnNullable*>(column_ptr.get());

            for (int i = 0; i < rows; ++i) {

                if (filter_map[i] && nullable_column->is_null_at(i)) {

                    // skip checks for non-mentioned columns in flexible partial update

                    if (skip_bitmaps == nullptr ||

                        !skip_bitmaps->at(i).contains(slot_desc->col_unique_id())) {

                        // clang-format off

                        if (_strict_mode && (_src_slot_descs_order_by_dest[dest_index]) &&

                            !_src_block_ptr->get_by_position(_dest_slot_to_src_slot_index[dest_index]).column->is_null_at(i)) {

                            filter_map[i] = false;

                            RETURN_IF_ERROR(_state->append_error_msg_to_file(

                                [&]() -> std::string {

                                    return _src_block_ptr->dump_one_line(i, _num_of_columns_from_file);

                                },

                                [&]() -> std::string {

                                    auto raw_value =

                                            _src_block_ptr->get_by_position(_dest_slot_to_src_slot_index[dest_index]).column->get_data_at(i);

                                    std::string raw_string = raw_value.to_string();

                                    fmt::memory_buffer error_msg;

                                    fmt::format_to(error_msg,"column({}) value is incorrect while strict mode is {}, src value is {}",

                                            slot_desc->col_name(), _strict_mode, raw_string);

                                    return fmt::to_string(error_msg);

                                }));

                        } else if (!slot_desc->is_nullable()) {

                            filter_map[i] = false;

                            RETURN_IF_ERROR(_state->append_error_msg_to_file(

                                [&]() -> std::string {

                                    return _src_block_ptr->dump_one_line(i, _num_of_columns_from_file);

                                },

                                [&]() -> std::string {

                                    fmt::memory_buffer error_msg;

                                    fmt::format_to(error_msg, "column({}) values is null while columns is not nullable", slot_desc->col_name());

                                    return fmt::to_string(error_msg);

                                }));

                        }

                        // clang-format on

                    }

                }

            }

            if (!slot_desc->is_nullable()) {

                column_ptr = remove_nullable(column_ptr);

            }

        } else if (slot_desc->is_nullable()) {

            column_ptr = make_nullable(column_ptr);

        }

        mutable_output_columns[j]->insert_range_from(*column_ptr, 0, rows);

        ctx_idx++;

    }

    scoped_mutable_output_block.restore();

    // after do the dest block insert operation, clear _src_block to remove the reference of origin column

    _src_block_ptr->clear();

    size_t dest_size = block->columns();

    // do filter

    block->insert(ColumnWithTypeAndName(std::move(filter_column), std::make_shared<DataTypeUInt8>(),

                                        "filter column"));

    RETURN_IF_ERROR(Block::filter_block(block, dest_size, dest_size));

    _counter.num_rows_filtered += rows - block->rows();

    return Status::OK();

}

Status FileScanner::_process_src_block_after_read(Block* block) {

    DCHECK(_process_src_block_after_read_handler != nullptr);

    return (this->*_process_src_block_after_read_handler)(block);

}

Status FileScanner::_process_src_block_after_read_for_query(Block* block) {

    _update_adaptive_batch_size_before_truncate(*block);

    // Truncate CHAR/VARCHAR columns when target size is smaller than file schema.

    // This is needed for external table queries with truncate_char_or_varchar_columns=true.

    RETURN_IF_ERROR(_truncate_char_or_varchar_columns(block));

    _update_adaptive_batch_size_after_truncate(*block);

    return Status::OK();

}

Status FileScanner::_process_src_block_after_read_for_load(Block* block) {

    // Convert the src block columns type in-place.

    RETURN_IF_ERROR(_cast_to_input_block(block));

    // All readers fill partition and missing columns inside get_next_block:

    //   ORC/Parquet: in _do_get_next_block via on_fill_partition_columns/on_fill_missing_columns

    //   CSV/JSON/others: in on_after_read_block via fill_remaining_columns

    // Assert all columns have consistent row counts.

    if (_src_block_ptr->columns() > 0) {

        size_t rows = _src_block_ptr->get_by_position(0).column->size();

        for (size_t i = 1; i < _src_block_ptr->columns(); ++i) {

            DCHECK_EQ(_src_block_ptr->get_by_position(i).column->size(), rows)

                    << "Column " << _src_block_ptr->get_by_position(i).name << " has "

                    << _src_block_ptr->get_by_position(i).column->size() << " rows, expected "

                    << rows;

        }

    }

    // Apply _pre_conjunct_ctxs to filter src block.

    RETURN_IF_ERROR(_pre_filter_src_block());

    // Convert src block to output block (dest block), then apply filters.

    RETURN_IF_ERROR(_convert_to_output_block(block));

    _update_adaptive_batch_size_before_truncate(*block);

    // Truncate CHAR/VARCHAR columns when target size is smaller than file schema.

    RETURN_IF_ERROR(_truncate_char_or_varchar_columns(block));

    _update_adaptive_batch_size_after_truncate(*block);

    return Status::OK();

}