// 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/table/hive_reader.h"

#include <algorithm>

#include <cctype>

#include <string_view>

#include "common/consts.h"

#include "format_v2/column_mapper.h"

#include "format_v2/file_reader.h"

#include "runtime/runtime_state.h"

namespace doris::format::hive {

namespace {

TFileFormatType::type format_type_from_context(const format::ProjectedColumnBuildContext& context) {

    DORIS_CHECK(context.scan_params != nullptr);

    if (context.range != nullptr && context.range->__isset.format_type) {

        return context.range->format_type;

    }

    return context.scan_params->format_type;

}

bool use_column_position_mapping(const format::ProjectedColumnBuildContext& context) {

    if (context.runtime_state == nullptr || context.scan_params == nullptr) {

        return false;

    }

    switch (format_type_from_context(context)) {

    case TFileFormatType::FORMAT_PARQUET:

        return !context.runtime_state->query_options().hive_parquet_use_column_names;

    case TFileFormatType::FORMAT_ORC:

        return !context.runtime_state->query_options().hive_orc_use_column_names;

    default:

        return false;

    }

}

bool is_file_column_position_slot(const TFileScanSlotInfo& slot_info,

                                  const std::string& column_name) {

    if (column_name.starts_with(BeConsts::GLOBAL_ROWID_COL) ||

        column_name == BeConsts::ICEBERG_ROWID_COL) {

        return false;

    }

    if (slot_info.__isset.is_file_slot) {

        return slot_info.is_file_slot;

    }

    return !slot_info.__isset.category || slot_info.category != TColumnCategory::PARTITION_KEY;

}

bool is_hive1_orc_column_name(std::string_view name) {

    if (name.size() <= 4 || name.substr(0, 4) != "_col") {

        return false;

    }

    return std::ranges::all_of(name.substr(4), [](unsigned char c) { return std::isdigit(c); });

}

bool is_hive1_orc_file_schema(const std::vector<format::ColumnDefinition>& file_schema) {

    bool has_data_column = false;

    for (const auto& field : file_schema) {

        if (field.column_type != format::ColumnType::DATA_COLUMN) {

            continue;

        }

        has_data_column = true;

        if (!is_hive1_orc_column_name(field.name)) {

            return false;

        }

    }

    return has_data_column;

}

bool is_file_column_for_hive1_position_mapping(const format::ColumnDefinition& column) {

    if (column.column_type != format::ColumnType::DATA_COLUMN || column.is_partition_key) {

        return false;

    }

    return !column.name.starts_with(BeConsts::GLOBAL_ROWID_COL) &&

           column.name != BeConsts::ICEBERG_ROWID_COL;

}

void add_name_mapping(std::vector<std::string>* name_mapping, const std::string& alias) {

    DORIS_CHECK(name_mapping != nullptr);

    if (std::ranges::find(*name_mapping, alias) == name_mapping->end()) {

        name_mapping->push_back(alias);

    }

}

} // namespace

Status HiveReader::prepare_split(const format::SplitReadOptions& options) {

    {

        // Keep derived validation visible without overlapping the base scopes on the same timers.

        SCOPED_TIMER(_profile.total_timer);

        SCOPED_TIMER(_profile.prepare_split_timer);

        if (options.current_split_format != _format) {

            return Status::InternalError(

                    "Hive scan expects all splits to use the same file format, "

                    "initialized_format={}, current_split_format={}",

                    static_cast<int>(_format), static_cast<int>(options.current_split_format));

        }

    }

    return format::TableReader::prepare_split(options);

}

format::TableColumnMappingMode HiveReader::mapping_mode() const {

    // Hive-specific behavior: choose the column matching mode based on file format and the

    // matching session variable.

    //   - hive_orc_use_column_names / hive_parquet_use_column_names == true

    //     => BY_NAME (modern Hive default, match by column name)

    //   - those options == false

    //     => BY_INDEX (mainly for Hive1 ORC `_col0` / `_col1`, match by top-level position;

    //                  Parquet exposes the same switch for consistency)

    // TableReader updates `_format` in prepare_split(), so this is evaluated per split.

    DORIS_CHECK(_runtime_state != nullptr);

    const auto& query_options = _runtime_state->query_options();

    bool use_column_names = true;

    if (_format == format::FileFormat::ORC) {

        use_column_names = query_options.hive_orc_use_column_names;

    } else if (_format == format::FileFormat::PARQUET) {

        use_column_names = query_options.hive_parquet_use_column_names;

    } else if (_format == format::FileFormat::CSV || _format == format::FileFormat::TEXT ||

               _format == format::FileFormat::JSON) {

        // Hive CSV/TEXT/JSON readers synthesize a file-local schema from FE-provided file slots

        // because these formats do not carry embedded column names or field ids. The scan params'

        // format-specific attributes still tell the physical reader how to read values, while the

        // table-level mapper can safely match the synthesized file schema by table column name.

        use_column_names = true;

    } else {

        DORIS_CHECK(false) << "HiveReader does not support this file reader format";

    }

    return use_column_names ? format::TableColumnMappingMode::BY_NAME

                            : format::TableColumnMappingMode::BY_INDEX;

}

Status HiveReader::annotate_projected_column(const TFileScanSlotInfo& slot_info,

                                             format::ProjectedColumnBuildContext* context,

                                             format::ColumnDefinition* column) const {

    RETURN_IF_ERROR(format::TableReader::annotate_projected_column(slot_info, context, column));

    DORIS_CHECK(context != nullptr);

    DORIS_CHECK(column != nullptr);

    if (!use_column_position_mapping(*context) ||

        !is_file_column_position_slot(slot_info, column->name)) {

        return Status::OK();

    }

    const auto* scan_params = context->scan_params;

    DORIS_CHECK(scan_params != nullptr);

    if (!scan_params->__isset.column_idxs ||

        context->next_file_column_idx >= scan_params->column_idxs.size()) {

        return Status::InvalidArgument(

                "Hive positional column mapping is missing file index for column '{}', "

                "required file slot ordinal={}, column_idxs_size={}",

                column->name, context->next_file_column_idx,

                scan_params->__isset.column_idxs ? scan_params->column_idxs.size() : 0);

    }

    const auto file_index = scan_params->column_idxs[context->next_file_column_idx];

    if (file_index < 0) {

        return Status::InvalidArgument(

                "Hive positional column mapping has negative file index {} for column '{}'",

                file_index, column->name);

    }

    column->identifier = Field::create_field<TYPE_INT>(file_index);

    ++context->next_file_column_idx;

    return Status::OK();

}

Status HiveReader::validate_projected_columns(

        const format::ProjectedColumnBuildContext& context) const {

    if (!use_column_position_mapping(context)) {

        return Status::OK();

    }

    DORIS_CHECK(context.scan_params != nullptr);

    if (context.scan_params->__isset.column_idxs &&

        context.next_file_column_idx != context.scan_params->column_idxs.size()) {

        return Status::InvalidArgument(

                "Hive positional column mapping has unused file indexes: consumed={}, "

                "column_idxs_size={}",

                context.next_file_column_idx, context.scan_params->column_idxs.size());

    }

    return Status::OK();

}

Status HiveReader::annotate_file_schema(std::vector<format::ColumnDefinition>* file_schema) {

    DORIS_CHECK(file_schema != nullptr);

    if (_format != format::FileFormat::ORC || _runtime_state == nullptr ||

        !_runtime_state->query_options().hive_orc_use_column_names ||

        !is_hive1_orc_file_schema(*file_schema)) {

        return Status::OK();

    }

    DORIS_CHECK(_scan_params != nullptr);

    if (!_scan_params->__isset.column_idxs) {

        return Status::InvalidArgument(

                "Hive ORC Hive1-style name mapping is missing positional column indexes");

    }

    size_t next_file_column_idx = 0;

    for (const auto& table_column : _projected_columns) {

        if (!is_file_column_for_hive1_position_mapping(table_column)) {

            continue;

        }

        if (next_file_column_idx >= _scan_params->column_idxs.size()) {

            return Status::InvalidArgument(

                    "Hive ORC Hive1-style name mapping is missing file index for column '{}', "

                    "required file slot ordinal={}, column_idxs_size={}",

                    table_column.name, next_file_column_idx, _scan_params->column_idxs.size());

        }

        const auto file_index = _scan_params->column_idxs[next_file_column_idx];

        if (file_index < 0) {

            return Status::InvalidArgument(

                    "Hive ORC Hive1-style name mapping has negative file index {} for column '{}'",

                    file_index, table_column.name);

        }

        ++next_file_column_idx;

        if (static_cast<size_t>(file_index) >= file_schema->size()) {

            continue;

        }

        auto& file_column = (*file_schema)[static_cast<size_t>(file_index)];

        add_name_mapping(&file_column.name_mapping, table_column.name);

        if (table_column.has_identifier_name()) {

            add_name_mapping(&file_column.name_mapping, table_column.get_identifier_name());

        }

        for (const auto& alias : table_column.name_mapping) {

            add_name_mapping(&file_column.name_mapping, alias);

        }

    }

    if (next_file_column_idx != _scan_params->column_idxs.size()) {

        return Status::InvalidArgument(

                "Hive ORC Hive1-style name mapping has unused file indexes: consumed={}, "

                "column_idxs_size={}",

                next_file_column_idx, _scan_params->column_idxs.size());

    }

    return Status::OK();

}

} // namespace doris::format::hive