// 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_v2/file_reader.h"

#include <sstream>

#include "format_v2/column_mapper.h"

#include "io/fs/buffered_reader.h"

#include "io/fs/tracing_file_reader.h"

#include "runtime/runtime_state.h"

namespace doris::format {

namespace {

std::unique_ptr<FileStructPredicateTarget> clone_struct_predicate_target(

        const std::unique_ptr<FileStructPredicateTarget>& target) {

    return target == nullptr ? nullptr : std::make_unique<FileStructPredicateTarget>(*target);

}

template <typename T, typename Formatter>

std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {

    std::ostringstream out;

    out << "[";

    for (size_t i = 0; i < values.size(); ++i) {

        if (i > 0) {

            out << ", ";

        }

        out << formatter(values[i]);

    }

    out << "]";

    return out.str();

}

Unexecuted instantiation: file_reader.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_16LocalColumnIndexEZNKS0_15FileScanRequest12debug_stringB5cxx11EvE3$_0EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISD_EET0_

Unexecuted instantiation: file_reader.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_16LocalColumnIndexEZNKS0_15FileScanRequest12debug_stringB5cxx11EvE3$_1EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISD_EET0_

Unexecuted instantiation: file_reader.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_25FileColumnPredicateFilterEZNKS0_15FileScanRequest12debug_stringB5cxx11EvE3$_2EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISD_EET0_

std::string int_vector_debug_string(const std::vector<int32_t>& values) {

    std::ostringstream out;

    out << "[";

    for (size_t i = 0; i < values.size(); ++i) {

        if (i > 0) {

            out << ", ";

        }

        out << values[i];

    }

    out << "]";

    return out.str();

}

void append_struct_predicate_path(const FileStructPredicateTarget* target,

                                  std::vector<int32_t>* path) {

    DORIS_CHECK(path != nullptr);

    for (const auto* current = target; current != nullptr; current = current->child.get()) {

        path->push_back(current->file_local_id);

    }

}

std::string struct_predicate_target_debug_string(const FileStructPredicateTarget* target) {

    if (target == nullptr) {

        return "null";

    }

    std::ostringstream out;

    out << "{file_local_id=" << target->file_local_id

        << ", file_child_name=" << target->file_child_name

        << ", child=" << struct_predicate_target_debug_string(target->child.get()) << "}";

    return out.str();

}

bool struct_predicate_targets_equal(const FileStructPredicateTarget* lhs,

                                    const FileStructPredicateTarget* rhs) {

    while (lhs != nullptr && rhs != nullptr) {

        if (lhs->file_local_id != rhs->file_local_id) {

            return false;

        }

        lhs = lhs->child.get();

        rhs = rhs->child.get();

    }

    return lhs == nullptr && rhs == nullptr;

}

} // namespace

FileStructPredicateTarget::FileStructPredicateTarget(const FileStructPredicateTarget& other)

        : file_local_id(other.file_local_id),

          file_child_name(other.file_child_name),

          child(clone_struct_predicate_target(other.child)) {}

FileStructPredicateTarget& FileStructPredicateTarget::operator=(

        const FileStructPredicateTarget& other) {

    if (this == &other) {

        return *this;

    }

    file_local_id = other.file_local_id;

    file_child_name = other.file_child_name;

    child = clone_struct_predicate_target(other.child);

    return *this;

}

FileNestedPredicateTarget::FileNestedPredicateTarget(const FileNestedPredicateTarget& other)

        : file_column_id(other.file_column_id),

          struct_target(clone_struct_predicate_target(other.struct_target)) {}

FileNestedPredicateTarget& FileNestedPredicateTarget::operator=(

        const FileNestedPredicateTarget& other) {

    if (this == &other) {

        return *this;

    }

    file_column_id = other.file_column_id;

    struct_target = clone_struct_predicate_target(other.struct_target);

    return *this;

}

LocalColumnId FileColumnPredicateFilter::effective_file_column_id() const {

    return target.is_valid() ? target.file_column_id : file_column_id;

}

std::vector<int32_t> FileColumnPredicateFilter::effective_file_child_id_path() const {

    if (!target.is_valid()) {

        return file_child_id_path;

    }

    std::vector<int32_t> path;

    append_struct_predicate_path(target.struct_target.get(), &path);

    return path;

}

bool FileColumnPredicateFilter::same_target_as(const FileColumnPredicateFilter& other) const {

    if (target.is_valid() && other.target.is_valid()) {

        return target.file_column_id == other.target.file_column_id &&

               struct_predicate_targets_equal(target.struct_target.get(),

                                              other.target.struct_target.get());

    }

    return effective_file_column_id() == other.effective_file_column_id() &&

           effective_file_child_id_path() == other.effective_file_child_id_path();

}

std::string FileColumnPredicateFilter::debug_string() const {

    std::ostringstream out;

    out << "FileColumnPredicateFilter{target={file_column_id=" << effective_file_column_id()

        << ", struct_target=" << struct_predicate_target_debug_string(target.struct_target.get())

        << "}, file_child_id_path=" << int_vector_debug_string(effective_file_child_id_path())

        << ", predicate_count=" << predicates.size() << "}";

    return out.str();

}

std::string FileScanRequest::debug_string() const {

    std::ostringstream out;

    out << "FileScanRequest{predicate_columns="

        << join_debug_strings(

                   predicate_columns,

                   [](const LocalColumnIndex& projection) { return projection.debug_string(); })

        << ", non_predicate_columns="

        << join_debug_strings(

                   non_predicate_columns,

                   [](const LocalColumnIndex& projection) { return projection.debug_string(); })

        << ", local_positions={";

    size_t position_idx = 0;

    for (const auto& [column_id, block_position] : local_positions) {

        if (position_idx++ > 0) {

            out << ", ";

        }

        out << column_id << ":" << block_position;

    }

    out << "}, conjunct_count=" << conjuncts.size()

        << ", delete_conjunct_count=" << delete_conjuncts.size() << ", column_predicate_filters="

        << join_debug_strings(

                   column_predicate_filters,

                   [](const FileColumnPredicateFilter& filter) { return filter.debug_string(); })

        << "}";

    return out.str();

}

Status FileReader::init(RuntimeState* state) {

    _init_profile();

    SCOPED_RAW_TIMER(&_reader_statistics.file_reader_create_time);

    ++_reader_statistics.open_file_num;

    io::FileReaderOptions reader_options =

            FileFactory::get_reader_options(state->query_options(), *_file_description);

    _file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(

            _profile, *_system_properties, *_file_description, reader_options,

            io::DelegateReader::AccessMode::RANDOM, _io_ctx));

    // IOContext can be present without file_reader_stats in standalone tests or callers that only

    // need extra IO state. TracingFileReader dereferences the stats pointer on every read, so only

    // wrap the physical reader when stats collection is actually available.

    _tracing_file_reader = _io_ctx && _io_ctx->file_reader_stats

                                   ? std::make_shared<io::TracingFileReader>(

                                             _file_reader, _io_ctx->file_reader_stats)

                                   : _file_reader;

    _eof = false;

    return Status::OK();

}

std::unique_ptr<TableColumnMapper> FileReader::create_column_mapper(

        TableColumnMapperOptions options) const {

    return std::make_unique<TableColumnMapper>(std::move(options));

}

} // namespace doris::format