// 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 "format/csv/csv_reader.h"

#include <fmt/format.h>
#include <gen_cpp/PlanNodes_types.h>
#include <gen_cpp/Types_types.h>
#include <glog/logging.h>

#include <algorithm>
#include <cstddef>
#include <map>
#include <memory>
#include <numeric>
#include <ostream>
#include <regex>
#include <utility>

#include "common/compiler_util.h" // IWYU pragma: keep
#include "common/config.h"
#include "common/consts.h"
#include "common/status.h"
#include "core/block/block.h"
#include "core/block/column_with_type_and_name.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type_serde/data_type_string_serde.h"
#include "exec/scan/scanner.h"
#include "format/file_reader/new_plain_binary_line_reader.h"
#include "format/file_reader/new_plain_text_line_reader.h"
#include "format/line_reader.h"
#include "io/file_factory.h"
#include "io/fs/broker_file_reader.h"
#include "io/fs/buffered_reader.h"
#include "io/fs/file_reader.h"
#include "io/fs/s3_file_reader.h"
#include "io/fs/tracing_file_reader.h"
#include "runtime/descriptors.h"
#include "runtime/runtime_state.h"
#include "util/decompressor.h"
#include "util/string_util.h"
#include "util/utf8_check.h"

namespace doris {
class RuntimeProfile;
class IColumn;
namespace io {
struct IOContext;
enum class FileCachePolicy : uint8_t;
} // namespace io
} // namespace doris

namespace doris {

namespace {

size_t columns_byte_size(const std::vector<MutableColumnPtr>& columns) {
    size_t bytes = 0;
    for (const auto& column : columns) {
        DCHECK(column.get() != nullptr);
        bytes += column->byte_size();
    }
    return bytes;
}

} // namespace

void EncloseCsvTextFieldSplitter::do_split(const Slice& line, std::vector<Slice>* splitted_values) {
    const char* data = line.data;
    const auto& column_sep_positions = _text_line_reader_ctx->column_sep_positions();
    size_t value_start_offset = 0;
    for (auto idx : column_sep_positions) {
        process_value_func(data, value_start_offset, idx - value_start_offset, _trimming_char,
                           splitted_values);
        value_start_offset = idx + _value_sep_len;
    }
    if (line.size >= value_start_offset) {
        // process the last column
        process_value_func(data, value_start_offset, line.size - value_start_offset, _trimming_char,
                           splitted_values);
    }
}

void PlainCsvTextFieldSplitter::_split_field_single_char(const Slice& line,
                                                         std::vector<Slice>* splitted_values) {
    const char* data = line.data;
    const size_t size = line.size;
    size_t value_start = 0;
    for (size_t i = 0; i < size; ++i) {
        if (data[i] == _value_sep[0]) {
            process_value_func(data, value_start, i - value_start, _trimming_char, splitted_values);
            value_start = i + _value_sep_len;
        }
    }
    process_value_func(data, value_start, size - value_start, _trimming_char, splitted_values);
}

void PlainCsvTextFieldSplitter::_split_field_multi_char(const Slice& line,
                                                        std::vector<Slice>* splitted_values) {
    size_t start = 0;  // point to the start pos of next col value.
    size_t curpos = 0; // point to the start pos of separator matching sequence.

    // value_sep : AAAA
    // line.data : 1234AAAA5678
    // -> 1234,5678

    //    start   start
    //      ▼       ▼
    //      1234AAAA5678\0
    //          ▲       ▲
    //      curpos     curpos

    //kmp
    std::vector<int> next(_value_sep_len);
    next[0] = -1;
    for (int i = 1, j = -1; i < _value_sep_len; i++) {
        while (j > -1 && _value_sep[i] != _value_sep[j + 1]) {
            j = next[j];
        }
        if (_value_sep[i] == _value_sep[j + 1]) {
            j++;
        }
        next[i] = j;
    }

    for (int i = 0, j = -1; i < line.size; i++) {
        // i : line
        // j : _value_sep
        while (j > -1 && line[i] != _value_sep[j + 1]) {
            j = next[j];
        }
        if (line[i] == _value_sep[j + 1]) {
            j++;
        }
        if (j == _value_sep_len - 1) {
            curpos = i - _value_sep_len + 1;

            /*
             * column_separator : "xx"
             * data.csv :  data1xxxxdata2
             *
             * Parse incorrectly:
             *      data1[xx]xxdata2
             *      data1x[xx]xdata2
             *      data1xx[xx]data2
             * The string "xxxx" is parsed into three "xx" delimiters.
             *
             * Parse correctly:
             *      data1[xx]xxdata2
             *      data1xx[xx]data2
             */

            if (curpos >= start) {
                process_value_func(line.data, start, curpos - start, _trimming_char,
                                   splitted_values);
                start = i + 1;
            }

            j = next[j];
        }
    }
    process_value_func(line.data, start, line.size - start, _trimming_char, splitted_values);
}

void PlainCsvTextFieldSplitter::do_split(const Slice& line, std::vector<Slice>* splitted_values) {
    if (is_single_char_delim) {
        _split_field_single_char(line, splitted_values);
    } else {
        _split_field_multi_char(line, splitted_values);
    }
}

CsvReader::CsvReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
                     const TFileScanRangeParams& params, const TFileRangeDesc& range,
                     const std::vector<SlotDescriptor*>& file_slot_descs, size_t batch_size,
                     io::IOContext* io_ctx, std::shared_ptr<io::IOContext> io_ctx_holder)
        : _profile(profile),
          _params(params),
          _file_reader(nullptr),
          _line_reader(nullptr),
          _decompressor(nullptr),
          _state(state),
          _counter(counter),
          _range(range),
          _file_slot_descs(file_slot_descs),
          _line_reader_eof(false),
          _skip_lines(0),
          _io_ctx(io_ctx),
          _io_ctx_holder(std::move(io_ctx_holder)),
          _batch_size(std::max(batch_size, 1UL)) {
    if (_io_ctx == nullptr && _io_ctx_holder) {
        _io_ctx = _io_ctx_holder.get();
    }
    _file_format_type = _params.format_type;
    _is_proto_format = _file_format_type == TFileFormatType::FORMAT_PROTO;
    if (_range.__isset.compress_type) {
        // for compatibility
        _file_compress_type = _range.compress_type;
    } else {
        _file_compress_type = _params.compress_type;
    }
    _size = _range.size;

    _split_values.reserve(_file_slot_descs.size());
    _init_system_properties();
    _init_file_description();
    _serdes = create_data_type_serdes(_file_slot_descs);
}

void CsvReader::_init_system_properties() {
    if (_range.__isset.file_type) {
        // for compatibility
        _system_properties.system_type = _range.file_type;
    } else {
        _system_properties.system_type = _params.file_type;
    }
    _system_properties.properties = _params.properties;
    _system_properties.hdfs_params = _params.hdfs_params;
    if (_params.__isset.broker_addresses) {
        _system_properties.broker_addresses.assign(_params.broker_addresses.begin(),
                                                   _params.broker_addresses.end());
    }
}

void CsvReader::_init_file_description() {
    _file_description.path = _range.path;
    _file_description.file_size = _range.__isset.file_size ? _range.file_size : -1;
    if (_range.__isset.fs_name) {
        _file_description.fs_name = _range.fs_name;
    }
    if (_range.__isset.file_cache_admission) {
        _file_description.file_cache_admission = _range.file_cache_admission;
    }
}

Status CsvReader::init_reader(bool is_load) {
    // set the skip lines and start offset
    _start_offset = _range.start_offset;
    if (_start_offset == 0) {
        // check header typer first
        if (_params.__isset.file_attributes && _params.file_attributes.__isset.header_type &&
            !_params.file_attributes.header_type.empty()) {
            std::string header_type = to_lower(_params.file_attributes.header_type);
            if (header_type == BeConsts::CSV_WITH_NAMES) {
                _skip_lines = 1;
            } else if (header_type == BeConsts::CSV_WITH_NAMES_AND_TYPES) {
                _skip_lines = 2;
            }
        } else if (_params.file_attributes.__isset.skip_lines) {
            _skip_lines = _params.file_attributes.skip_lines;
        }
    } else if (_start_offset != 0) {
        if ((_file_compress_type != TFileCompressType::PLAIN) ||
            (_file_compress_type == TFileCompressType::UNKNOWN &&
             _file_format_type != TFileFormatType::FORMAT_CSV_PLAIN)) {
            return Status::InternalError<false>("For now we do not support split compressed file");
        }
        // pre-read to promise first line skipped always read
        int64_t pre_read_len = std::min(
                static_cast<int64_t>(_params.file_attributes.text_params.line_delimiter.size()),
                _start_offset);
        _start_offset -= pre_read_len;
        _size += pre_read_len;
        // not first range will always skip one line
        _skip_lines = 1;
    }

    _use_nullable_string_opt.resize(_file_slot_descs.size());
    for (int i = 0; i < _file_slot_descs.size(); ++i) {
        auto data_type_ptr = _file_slot_descs[i]->get_data_type_ptr();
        if (data_type_ptr->is_nullable() && is_string_type(data_type_ptr->get_primitive_type())) {
            _use_nullable_string_opt[i] = 1;
        }
    }

    RETURN_IF_ERROR(_init_options());
    RETURN_IF_ERROR(_create_file_reader(false));
    RETURN_IF_ERROR(_create_decompressor());
    RETURN_IF_ERROR(_create_line_reader());

    _is_load = is_load;
    if (!_is_load) {
        // For query task, there are 2 slot mapping.
        // One is from file slot to values in line.
        //      eg, the file_slot_descs is k1, k3, k5, and values in line are k1, k2, k3, k4, k5
        //      the _col_idxs will save: 0, 2, 4
        // The other is from file slot to columns in output block
        //      eg, the file_slot_descs is k1, k3, k5, and columns in block are p1, k1, k3, k5
        //      where "p1" is the partition col which does not exist in file
        //      the _file_slot_idx_map will save: 1, 2, 3
        DCHECK(_params.__isset.column_idxs);
        _col_idxs = _params.column_idxs;
        int idx = 0;
        for (const auto& slot_info : _params.required_slots) {
            if (slot_info.is_file_slot) {
                _file_slot_idx_map.push_back(idx);
            }
            idx++;
        }
    } else {
        // For load task, the column order is same as file column order
        int i = 0;
        for (const auto& desc [[maybe_unused]] : _file_slot_descs) {
            _col_idxs.push_back(i++);
        }
    }

    _line_reader_eof = false;
    return Status::OK();
}

// ---- Unified init_reader(ReaderInitContext*) overrides ----

Status CsvReader::_open_file_reader(ReaderInitContext* base_ctx) {
    _start_offset = _range.start_offset;
    if (_start_offset == 0) {
        if (_params.__isset.file_attributes && _params.file_attributes.__isset.header_type &&
            !_params.file_attributes.header_type.empty()) {
            std::string header_type = to_lower(_params.file_attributes.header_type);
            if (header_type == BeConsts::CSV_WITH_NAMES) {
                _skip_lines = 1;
            } else if (header_type == BeConsts::CSV_WITH_NAMES_AND_TYPES) {
                _skip_lines = 2;
            }
        } else if (_params.file_attributes.__isset.skip_lines) {
            _skip_lines = _params.file_attributes.skip_lines;
        }
    } else if (_start_offset != 0) {
        if ((_file_compress_type != TFileCompressType::PLAIN) ||
            (_file_compress_type == TFileCompressType::UNKNOWN &&
             _file_format_type != TFileFormatType::FORMAT_CSV_PLAIN)) {
            return Status::InternalError<false>("For now we do not support split compressed file");
        }
        int64_t pre_read_len = std::min(
                static_cast<int64_t>(_params.file_attributes.text_params.line_delimiter.size()),
                _start_offset);
        _start_offset -= pre_read_len;
        _size += pre_read_len;
        _skip_lines = 1;
    }

    RETURN_IF_ERROR(_init_options());
    RETURN_IF_ERROR(_create_file_reader(false));
    return Status::OK();
}

Status CsvReader::_do_init_reader(ReaderInitContext* base_ctx) {
    auto* ctx = checked_context_cast<CsvInitContext>(base_ctx);
    _is_load = ctx->is_load;

    _use_nullable_string_opt.resize(_file_slot_descs.size());
    for (int i = 0; i < _file_slot_descs.size(); ++i) {
        auto data_type_ptr = _file_slot_descs[i]->get_data_type_ptr();
        if (data_type_ptr->is_nullable() && is_string_type(data_type_ptr->get_primitive_type())) {
            _use_nullable_string_opt[i] = 1;
        }
    }

    RETURN_IF_ERROR(_create_decompressor());
    RETURN_IF_ERROR(_create_line_reader());

    if (!_is_load) {
        DCHECK(_params.__isset.column_idxs);
        _col_idxs = _params.column_idxs;
        int idx = 0;
        for (const auto& slot_info : _params.required_slots) {
            if (slot_info.is_file_slot) {
                _file_slot_idx_map.push_back(idx);
            }
            idx++;
        }
    } else {
        int i = 0;
        for (const auto& desc [[maybe_unused]] : _file_slot_descs) {
            _col_idxs.push_back(i++);
        }
    }
    _line_reader_eof = false;
    return Status::OK();
}

void CsvReader::set_batch_size(size_t batch_size) {
    // 0 means "not set" / "use default" for the row-based readers; we must
    // never let _batch_size be 0 because _do_get_next_block uses it as the
    // upper bound of a `while (rows < _batch_size)` loop and a 0 would make
    // the reader return empty blocks and incorrectly signal EOF.
    _batch_size = std::max(batch_size, 1UL);
}

// !FIXME: Here we should use MutableBlock
Status CsvReader::_do_get_next_block(Block* block, size_t* read_rows, bool* eof) {
    if (_line_reader_eof) {
        *eof = true;
        return Status::OK();
    }

    const size_t batch_size = _batch_size;
    const auto max_block_bytes = _state->preferred_block_size_bytes();
    size_t rows = 0;

    bool success = false;
    bool is_remove_bom = false;
    if (_push_down_agg_type == TPushAggOp::type::COUNT) {
        while (rows < batch_size && !_line_reader_eof) {
            const uint8_t* ptr = nullptr;
            size_t size = 0;
            RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size, &_line_reader_eof, _io_ctx));

            // _skip_lines == 0 means this line is the actual data beginning line for the entire file
            // is_remove_bom means _remove_bom should only execute once
            if (_skip_lines == 0 && !is_remove_bom) {
                ptr = _remove_bom(ptr, size);
                is_remove_bom = true;
            }

            // _skip_lines > 0 means we do not need to remove bom
            if (_skip_lines > 0) {
                _skip_lines--;
                is_remove_bom = true;
                continue;
            }
            if (size == 0) {
                if (!_line_reader_eof) {
                    if (_empty_line_as_record() || _state->is_read_csv_empty_line_as_null()) {
                        ++rows;
                    }
                }
                // Read empty line, continue
                continue;
            }

            RETURN_IF_ERROR(_validate_line(Slice(ptr, size), &success));
            ++rows;
        }
        auto mutable_columns_guard = block->mutate_columns_scoped();
        auto& mutate_columns = mutable_columns_guard.mutable_columns();
        for (auto& col : mutate_columns) {
            col->resize(rows);
        }
    } else {
        auto columns_guard = block->mutate_columns_scoped();
        auto& columns = columns_guard.mutable_columns();
        _reserve_nullable_string_columns(columns, batch_size);
        while (rows < batch_size && !_line_reader_eof &&
               (columns_byte_size(columns) < max_block_bytes)) {
            const uint8_t* ptr = nullptr;
            size_t size = 0;
            RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size, &_line_reader_eof, _io_ctx));

            // _skip_lines == 0 means this line is the actual data beginning line for the entire file
            // is_remove_bom means _remove_bom should only execute once
            if (!is_remove_bom && _skip_lines == 0) {
                ptr = _remove_bom(ptr, size);
                is_remove_bom = true;
            }

            // _skip_lines > 0 means we do not remove bom
            if (_skip_lines > 0) {
                _skip_lines--;
                is_remove_bom = true;
                continue;
            }
            if (size == 0) {
                if (!_line_reader_eof) {
                    if (_empty_line_as_record()) {
                        Slice empty_line("", 0);
                        RETURN_IF_ERROR(_validate_line(empty_line, &success));
                        if (success) {
                            RETURN_IF_ERROR(_fill_dest_columns(empty_line, columns, &rows));
                        }
                    } else if (_state->is_read_csv_empty_line_as_null()) {
                        RETURN_IF_ERROR(_fill_empty_line(columns, &rows));
                    }
                }
                // Read empty line, continue
                continue;
            }

            RETURN_IF_ERROR(_validate_line(Slice(ptr, size), &success));
            if (!success) {
                continue;
            }
            RETURN_IF_ERROR(_fill_dest_columns(Slice(ptr, size), columns, &rows));
        }
    }

    *eof = (rows == 0);
    *read_rows = rows;

    return Status::OK();
}

Status CsvReader::_get_columns_impl(std::unordered_map<std::string, DataTypePtr>* name_to_type) {
    for (const auto& slot : _file_slot_descs) {
        name_to_type->emplace(slot->col_name(), slot->type());
    }
    return Status::OK();
}

// init decompressor, file reader and line reader for parsing schema
Status CsvReader::init_schema_reader() {
    _start_offset = _range.start_offset;
    if (_start_offset != 0) {
        return Status::InvalidArgument(
                "start offset of TFileRangeDesc must be zero in get parsered schema");
    }
    if (_params.file_type == TFileType::FILE_BROKER) {
        return Status::InternalError<false>(
                "Getting parsered schema from csv file do not support stream load and broker "
                "load.");
    }

    // csv file without names line and types line.
    _read_line = 1;
    _is_parse_name = false;

    if (_params.__isset.file_attributes && _params.file_attributes.__isset.header_type &&
        !_params.file_attributes.header_type.empty()) {
        std::string header_type = to_lower(_params.file_attributes.header_type);
        if (header_type == BeConsts::CSV_WITH_NAMES) {
            _is_parse_name = true;
        } else if (header_type == BeConsts::CSV_WITH_NAMES_AND_TYPES) {
            _read_line = 2;
            _is_parse_name = true;
        }
    }

    RETURN_IF_ERROR(_init_options());
    RETURN_IF_ERROR(_create_file_reader(true));
    RETURN_IF_ERROR(_create_decompressor());
    RETURN_IF_ERROR(_create_line_reader());
    return Status::OK();
}

Status CsvReader::get_parsed_schema(std::vector<std::string>* col_names,
                                    std::vector<DataTypePtr>* col_types) {
    if (_read_line == 1) {
        if (!_is_parse_name) { //parse csv file without names and types
            size_t col_nums = 0;
            RETURN_IF_ERROR(_parse_col_nums(&col_nums));
            for (size_t i = 0; i < col_nums; ++i) {
                col_names->emplace_back("c" + std::to_string(i + 1));
            }
        } else { // parse csv file with names
            RETURN_IF_ERROR(_parse_col_names(col_names));
        }

        for (size_t j = 0; j < col_names->size(); ++j) {
            col_types->emplace_back(
                    DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_STRING, true));
        }
    } else { // parse csv file with names and types
        RETURN_IF_ERROR(_parse_col_names(col_names));
        RETURN_IF_ERROR(_parse_col_types(col_names->size(), col_types));
    }
    return Status::OK();
}

Status CsvReader::_deserialize_nullable_string(IColumn& column, Slice& slice) {
    // This is the per-row per-column hot path of CSV load (load reads every column as
    // nullable string). The column type was already verified by the checked assert_cast
    // in _reserve_nullable_string_columns at the beginning of the batch, so the casts
    // here can skip the release-build typeid check.
    auto& null_column = assert_cast<ColumnNullable&, TypeCheckOnRelease::DISABLE>(column);
    auto& string_column = assert_cast<ColumnString&, TypeCheckOnRelease::DISABLE>(
            null_column.get_nested_column());
    if (_empty_field_as_null && slice.size == 0) {
        string_column.insert_default();
        null_column.get_null_map_data().push_back(1);
        return Status::OK();
    }
    if (_options.null_len > 0 && !(_options.converted_from_string && slice.trim_double_quotes())) {
        if (slice.compare(Slice(_options.null_format, _options.null_len)) == 0) {
            string_column.insert_default();
            null_column.get_null_map_data().push_back(1);
            return Status::OK();
        }
    }
    // Same as DataTypeStringSerDe::deserialize_one_cell_from_csv (which never fails),
    // written out here to skip the SerDe layer and its per-cell assert_cast.
    if (_options.escape_char != 0) {
        escape_string_for_csv(slice.data, &slice.size, _options.escape_char, _options.quote_char);
    }
    string_column.insert_data(slice.data, slice.size);
    null_column.get_null_map_data().push_back(0);
    return Status::OK();
}

Status CsvReader::_init_options() {
    // get column_separator and line_delimiter
    _value_separator = _params.file_attributes.text_params.column_separator;
    _value_separator_length = _value_separator.size();
    _line_delimiter = _params.file_attributes.text_params.line_delimiter;
    _line_delimiter_length = _line_delimiter.size();
    if (_params.file_attributes.text_params.__isset.enclose) {
        _enclose = _params.file_attributes.text_params.enclose;
    }
    if (_params.file_attributes.text_params.__isset.escape) {
        _escape = _params.file_attributes.text_params.escape;
    }

    _trim_tailing_spaces =
            (_state != nullptr && _state->trim_tailing_spaces_for_external_table_query());

    _options.escape_char = _escape;
    _options.quote_char = _enclose;

    if (_params.file_attributes.text_params.collection_delimiter.empty()) {
        _options.collection_delim = ',';
    } else {
        _options.collection_delim = _params.file_attributes.text_params.collection_delimiter[0];
    }
    if (_params.file_attributes.text_params.mapkv_delimiter.empty()) {
        _options.map_key_delim = ':';
    } else {
        _options.map_key_delim = _params.file_attributes.text_params.mapkv_delimiter[0];
    }

    if (_params.file_attributes.text_params.__isset.null_format) {
        _options.null_format = _params.file_attributes.text_params.null_format.data();
        _options.null_len = _params.file_attributes.text_params.null_format.length();
    }

    if (_params.file_attributes.__isset.trim_double_quotes) {
        _trim_double_quotes = _params.file_attributes.trim_double_quotes;
    }
    _options.converted_from_string = _trim_double_quotes;

    if (_state != nullptr) {
        _keep_cr = _state->query_options().keep_carriage_return;
    }

    if (_params.file_attributes.text_params.__isset.empty_field_as_null) {
        _empty_field_as_null = _params.file_attributes.text_params.empty_field_as_null;
    }
    return Status::OK();
}

Status CsvReader::_create_decompressor() {
    if (_file_compress_type != TFileCompressType::UNKNOWN) {
        RETURN_IF_ERROR(Decompressor::create_decompressor(_file_compress_type, &_decompressor));
    } else {
        RETURN_IF_ERROR(Decompressor::create_decompressor(_file_format_type, &_decompressor));
    }

    return Status::OK();
}

Status CsvReader::_create_file_reader(bool need_schema) {
    if (_params.file_type == TFileType::FILE_STREAM) {
        RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state,
                                                        need_schema));
    } else {
        _file_description.mtime = _range.__isset.modification_time ? _range.modification_time : 0;
        io::FileReaderOptions reader_options = FileFactory::get_reader_options(
                _state ? _state->query_options() : _default_query_options, _file_description);
        io::FileReaderSPtr file_reader;
        if (_io_ctx_holder) {
            file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
                    _profile, _system_properties, _file_description, reader_options,
                    io::DelegateReader::AccessMode::SEQUENTIAL,
                    std::static_pointer_cast<const io::IOContext>(_io_ctx_holder),
                    io::PrefetchRange(_range.start_offset, _range.start_offset + _range.size)));
        } else {
            file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
                    _profile, _system_properties, _file_description, reader_options,
                    io::DelegateReader::AccessMode::SEQUENTIAL, _io_ctx,
                    io::PrefetchRange(_range.start_offset, _range.start_offset + _range.size)));
        }
        _file_reader = _io_ctx && _io_ctx->file_reader_stats
                               ? std::make_shared<io::TracingFileReader>(std::move(file_reader),
                                                                         _io_ctx->file_reader_stats)
                               : file_reader;
    }
    if (_file_reader->size() == 0 && _params.file_type != TFileType::FILE_STREAM &&
        _params.file_type != TFileType::FILE_BROKER) {
        return Status::EndOfFile("init reader failed, empty csv file: " + _range.path);
    }
    return Status::OK();
}

Status CsvReader::_create_line_reader() {
    std::shared_ptr<TextLineReaderContextIf> text_line_reader_ctx;
    if (_enclose == 0) {
        text_line_reader_ctx = std::make_shared<PlainTextLineReaderCtx>(
                _line_delimiter, _line_delimiter_length, _keep_cr);
        _fields_splitter = std::make_unique<PlainCsvTextFieldSplitter>(
                _trim_tailing_spaces, false, _value_separator, _value_separator_length, -1);

    } else {
        // in load task, the _file_slot_descs is empty vector, so we need to set col_sep_num to 0
        size_t col_sep_num = _file_slot_descs.size() > 1 ? _file_slot_descs.size() - 1 : 0;
        _enclose_reader_ctx = std::make_shared<EncloseCsvLineReaderCtx>(
                _line_delimiter, _line_delimiter_length, _value_separator, _value_separator_length,
                col_sep_num, _enclose, _escape, _keep_cr);
        text_line_reader_ctx = _enclose_reader_ctx;

        _fields_splitter = std::make_unique<EncloseCsvTextFieldSplitter>(
                _trim_tailing_spaces, true, _enclose_reader_ctx, _value_separator_length, _enclose);
    }
    switch (_file_format_type) {
    case TFileFormatType::FORMAT_CSV_PLAIN:
        [[fallthrough]];
    case TFileFormatType::FORMAT_CSV_GZ:
        [[fallthrough]];
    case TFileFormatType::FORMAT_CSV_BZ2:
        [[fallthrough]];
    case TFileFormatType::FORMAT_CSV_LZ4FRAME:
        [[fallthrough]];
    case TFileFormatType::FORMAT_CSV_LZ4BLOCK:
        [[fallthrough]];
    case TFileFormatType::FORMAT_CSV_LZOP:
        [[fallthrough]];
    case TFileFormatType::FORMAT_CSV_SNAPPYBLOCK:
        [[fallthrough]];
    case TFileFormatType::FORMAT_CSV_DEFLATE:
        _line_reader =
                NewPlainTextLineReader::create_unique(_profile, _file_reader, _decompressor.get(),
                                                      text_line_reader_ctx, _size, _start_offset);

        break;
    case TFileFormatType::FORMAT_PROTO:
        _fields_splitter = std::make_unique<CsvProtoFieldSplitter>();
        _line_reader = NewPlainBinaryLineReader::create_unique(_file_reader);
        break;
    default:
        return Status::InternalError<false>(
                "Unknown format type, cannot init line reader in csv reader, type={}",
                _file_format_type);
    }
    return Status::OK();
}

Status CsvReader::_deserialize_one_cell(DataTypeSerDeSPtr serde, IColumn& column, Slice& slice) {
    return serde->deserialize_one_cell_from_csv(column, slice, _options);
}

Status CsvReader::_fill_dest_columns(const Slice& line, std::vector<MutableColumnPtr>& columns,
                                     size_t* rows) {
    bool is_success = false;

    RETURN_IF_ERROR(_line_split_to_values(line, &is_success));
    if (UNLIKELY(!is_success)) {
        // If not success, which means we met an invalid row, filter this row and return.
        return Status::OK();
    }

    for (int i = 0; i < _file_slot_descs.size(); ++i) {
        int col_idx = _col_idxs[i];
        // col idx is out of range, fill with null format
        auto value = col_idx < _split_values.size()
                             ? _split_values[col_idx]
                             : Slice(_options.null_format, _options.null_len);

        IColumn* col_ptr = columns[i].get();
        if (!_is_load) {
            col_ptr = columns[_file_slot_idx_map[i]].get();
        }

        if (_use_nullable_string_opt[i]) {
            // For load task, we always read "string" from file.
            // So serdes[i] here must be DataTypeNullableSerDe, and DataTypeNullableSerDe -> nested_serde must be DataTypeStringSerDe.
            // So we use deserialize_nullable_string and stringSerDe to reduce virtual function calls.
            RETURN_IF_ERROR(_deserialize_nullable_string(*col_ptr, value));
        } else {
            RETURN_IF_ERROR(_deserialize_one_cell(_serdes[i], *col_ptr, value));
        }
    }
    ++(*rows);

    return Status::OK();
}

void CsvReader::_reserve_nullable_string_columns(std::vector<MutableColumnPtr>& columns,
                                                 size_t batch_size) {
    for (int i = 0; i < _file_slot_descs.size(); ++i) {
        if (!_use_nullable_string_opt[i]) {
            continue;
        }
        IColumn* col_ptr = _is_load ? columns[i].get() : columns[_file_slot_idx_map[i]].get();
        // The checked casts here (once per batch) guarantee the column types for the
        // unchecked per-row casts in _deserialize_nullable_string.
        auto& null_column = assert_cast<ColumnNullable&>(*col_ptr);
        auto& string_column = assert_cast<ColumnString&>(null_column.get_nested_column());
        // Reserve up front so the per-row loop does not pay for incremental growth.
        // The string chars are not reserved because their total size is unpredictable.
        string_column.get_offsets().reserve(string_column.size() + batch_size);
        null_column.get_null_map_data().reserve(null_column.get_null_map_data().size() +
                                                batch_size);
    }
}

Status CsvReader::_fill_empty_line(std::vector<MutableColumnPtr>& columns, size_t* rows) {
    for (int i = 0; i < _file_slot_descs.size(); ++i) {
        IColumn* col_ptr = columns[i].get();
        if (!_is_load) {
            col_ptr = columns[_file_slot_idx_map[i]].get();
        }
        auto& null_column = assert_cast<ColumnNullable&>(*col_ptr);
        null_column.insert_data(nullptr, 0);
    }
    ++(*rows);
    return Status::OK();
}

Status CsvReader::_validate_line(const Slice& line, bool* success) {
    if (!_is_proto_format && !validate_utf8(_params, line.data, line.size)) {
        if (!_is_load) {
            return Status::InternalError<false>("Only support csv data in utf8 codec");
        } else {
            _counter->num_rows_filtered++;
            *success = false;
            RETURN_IF_ERROR(_state->append_error_msg_to_file(
                    [&]() -> std::string { return std::string(line.data, line.size); },
                    [&]() -> std::string {
                        return "Invalid file encoding: all CSV files must be UTF-8 encoded";
                    }));
            return Status::OK();
        }
    }
    *success = true;
    return Status::OK();
}

Status CsvReader::_line_split_to_values(const Slice& line, bool* success) {
    _split_line(line);

    if (_is_load) {
        // Only check for load task. For query task, the non exist column will be filled "null".
        // if actual column number in csv file is not equal to _file_slot_descs.size()
        // then filter this line.
        bool ignore_col = false;
        ignore_col = _params.__isset.file_attributes &&
                     _params.file_attributes.__isset.ignore_csv_redundant_col &&
                     _params.file_attributes.ignore_csv_redundant_col;

        if ((!ignore_col && _split_values.size() != _file_slot_descs.size()) ||
            (ignore_col && _split_values.size() < _file_slot_descs.size())) {
            _counter->num_rows_filtered++;
            *success = false;
            RETURN_IF_ERROR(_state->append_error_msg_to_file(
                    [&]() -> std::string { return std::string(line.data, line.size); },
                    [&]() -> std::string {
                        fmt::memory_buffer error_msg;
                        fmt::format_to(error_msg,
                                       "Column count mismatch: expected {}, but found {}",
                                       _file_slot_descs.size(), _split_values.size());
                        std::string escaped_separator =
                                std::regex_replace(_value_separator, std::regex("\t"), "\\t");
                        std::string escaped_delimiter =
                                std::regex_replace(_line_delimiter, std::regex("\n"), "\\n");
                        fmt::format_to(error_msg, " (sep:{} delim:{}", escaped_separator,
                                       escaped_delimiter);
                        if (_enclose != 0) {
                            fmt::format_to(error_msg, " encl:{}", _enclose);
                        }
                        if (_escape != 0) {
                            fmt::format_to(error_msg, " esc:{}", _escape);
                        }
                        fmt::format_to(error_msg, ")");
                        return fmt::to_string(error_msg);
                    }));
            return Status::OK();
        }
    }

    *success = true;
    return Status::OK();
}

void CsvReader::_split_line(const Slice& line) {
    _split_values.clear();
    _fields_splitter->split_line(line, &_split_values);
}

Status CsvReader::_parse_col_nums(size_t* col_nums) {
    const uint8_t* ptr = nullptr;
    size_t size = 0;
    RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size, &_line_reader_eof, _io_ctx));
    if (size == 0) {
        return Status::InternalError<false>(
                "The first line is empty, can not parse column numbers");
    }
    if (!validate_utf8(_params, reinterpret_cast<const char*>(ptr), size)) {
        return Status::InternalError<false>("Only support csv data in utf8 codec");
    }
    ptr = _remove_bom(ptr, size);
    _split_line(Slice(ptr, size));
    *col_nums = _split_values.size();
    return Status::OK();
}

Status CsvReader::_parse_col_names(std::vector<std::string>* col_names) {
    const uint8_t* ptr = nullptr;
    size_t size = 0;
    // no use of _line_reader_eof
    RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size, &_line_reader_eof, _io_ctx));
    if (size == 0) {
        return Status::InternalError<false>("The first line is empty, can not parse column names");
    }
    if (!validate_utf8(_params, reinterpret_cast<const char*>(ptr), size)) {
        return Status::InternalError<false>("Only support csv data in utf8 codec");
    }
    ptr = _remove_bom(ptr, size);
    _split_line(Slice(ptr, size));
    for (auto _split_value : _split_values) {
        col_names->emplace_back(_split_value.to_string());
    }
    return Status::OK();
}

// TODO(ftw): parse type
Status CsvReader::_parse_col_types(size_t col_nums, std::vector<DataTypePtr>* col_types) {
    // delete after.
    for (size_t i = 0; i < col_nums; ++i) {
        col_types->emplace_back(make_nullable(std::make_shared<DataTypeString>()));
    }

    // 1. check _line_reader_eof
    // 2. read line
    // 3. check utf8
    // 4. check size
    // 5. check _split_values.size must equal to col_nums.
    // 6. fill col_types
    return Status::OK();
}

const uint8_t* CsvReader::_remove_bom(const uint8_t* ptr, size_t& size) {
    if (size >= 3 && ptr[0] == 0xEF && ptr[1] == 0xBB && ptr[2] == 0xBF) {
        LOG(INFO) << "remove bom";
        constexpr size_t bom_size = 3;
        size -= bom_size;
        // In enclose mode, column_sep_positions were computed on the original line
        // (including BOM). After shifting the pointer, we must adjust those positions
        // so they remain correct relative to the new start.
        if (_enclose_reader_ctx) {
            _enclose_reader_ctx->adjust_column_sep_positions(bom_size);
        }
        return ptr + bom_size;
    }
    return ptr;
}

Status CsvReader::close() {
    if (_line_reader) {
        _line_reader->close();
    }

    if (_file_reader) {
        RETURN_IF_ERROR(_file_reader->close());
    }

    return Status::OK();
}

} // namespace doris