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

#include <fmt/format.h>

#include <algorithm>
#include <charconv>
#include <map>
#include <string>
#include <string_view>
#include <utility>

#include "common/cast_set.h"
#include "common/consts.h"
#include "core/assert_cast.h"
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_struct.h"
#include "runtime/descriptors.h"
#include "util/string_util.h"

namespace doris {
namespace {

bool is_scanner_materialized_virtual_column(const std::string& column_name) {
    return column_name == BeConsts::ICEBERG_ROWID_COL;
}

bool parse_non_negative_int(std::string_view value, int32_t* result) {
    DORIS_CHECK(result != nullptr);
    int32_t parsed = -1;
    const auto* begin = value.data();
    const auto* end = begin + value.size();
    const auto [ptr, ec] = std::from_chars(begin, end, parsed);
    if (ec != std::errc() || ptr != end || parsed < 0) {
        return false;
    }
    *result = parsed;
    return true;
}

std::string access_path_to_string(const std::vector<std::string>& path) {
    return fmt::format("{}", fmt::join(path, "."));
}

format::ColumnDefinition* find_or_add_child(format::ColumnDefinition* parent, int32_t id,
                                            std::string name, DataTypePtr type) {
    DORIS_CHECK(parent != nullptr);
    for (auto& child : parent->children) {
        if ((child.has_identifier_field_id() && child.get_identifier_field_id() == id) ||
            child.name == name) {
            return &child;
        }
    }
    parent->children.push_back({
            .identifier = Field::create_field<TYPE_INT>(id),
            .name = std::move(name),
            .type = std::move(type),
            .children = {},
            .default_expr = nullptr,
            .is_partition_key = false,
    });
    return &parent->children.back();
}

void inherit_schema_metadata(format::ColumnDefinition* column,
                             const format::ColumnDefinition* schema_column) {
    if (column == nullptr || schema_column == nullptr) {
        return;
    }
    column->name_mapping = schema_column->name_mapping;
    // The presence bit is part of the mapping contract: an explicit empty mapping must remain
    // authoritative after access-path pruning instead of enabling current-name fallback.
    column->has_name_mapping = schema_column->has_name_mapping;
    // Initial defaults describe the logical value of fields absent from older files. Nested
    // access-path pruning must retain them just like it retains rename metadata.
    column->initial_default_value = schema_column->initial_default_value;
    column->initial_default_value_is_base64 = schema_column->initial_default_value_is_base64;
}

const format::ColumnDefinition* find_schema_child_by_path(
        const format::ColumnDefinition* schema_column, const std::string& child_path,
        bool prefer_exact_name_match) {
    if (schema_column == nullptr) {
        return nullptr;
    }
    int32_t parsed_field_id = -1;
    if (parse_non_negative_int(child_path, &parsed_field_id)) {
        const auto child_it = std::ranges::find_if(
                schema_column->children, [&](const format::ColumnDefinition& child) {
                    return child.has_identifier_field_id() &&
                           child.get_identifier_field_id() == parsed_field_id;
                });
        return child_it == schema_column->children.end() ? nullptr : &*child_it;
    }
    if (!prefer_exact_name_match) {
        const auto child_it = std::ranges::find_if(schema_column->children, [&](const auto& child) {
            if (to_lower(child.name) == to_lower(child_path)) {
                return true;
            }
            return std::ranges::any_of(child.name_mapping, [&](const std::string& alias) {
                return to_lower(alias) == to_lower(child_path);
            });
        });
        return child_it == schema_column->children.end() ? nullptr : &*child_it;
    }
    // Iceberg can reuse a historical name for a newly added sibling. Current names therefore
    // have precedence across the entire struct; an earlier alias must not steal that access path.
    const auto exact_it = std::ranges::find_if(schema_column->children, [&](const auto& child) {
        return to_lower(child.name) == to_lower(child_path);
    });
    if (exact_it != schema_column->children.end()) {
        return &*exact_it;
    }
    const auto alias_it = std::ranges::find_if(schema_column->children, [&](const auto& child) {
        return std::ranges::any_of(child.name_mapping, [&](const std::string& alias) {
            return to_lower(alias) == to_lower(child_path);
        });
    });
    return alias_it == schema_column->children.end() ? nullptr : &*alias_it;
}

int32_t schema_field_id(const format::ColumnDefinition* schema_column) {
    if (schema_column == nullptr || !schema_column->has_identifier_field_id()) {
        return -1;
    }
    return schema_column->get_identifier_field_id();
}

int32_t schema_field_id_or(const format::ColumnDefinition* schema_column, int32_t fallback) {
    const auto field_id = schema_field_id(schema_column);
    return field_id >= 0 ? field_id : fallback;
}

std::string schema_field_name_or(const format::ColumnDefinition* schema_column,
                                 std::string fallback) {
    return schema_column == nullptr || schema_column->name.empty() ? std::move(fallback)
                                                                   : schema_column->name;
}

struct AccessPathNode {
    bool project_all = false;
    std::map<std::string, AccessPathNode> children;
};

void merge_access_path_node(AccessPathNode* dst, const AccessPathNode& src) {
    DORIS_CHECK(dst != nullptr);
    if (dst->project_all) {
        return;
    }
    if (src.project_all) {
        dst->project_all = true;
        dst->children.clear();
        return;
    }
    for (const auto& [path, child] : src.children) {
        merge_access_path_node(&dst->children[path], child);
    }
}

void insert_access_path(AccessPathNode* root, const std::vector<std::string>& path,
                        size_t path_idx) {
    DORIS_CHECK(root != nullptr);
    if (root->project_all) {
        return;
    }
    if (path_idx >= path.size()) {
        root->project_all = true;
        root->children.clear();
        return;
    }
    insert_access_path(&root->children[path[path_idx]], path, path_idx + 1);
}

Status build_nested_children_from_access_node(format::ColumnDefinition* column,
                                              const DataTypePtr& type, const AccessPathNode& node,
                                              const std::string& path,
                                              const format::ColumnDefinition* schema_column,
                                              bool prefer_exact_name_match);

// Expand a full complex-column projection into table-schema children when the table format provides
// an external/current schema. Without this, `SELECT complex_col` or `SELECT *` leaves
// ColumnDefinition::children empty, so ColumnMapper treats the root complex column as a scalar
// mapping and later tries to cast the old file shape to the current table shape directly.
//
// Examples:
//   - STRUCT country/city projected from an old file STRUCT country/population/location should
//     create children country and city, so city can be materialized as missing/default.
//   - ARRAY<STRUCT<item, quantity>> should create the array element wrapper and then the element
//     struct children item and quantity.
//   - MAP<STRING, STRUCT<full_name, age>> should create semantic children key/value directly, then
//     expand the value struct children full_name and age. Do not introduce a physical entries
//     wrapper here: ColumnMapper and TableReader treat MAP children as [key, value].
Status build_all_nested_children_from_schema(format::ColumnDefinition* column,
                                             const DataTypePtr& type, const std::string& path,
                                             const format::ColumnDefinition* schema_column,
                                             bool prefer_exact_name_match) {
    DORIS_CHECK(column != nullptr);

    const auto nested_type = remove_nullable(type);
    AccessPathNode project_all;
    project_all.project_all = true;
    switch (nested_type->get_primitive_type()) {
    case TYPE_STRUCT: {
        const auto& struct_type = assert_cast<const DataTypeStruct&>(*nested_type);
        for (size_t field_idx = 0; field_idx < struct_type.get_elements().size(); ++field_idx) {
            const auto field_name = struct_type.get_element_name(field_idx);
            const auto* schema_child =
                    find_schema_child_by_path(schema_column, field_name, prefer_exact_name_match);
            auto* child = find_or_add_child(
                    column, schema_field_id_or(schema_child, cast_set<int32_t>(field_idx)),
                    schema_field_name_or(schema_child, field_name),
                    struct_type.get_element(field_idx));
            inherit_schema_metadata(child, schema_child);
            RETURN_IF_ERROR(build_nested_children_from_access_node(
                    child, child->type, project_all, path + "." + child->name, schema_child,
                    prefer_exact_name_match));
        }
        return Status::OK();
    }
    case TYPE_ARRAY: {
        const auto& array_type = assert_cast<const DataTypeArray&>(*nested_type);
        const auto* element_schema = schema_column != nullptr && !schema_column->children.empty()
                                             ? &schema_column->children[0]
                                             : nullptr;
        auto* child = find_or_add_child(column, schema_field_id_or(element_schema, 0), "element",
                                        array_type.get_nested_type());
        inherit_schema_metadata(child, element_schema);
        return build_nested_children_from_access_node(child, child->type, project_all, path + ".*",
                                                      element_schema, prefer_exact_name_match);
    }
    case TYPE_MAP: {
        const auto& map_type = assert_cast<const DataTypeMap&>(*nested_type);
        const auto* key_schema = schema_column != nullptr && !schema_column->children.empty()
                                         ? &schema_column->children[0]
                                         : nullptr;
        const auto* value_schema = schema_column != nullptr && schema_column->children.size() > 1
                                           ? &schema_column->children[1]
                                           : nullptr;
        auto* key_child = find_or_add_child(column, schema_field_id_or(key_schema, 0), "key",
                                            map_type.get_key_type());
        inherit_schema_metadata(key_child, key_schema);
        RETURN_IF_ERROR(build_nested_children_from_access_node(
                key_child, key_child->type, project_all, path + ".KEYS", key_schema,
                prefer_exact_name_match));
        auto* value_child = find_or_add_child(column, schema_field_id_or(value_schema, 1), "value",
                                              map_type.get_value_type());
        inherit_schema_metadata(value_child, value_schema);
        RETURN_IF_ERROR(build_nested_children_from_access_node(
                value_child, value_child->type, project_all, path + ".VALUES", value_schema,
                prefer_exact_name_match));
        return Status::OK();
    }
    default:
        return Status::OK();
    }
}

Status build_struct_children_from_access_node(format::ColumnDefinition* column,
                                              const DataTypeStruct& struct_type,
                                              const AccessPathNode& node, const std::string& path,
                                              const format::ColumnDefinition* schema_column,
                                              bool prefer_exact_name_match) {
    DORIS_CHECK(column != nullptr);
    for (const auto& [child_path, child_node] : node.children) {
        // Struct children are resolved by name or schema field id. We do not treat a numeric
        // child token as a struct ordinal, because `col.0` becomes ambiguous once the struct
        // evolves. Position-based access needs a separate design if it is required later.
        if (child_path == "OFFSET" || child_path == "*" || child_path == "KEYS" ||
            child_path == "VALUES") {
            return Status::NotSupported(
                    "AccessPathParser does not support access path {} for slot {}",
                    path + "." + child_path, column->name);
        }

        // Prefer the table/schema ColumnDefinition because it carries field ids and aliases.
        // Fallback to the struct type name only for formats without external schema metadata.
        const auto* schema_child =
                find_schema_child_by_path(schema_column, child_path, prefer_exact_name_match);
        int32_t field_id = schema_field_id(schema_child);
        std::string field_name = schema_child == nullptr ? child_path : schema_child->name;
        DataTypePtr field_type = schema_child == nullptr ? nullptr : schema_child->type;
        if (field_id < 0 || field_type == nullptr) {
            for (size_t field_idx = 0; field_idx < struct_type.get_elements().size(); ++field_idx) {
                if (to_lower(struct_type.get_element_name(field_idx)) == to_lower(field_name)) {
                    field_id = cast_set<int32_t>(field_idx);
                    field_name = struct_type.get_element_name(field_idx);
                    field_type = struct_type.get_element(field_idx);
                    break;
                }
            }
        }

        if (field_id < 0 || field_type == nullptr) {
            return Status::NotSupported(
                    "AccessPathParser does not support access path {} for slot {}",
                    path + "." + child_path, column->name);
        }
        // TODO: For TVF Parquet files without field ids, this fallback uses the struct ordinal as
        // the table child identifier. BY_NAME mapping should instead keep a string identifier and
        // let TableColumnMapper resolve the file-local child id from the Parquet schema.
        auto* child = find_or_add_child(column, field_id, field_name, field_type);
        inherit_schema_metadata(child, schema_child);
        RETURN_IF_ERROR(build_nested_children_from_access_node(
                child, child->type, child_node, path + "." + child_path, schema_child,
                prefer_exact_name_match));
    }
    return Status::OK();
}

Status build_map_children_from_access_node(format::ColumnDefinition* column,
                                           const DataTypeMap& map_type, const AccessPathNode& node,
                                           const std::string& path,
                                           const format::ColumnDefinition* schema_column,
                                           bool prefer_exact_name_match) {
    DORIS_CHECK(column != nullptr);
    AccessPathNode key_node;
    AccessPathNode value_node;
    bool need_key = false;
    bool need_value = false;

    for (const auto& [child_path, child_node] : node.children) {
        if (child_path == "OFFSET") {
            return Status::NotSupported(
                    "AccessPathParser does not support access path {} for slot {}",
                    path + "." + child_path, column->name);
        }
        if (child_path == "KEYS") {
            need_key = true;
            merge_access_path_node(&key_node, child_node);
            continue;
        }
        if (child_path == "VALUES") {
            need_key = true;
            key_node.project_all = true;
            key_node.children.clear();
            need_value = true;
            merge_access_path_node(&value_node, child_node);
            continue;
        }
        if (child_path == "*") {
            need_key = true;
            key_node.project_all = true;
            key_node.children.clear();
            need_value = true;
            merge_access_path_node(&value_node, child_node);
            continue;
        }
        return Status::NotSupported("AccessPathParser does not support access path {} for slot {}",
                                    path + "." + child_path, column->name);
    }
    if (need_key && !need_value) {
        // A key-only MAP projection is not independently materializable yet. FileScannerV2 can
        // describe a projection such as `m.KEYS`, but the downstream file block -> table block path
        // still builds a ColumnMap from key column + value column + offsets. If the value child is
        // omitted here, TableReader/ColumnMapper cannot reconstruct a valid table MAP column even
        // though the query only needs keys.
        //
        // Example:
        //   SELECT map_keys(m) FROM t;
        // or
        //   SELECT * FROM t WHERE array_contains(map_keys(m), 'k1');
        //
        // The access path only asks for `m.KEYS`, but the scan still has to read `m.VALUES` as a
        // temporary full projection until map materialization supports constructing a table MAP
        // from keys only.
        need_value = true;
        value_node.project_all = true;
        value_node.children.clear();
    }

    if (!need_key && !need_value) {
        return Status::OK();
    }

    const auto* key_schema = schema_column != nullptr && !schema_column->children.empty()
                                     ? &schema_column->children[0]
                                     : nullptr;
    const auto* value_schema = schema_column != nullptr && schema_column->children.size() > 1
                                       ? &schema_column->children[1]
                                       : nullptr;
    if (need_key) {
        auto* key_child = find_or_add_child(column, schema_field_id_or(key_schema, 0), "key",
                                            map_type.get_key_type());
        inherit_schema_metadata(key_child, key_schema);
        RETURN_IF_ERROR(build_nested_children_from_access_node(key_child, key_child->type, key_node,
                                                               path + ".KEYS", key_schema,
                                                               prefer_exact_name_match));
    }
    if (need_value) {
        auto* value_child = find_or_add_child(column, schema_field_id_or(value_schema, 1), "value",
                                              map_type.get_value_type());
        inherit_schema_metadata(value_child, value_schema);
        RETURN_IF_ERROR(build_nested_children_from_access_node(
                value_child, value_child->type, value_node, path + ".VALUES", value_schema,
                prefer_exact_name_match));
    }
    return Status::OK();
}

Status build_nested_children_from_access_node(format::ColumnDefinition* column,
                                              const DataTypePtr& type, const AccessPathNode& node,
                                              const std::string& path,
                                              const format::ColumnDefinition* schema_column,
                                              bool prefer_exact_name_match) {
    DORIS_CHECK(column != nullptr);
    if (node.project_all || node.children.empty()) {
        return build_all_nested_children_from_schema(column, type, path, schema_column,
                                                     prefer_exact_name_match);
    }

    const auto nested_type = remove_nullable(type);
    switch (nested_type->get_primitive_type()) {
    case TYPE_STRUCT:
        return build_struct_children_from_access_node(
                column, assert_cast<const DataTypeStruct&>(*nested_type), node, path, schema_column,
                prefer_exact_name_match);
    case TYPE_ARRAY: {
        if (node.children.size() != 1 || !node.children.contains("*")) {
            return Status::NotSupported(
                    "AccessPathParser does not support access path {} for slot {}", path,
                    column->name);
        }
        const auto& array_type = assert_cast<const DataTypeArray&>(*nested_type);
        const auto* element_schema = schema_column != nullptr && !schema_column->children.empty()
                                             ? &schema_column->children[0]
                                             : nullptr;
        auto* child = find_or_add_child(column, schema_field_id_or(element_schema, 0), "element",
                                        array_type.get_nested_type());
        inherit_schema_metadata(child, element_schema);
        return build_nested_children_from_access_node(child, child->type, node.children.at("*"),
                                                      path + ".*", element_schema,
                                                      prefer_exact_name_match);
    }
    case TYPE_MAP:
        return build_map_children_from_access_node(
                column, assert_cast<const DataTypeMap&>(*nested_type), node, path, schema_column,
                prefer_exact_name_match);
    default:
        return Status::NotSupported("AccessPathParser does not support access path {} for slot {}",
                                    path, column->name);
    }
}

} // namespace

Status AccessPathParser::build_nested_children(format::ColumnDefinition* column,
                                               const std::vector<TColumnAccessPath>& access_paths,
                                               const format::ColumnDefinition* schema_column,
                                               bool prefer_exact_name_match) {
    DORIS_CHECK(column != nullptr);
    if (is_scanner_materialized_virtual_column(column->name)) {
        return Status::OK();
    }
    if (!is_complex_type(remove_nullable(column->type)->get_primitive_type())) {
        return Status::OK();
    }

    AccessPathNode root;
    // Build tree for AccessPathNode.
    // For example, for access paths ["a.b", "a.c", "d"], the tree will be:
    // root
    // ├── a
    // │   ├── b
    // │   └── c
    // └── d
    for (const auto& access_path : access_paths) {
        // TODO: Support META access paths if needed. Currently AccessPathParser only supports
        // DATA access paths.
        if (access_path.type != TAccessPathType::DATA || !access_path.__isset.data_access_path) {
            return Status::NotSupported(
                    "AccessPathParser only supports DATA access paths for slot {}", column->name);
        }
        const auto& path = access_path.data_access_path.path;
        if (path.empty()) {
            insert_access_path(&root, path, 0);
            continue;
        }
        int32_t top_level_id = -1;
        if (to_lower(path.front()) != to_lower(column->name) &&
            (!parse_non_negative_int(path.front(), &top_level_id) ||
             !column->has_identifier_field_id() ||
             top_level_id != column->get_identifier_field_id())) {
            return Status::NotSupported("AccessPathParser access path {} does not match slot {}",
                                        access_path_to_string(path), column->name);
        }
        insert_access_path(&root, path, 1);
    }
    // Recursively build nested children for the column based on the AccessPathNode tree.
    return build_nested_children_from_access_node(column, column->type, root, column->name,
                                                  schema_column, prefer_exact_name_match);
}

Status AccessPathParser::build_nested_children(format::ColumnDefinition* column,
                                               const SlotDescriptor* slot_desc,
                                               const format::ColumnDefinition* schema_column,
                                               bool prefer_exact_name_match) {
    DORIS_CHECK(column != nullptr);
    DORIS_CHECK(slot_desc != nullptr);
    return build_nested_children(column, slot_desc->all_access_paths(), schema_column,
                                 prefer_exact_name_match);
}

} // namespace doris