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

#include <ctype.h>

#include <algorithm>

#include <ostream>

#include <utility>

#include "common/cast_set.h"

#include "common/logging.h"

#include "core/data_type/data_type_array.h"

#include "core/data_type/data_type_factory.hpp"

#include "core/data_type/data_type_map.h"

#include "core/data_type/data_type_spatial.h"

#include "core/data_type/data_type_struct.h"

#include "core/data_type/define_primitive_type.h"

#include "format/generic_reader.h"

#include "format/table/table_schema_change_helper.h"

#include "util/slice.h"

#include "util/string_util.h"

namespace doris {

static bool is_group_node(const tparquet::SchemaElement& schema) {

    return schema.num_children > 0;

}

static bool is_list_node(const tparquet::SchemaElement& schema) {

    return schema.__isset.converted_type && schema.converted_type == tparquet::ConvertedType::LIST;

}

static bool is_map_node(const tparquet::SchemaElement& schema) {

    return schema.__isset.converted_type &&

           (schema.converted_type == tparquet::ConvertedType::MAP ||

            schema.converted_type == tparquet::ConvertedType::MAP_KEY_VALUE);

}

static bool is_repeated_node(const tparquet::SchemaElement& schema) {

    return schema.__isset.repetition_type &&

           schema.repetition_type == tparquet::FieldRepetitionType::REPEATED;

}

static bool is_required_node(const tparquet::SchemaElement& schema) {

    return schema.__isset.repetition_type &&

           schema.repetition_type == tparquet::FieldRepetitionType::REQUIRED;

}

static bool is_optional_node(const tparquet::SchemaElement& schema) {

    return schema.__isset.repetition_type &&

           schema.repetition_type == tparquet::FieldRepetitionType::OPTIONAL;

}

static int num_children_node(const tparquet::SchemaElement& schema) {

    return schema.__isset.num_children ? schema.num_children : 0;

}

/**

 * `repeated_parent_def_level` is the definition level of the first ancestor node whose repetition_type equals REPEATED.

 * Empty array/map values are not stored in doris columns, so have to use `repeated_parent_def_level` to skip the

 * empty or null values in ancestor node.

 *

 * For instance, considering an array of strings with 3 rows like the following:

 * null, [], [a, b, c]

 * We can store four elements in data column: null, a, b, c

 * and the offsets column is: 1, 1, 4

 * and the null map is: 1, 0, 0

 * For the i-th row in array column: range from `offsets[i - 1]` until `offsets[i]` represents the elements in this row,

 * so we can't store empty array/map values in doris data column.

 * As a comparison, spark does not require `repeated_parent_def_level`,

 * because the spark column stores empty array/map values , and use anther length column to indicate empty values.

 * Please reference: https://github.com/apache/spark/blob/master/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnVector.java

 *

 * Furthermore, we can also avoid store null array/map values in doris data column.

 * The same three rows as above, We can only store three elements in data column: a, b, c

 * and the offsets column is: 0, 0, 3

 * and the null map is: 1, 0, 0

 *

 * Inherit the repetition and definition level from parent node, if the parent node is repeated,

 * we should set repeated_parent_def_level = definition_level, otherwise as repeated_parent_def_level.

 * @param parent parent node

 * @param repeated_parent_def_level the first ancestor node whose repetition_type equals REPEATED

 */

static void set_child_node_level(FieldSchema* parent, int16_t repeated_parent_def_level) {

    for (auto& child : parent->children) {

        child.repetition_level = parent->repetition_level;

        child.definition_level = parent->definition_level;

        child.repeated_parent_def_level = repeated_parent_def_level;

    }

}

static bool is_struct_list_node(const tparquet::SchemaElement& schema) {

    const std::string& name = schema.name;

    static const Slice array_slice("array", 5);

    static const Slice tuple_slice("_tuple", 6);

    Slice slice(name);

    return slice == array_slice || slice.ends_with(tuple_slice);

}

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

    std::stringstream ss;

    ss << "FieldSchema(name=" << name << ", R=" << repetition_level << ", D=" << definition_level;

    if (children.size() > 0) {

        ss << ", type=" << data_type->get_name() << ", children=[";

        for (int i = 0; i < children.size(); ++i) {

            if (i != 0) {

                ss << ", ";

            }

            ss << children[i].debug_string();

        }

        ss << "]";

    } else {

        ss << ", physical_type=" << physical_type;

        ss << " , doris_type=" << data_type->get_name();

    }

    ss << ")";

    return ss.str();

}

Status FieldDescriptor::parse_from_thrift(const std::vector<tparquet::SchemaElement>& t_schemas) {

    if (t_schemas.size() == 0 || !is_group_node(t_schemas[0])) {

        return Status::InvalidArgument("Wrong parquet root schema element");

    }

    const auto& root_schema = t_schemas[0];

    _fields.resize(root_schema.num_children);

    _next_schema_pos = 1;

    for (int i = 0; i < root_schema.num_children; ++i) {

        RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, &_fields[i]));

        if (_name_to_field.find(_fields[i].name) != _name_to_field.end()) {

            return Status::InvalidArgument("Duplicated field name: {}", _fields[i].name);

        }

        _name_to_field.emplace(_fields[i].name, &_fields[i]);

    }

    if (_next_schema_pos != t_schemas.size()) {

        return Status::InvalidArgument("Remaining {} unparsed schema elements",

                                       t_schemas.size() - _next_schema_pos);

    }

    return Status::OK();

}

Status FieldDescriptor::parse_node_field(const std::vector<tparquet::SchemaElement>& t_schemas,

                                         size_t curr_pos, FieldSchema* node_field) {

    if (curr_pos >= t_schemas.size()) {

        return Status::InvalidArgument("Out-of-bounds index of schema elements");

    }

    auto& t_schema = t_schemas[curr_pos];

    if (is_group_node(t_schema)) {

        // nested structure or nullable list

        return parse_group_field(t_schemas, curr_pos, node_field);

    }

    if (is_repeated_node(t_schema)) {

        // repeated <primitive-type> <name> (LIST)

        // produce required list<element>

        node_field->repetition_level++;

        node_field->definition_level++;

        node_field->children.resize(1);

        set_child_node_level(node_field, node_field->definition_level);

        auto child = &node_field->children[0];

        parse_physical_field(t_schema, false, child);

        node_field->name = t_schema.name;

        node_field->lower_case_name = to_lower(t_schema.name);

        node_field->data_type = std::make_shared<DataTypeArray>(make_nullable(child->data_type));

        _next_schema_pos = curr_pos + 1;

        node_field->field_id = t_schema.__isset.field_id ? t_schema.field_id : -1;

    } else {

        bool is_optional = is_optional_node(t_schema);

        if (is_optional) {

            node_field->definition_level++;

        }

        parse_physical_field(t_schema, is_optional, node_field);

        _next_schema_pos = curr_pos + 1;

    }

    return Status::OK();

}

void FieldDescriptor::parse_physical_field(const tparquet::SchemaElement& physical_schema,

                                           bool is_nullable, FieldSchema* physical_field) {

    physical_field->name = physical_schema.name;

    physical_field->lower_case_name = to_lower(physical_field->name);

    physical_field->parquet_schema = physical_schema;

    physical_field->physical_type = physical_schema.type;

    physical_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id

    _physical_fields.push_back(physical_field);

    physical_field->physical_column_index = cast_set<int>(_physical_fields.size() - 1);

    auto type = get_doris_type(physical_schema, is_nullable);

    physical_field->data_type = type.first;

    physical_field->is_type_compatibility = type.second;

    physical_field->field_id = physical_schema.__isset.field_id ? physical_schema.field_id : -1;

}

std::pair<DataTypePtr, bool> FieldDescriptor::get_doris_type(

        const tparquet::SchemaElement& physical_schema, bool nullable) {

    std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(), false};

    try {

        if (physical_schema.__isset.logicalType) {

            ans = convert_to_doris_type(physical_schema.logicalType, nullable);

        } else if (physical_schema.__isset.converted_type) {

            ans = convert_to_doris_type(physical_schema, nullable);

        }

    } catch (...) {

        // now the Not supported exception are ignored

        // so those byte_array maybe be treated as varbinary(now) : string(before)

    }

    if (ans.first->get_primitive_type() == PrimitiveType::INVALID_TYPE) {

        switch (physical_schema.type) {

        case tparquet::Type::BOOLEAN:

            ans.first = DataTypeFactory::instance().create_data_type(TYPE_BOOLEAN, nullable);

            break;

        case tparquet::Type::INT32:

            ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);

            break;

        case tparquet::Type::INT64:

            ans.first = DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);

            break;

        case tparquet::Type::INT96:

            // INT96 has no timezone annotation. A table's logical schema may override this

            // for known instant columns, but a catalog option cannot supply missing semantics.

            ans.first =

                    DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable, 0, 6);

            break;

        case tparquet::Type::FLOAT:

            ans.first = DataTypeFactory::instance().create_data_type(TYPE_FLOAT, nullable);

            break;

        case tparquet::Type::DOUBLE:

            ans.first = DataTypeFactory::instance().create_data_type(TYPE_DOUBLE, nullable);

            break;

        case tparquet::Type::BYTE_ARRAY:

            if (_enable_mapping_varbinary) {

                // if physical_schema not set logicalType and converted_type,

                // we treat BYTE_ARRAY as VARBINARY by default, so that we can read all data directly.

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_VARBINARY, nullable);

            } else {

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);

            }

            break;

        case tparquet::Type::FIXED_LEN_BYTE_ARRAY:

            ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);

            break;

        default:

            throw Exception(Status::InternalError("Not supported parquet logicalType{}",

                                                  physical_schema.type));

            break;

        }

    }

    return ans;

}

std::pair<DataTypePtr, bool> FieldDescriptor::convert_to_doris_type(

        tparquet::LogicalType logicalType, bool nullable) {

    std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(), false};

    bool& is_type_compatibility = ans.second;

    if (logicalType.__isset.GEOMETRY) {

        const std::string crs =

                logicalType.GEOMETRY.__isset.crs ? logicalType.GEOMETRY.crs : "OGC:CRS84";

        // OPTIONAL spatial fields must retain their definition-level threshold for COUNT.

        auto spatial_type = std::make_shared<DataTypeSpatial>(TYPE_GEOMETRY, crs);

        ans.first = nullable ? make_nullable(spatial_type) : spatial_type;

    } else if (logicalType.__isset.GEOGRAPHY) {

        const std::string crs =

                logicalType.GEOGRAPHY.__isset.crs ? logicalType.GEOGRAPHY.crs : "OGC:CRS84";

        const std::string algorithm = logicalType.GEOGRAPHY.__isset.algorithm

                                              ? tparquet::to_string(logicalType.GEOGRAPHY.algorithm)

                                              : "spherical";

        // OPTIONAL spatial fields must retain their definition-level threshold for COUNT.

        auto spatial_type = std::make_shared<DataTypeSpatial>(TYPE_GEOGRAPHY, crs, algorithm);

        ans.first = nullable ? make_nullable(spatial_type) : spatial_type;

    } else if (logicalType.__isset.STRING) {

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);

    } else if (logicalType.__isset.DECIMAL) {

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DECIMAL128I, nullable,

                                                                 logicalType.DECIMAL.precision,

                                                                 logicalType.DECIMAL.scale);

    } else if (logicalType.__isset.DATE) {

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATEV2, nullable);

    } else if (logicalType.__isset.INTEGER) {

        if (logicalType.INTEGER.isSigned) {

            if (logicalType.INTEGER.bitWidth <= 8) {

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_TINYINT, nullable);

            } else if (logicalType.INTEGER.bitWidth <= 16) {

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);

            } else if (logicalType.INTEGER.bitWidth <= 32) {

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);

            } else {

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);

            }

        } else {

            is_type_compatibility = true;

            if (logicalType.INTEGER.bitWidth <= 8) {

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);

            } else if (logicalType.INTEGER.bitWidth <= 16) {

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);

            } else if (logicalType.INTEGER.bitWidth <= 32) {

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);

            } else {

                ans.first = DataTypeFactory::instance().create_data_type(TYPE_LARGEINT, nullable);

            }

        }

    } else if (logicalType.__isset.TIME) {

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2, nullable);

    } else if (logicalType.__isset.TIMESTAMP) {

        if (_enable_mapping_timestamp_tz) {

            if (logicalType.TIMESTAMP.isAdjustedToUTC) {

                // treat TIMESTAMP with isAdjustedToUTC as TIMESTAMPTZ

                ans.first = DataTypeFactory::instance().create_data_type(

                        TYPE_TIMESTAMPTZ, nullable, 0,

                        logicalType.TIMESTAMP.unit.__isset.MILLIS ? 3 : 6);

                return ans;

            }

        }

        ans.first = DataTypeFactory::instance().create_data_type(

                TYPE_DATETIMEV2, nullable, 0, logicalType.TIMESTAMP.unit.__isset.MILLIS ? 3 : 6);

    } else if (logicalType.__isset.JSON) {

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);

    } else if (logicalType.__isset.UUID) {

        // UUID is a logical type; its binary storage does not make it an untyped byte array.

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_UUID, nullable);

    } else if (logicalType.__isset.FLOAT16) {

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_FLOAT, nullable);

    } else {

        throw Exception(Status::InternalError("Not supported parquet logicalType"));

    }

    return ans;

}

std::pair<DataTypePtr, bool> FieldDescriptor::convert_to_doris_type(

        const tparquet::SchemaElement& physical_schema, bool nullable) {

    std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(), false};

    bool& is_type_compatibility = ans.second;

    switch (physical_schema.converted_type) {

    case tparquet::ConvertedType::type::UTF8:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);

        break;

    case tparquet::ConvertedType::type::DECIMAL:

        ans.first = DataTypeFactory::instance().create_data_type(

                TYPE_DECIMAL128I, nullable, physical_schema.precision, physical_schema.scale);

        break;

    case tparquet::ConvertedType::type::DATE:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATEV2, nullable);

        break;

    case tparquet::ConvertedType::type::TIME_MILLIS:

        [[fallthrough]];

    case tparquet::ConvertedType::type::TIME_MICROS:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2, nullable);

        break;

    case tparquet::ConvertedType::type::TIMESTAMP_MILLIS:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable, 0, 3);

        break;

    case tparquet::ConvertedType::type::TIMESTAMP_MICROS:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable, 0, 6);

        break;

    case tparquet::ConvertedType::type::INT_8:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_TINYINT, nullable);

        break;

    case tparquet::ConvertedType::type::UINT_8:

        is_type_compatibility = true;

        [[fallthrough]];

    case tparquet::ConvertedType::type::INT_16:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);

        break;

    case tparquet::ConvertedType::type::UINT_16:

        is_type_compatibility = true;

        [[fallthrough]];

    case tparquet::ConvertedType::type::INT_32:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);

        break;

    case tparquet::ConvertedType::type::UINT_32:

        is_type_compatibility = true;

        [[fallthrough]];

    case tparquet::ConvertedType::type::INT_64:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);

        break;

    case tparquet::ConvertedType::type::UINT_64:

        is_type_compatibility = true;

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_LARGEINT, nullable);

        break;

    case tparquet::ConvertedType::type::JSON:

        ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);

        break;

    default:

        throw Exception(Status::InternalError("Not supported parquet ConvertedType: {}",

                                              physical_schema.converted_type));

    }

    return ans;

}

Status FieldDescriptor::parse_group_field(const std::vector<tparquet::SchemaElement>& t_schemas,

                                          size_t curr_pos, FieldSchema* group_field) {

    auto& group_schema = t_schemas[curr_pos];

    if (is_map_node(group_schema)) {

        // the map definition:

        // optional group <name> (MAP) {

        //   repeated group map (MAP_KEY_VALUE) {

        //     required <type> key;

        //     optional <type> value;

        //   }

        // }

        return parse_map_field(t_schemas, curr_pos, group_field);

    }

    if (is_list_node(group_schema)) {

        // the list definition:

        // optional group <name> (LIST) {

        //   repeated group [bag | list] { // hive or spark

        //     optional <type> [array_element | element]; // hive or spark

        //   }

        // }

        return parse_list_field(t_schemas, curr_pos, group_field);

    }

    if (is_repeated_node(group_schema)) {

        group_field->repetition_level++;

        group_field->definition_level++;

        group_field->children.resize(1);

        set_child_node_level(group_field, group_field->definition_level);

        auto struct_field = &group_field->children[0];

        // the list of struct:

        // repeated group <name> (LIST) {

        //   optional/required <type> <name>;

        //   ...

        // }

        // produce a non-null list<struct>

        RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos, struct_field));

        group_field->name = group_schema.name;

        group_field->lower_case_name = to_lower(group_field->name);

        group_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id

        group_field->data_type =

                std::make_shared<DataTypeArray>(make_nullable(struct_field->data_type));

        group_field->field_id = group_schema.__isset.field_id ? group_schema.field_id : -1;

    } else {

        RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos, group_field));

    }

    return Status::OK();

}

Status FieldDescriptor::parse_list_field(const std::vector<tparquet::SchemaElement>& t_schemas,

                                         size_t curr_pos, FieldSchema* list_field) {

    // the list definition:

    // spark and hive have three level schemas but with different schema name

    // spark: <column-name> - "list" - "element"

    // hive: <column-name> - "bag" - "array_element"

    // parse three level schemas to two level primitive like: LIST<INT>,

    // or nested structure like: LIST<MAP<INT, INT>>

    auto& first_level = t_schemas[curr_pos];

    if (first_level.num_children != 1) {

        return Status::InvalidArgument("List element should have only one child");

    }

    if (curr_pos + 1 >= t_schemas.size()) {

        return Status::InvalidArgument("List element should have the second level schema");

    }

    if (first_level.repetition_type == tparquet::FieldRepetitionType::REPEATED) {

        return Status::InvalidArgument("List element can't be a repeated schema");

    }

    // the repeated schema element

    auto& second_level = t_schemas[curr_pos + 1];

    if (second_level.repetition_type != tparquet::FieldRepetitionType::REPEATED) {

        return Status::InvalidArgument("The second level of list element should be repeated");

    }

    // This indicates if this list is nullable.

    bool is_optional = is_optional_node(first_level);

    if (is_optional) {

        list_field->definition_level++;

    }

    list_field->repetition_level++;

    list_field->definition_level++;

    list_field->children.resize(1);

    FieldSchema* list_child = &list_field->children[0];

    size_t num_children = num_children_node(second_level);

    if (num_children > 0) {

        if (num_children == 1 && !is_struct_list_node(second_level)) {

            // optional field, and the third level element is the nested structure in list

            // produce nested structure like: LIST<INT>, LIST<MAP>, LIST<LIST<...>>

            // skip bag/list, it's a repeated element.

            set_child_node_level(list_field, list_field->definition_level);

            RETURN_IF_ERROR(parse_node_field(t_schemas, curr_pos + 2, list_child));

        } else {

            // required field, produce the list of struct

            set_child_node_level(list_field, list_field->definition_level);

            RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos + 1, list_child));

        }

    } else if (num_children == 0) {

        // required two level list, for compatibility reason.

        set_child_node_level(list_field, list_field->definition_level);

        parse_physical_field(second_level, false, list_child);

        _next_schema_pos = curr_pos + 2;

    }

    list_field->name = first_level.name;

    list_field->lower_case_name = to_lower(first_level.name);

    list_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id

    list_field->data_type =

            std::make_shared<DataTypeArray>(make_nullable(list_field->children[0].data_type));

    if (is_optional) {

        list_field->data_type = make_nullable(list_field->data_type);

    }

    list_field->field_id = first_level.__isset.field_id ? first_level.field_id : -1;

    return Status::OK();

}

Status FieldDescriptor::parse_map_field(const std::vector<tparquet::SchemaElement>& t_schemas,

                                        size_t curr_pos, FieldSchema* map_field) {

    // the map definition in parquet:

    // optional group <name> (MAP) {

    //   repeated group map (MAP_KEY_VALUE) {

    //     required <type> key;

    //     optional <type> value;

    //   }

    // }

    // Map value can be optional, the map without values is a SET

    if (curr_pos + 2 >= t_schemas.size()) {

        return Status::InvalidArgument("Map element should have at least three levels");

    }

    auto& map_schema = t_schemas[curr_pos];

    if (map_schema.num_children != 1) {

        return Status::InvalidArgument(

                "Map element should have only one child(name='map', type='MAP_KEY_VALUE')");

    }

    if (is_repeated_node(map_schema)) {

        return Status::InvalidArgument("Map element can't be a repeated schema");

    }

    auto& map_key_value = t_schemas[curr_pos + 1];

    if (!is_group_node(map_key_value) || !is_repeated_node(map_key_value)) {

        return Status::InvalidArgument(

                "the second level in map must be a repeated group(key and value)");

    }

    auto& map_key = t_schemas[curr_pos + 2];

    if (!is_required_node(map_key)) {

        LOG(WARNING) << "Filed " << map_schema.name << " is map type, but with nullable key column";

    }

    if (map_key_value.num_children == 1) {

        // The map with three levels is a SET

        return parse_list_field(t_schemas, curr_pos, map_field);

    }

    if (map_key_value.num_children != 2) {

        // A standard map should have four levels

        return Status::InvalidArgument(

                "the second level in map(MAP_KEY_VALUE) should have two children");

    }

    // standard map

    bool is_optional = is_optional_node(map_schema);

    if (is_optional) {

        map_field->definition_level++;

    }

    map_field->repetition_level++;

    map_field->definition_level++;

    // Directly create key and value children instead of intermediate key_value node

    map_field->children.resize(2);

    // map is a repeated node, we should set the `repeated_parent_def_level` of its children as `definition_level`

    set_child_node_level(map_field, map_field->definition_level);

    auto key_field = &map_field->children[0];

    auto value_field = &map_field->children[1];

    // Parse key and value fields directly from the key_value group's children

    _next_schema_pos = curr_pos + 2; // Skip key_value group, go directly to key

    RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, key_field));

    RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, value_field));

    map_field->name = map_schema.name;

    map_field->lower_case_name = to_lower(map_field->name);

    map_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id

    map_field->data_type = std::make_shared<DataTypeMap>(make_nullable(key_field->data_type),

                                                         make_nullable(value_field->data_type));

    if (is_optional) {

        map_field->data_type = make_nullable(map_field->data_type);

    }

    map_field->field_id = map_schema.__isset.field_id ? map_schema.field_id : -1;

    return Status::OK();

}

Status FieldDescriptor::parse_struct_field(const std::vector<tparquet::SchemaElement>& t_schemas,

                                           size_t curr_pos, FieldSchema* struct_field) {

    // the nested column in parquet, parse group to struct.

    auto& struct_schema = t_schemas[curr_pos];

    bool is_optional = is_optional_node(struct_schema);

    if (is_optional) {

        struct_field->definition_level++;

    }

    auto num_children = struct_schema.num_children;

    struct_field->children.resize(num_children);

    set_child_node_level(struct_field, struct_field->repeated_parent_def_level);

    _next_schema_pos = curr_pos + 1;

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

        RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, &struct_field->children[i]));

    }

    struct_field->name = struct_schema.name;

    struct_field->lower_case_name = to_lower(struct_field->name);

    struct_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id

    struct_field->field_id = struct_schema.__isset.field_id ? struct_schema.field_id : -1;

    DataTypes res_data_types;

    std::vector<String> names;

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

        res_data_types.push_back(make_nullable(struct_field->children[i].data_type));

        names.push_back(struct_field->children[i].name);

    }

    struct_field->data_type = std::make_shared<DataTypeStruct>(res_data_types, names);

    if (is_optional) {

        struct_field->data_type = make_nullable(struct_field->data_type);

    }

    return Status::OK();

}

int FieldDescriptor::get_column_index(const std::string& column) const {

    for (int32_t i = 0; i < _fields.size(); i++) {

        if (_fields[i].name == column) {

            return i;

        }

    }

    return -1;

}

FieldSchema* FieldDescriptor::get_column(const std::string& name) const {

    auto it = _name_to_field.find(name);

    if (it != _name_to_field.end()) {

        return it->second;

    }

    throw Exception(Status::InternalError("Name {} not found in FieldDescriptor!", name));

    return nullptr;

}

void FieldDescriptor::get_column_names(std::unordered_set<std::string>* names) const {

    names->clear();

    for (const FieldSchema& f : _fields) {

        names->emplace(f.name);

    }

}

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

    std::stringstream ss;

    ss << "fields=[";

    for (int i = 0; i < _fields.size(); ++i) {

        if (i != 0) {

            ss << ", ";

        }

        ss << _fields[i].debug_string();

    }

    ss << "]";

    return ss.str();

}

void FieldDescriptor::assign_ids() {

    uint64_t next_id = 1;

    for (auto& field : _fields) {

        field.assign_ids(next_id);

    }

}

const FieldSchema* FieldDescriptor::find_column_by_id(uint64_t column_id) const {

    for (const auto& field : _fields) {

        if (auto result = field.find_column_by_id(column_id)) {

            return result;

        }

    }

    return nullptr;

}

void FieldSchema::assign_ids(uint64_t& next_id) {

    column_id = next_id++;

    for (auto& child : children) {

        child.assign_ids(next_id);

    }

    max_column_id = next_id - 1;

}

const FieldSchema* FieldSchema::find_column_by_id(uint64_t target_id) const {

    if (column_id == target_id) {

        return this;

    }

    for (const auto& child : children) {

        if (auto result = child.find_column_by_id(target_id)) {

            return result;

        }

    }

    return nullptr;

}

uint64_t FieldSchema::get_column_id() const {

    return column_id;

}

void FieldSchema::set_column_id(uint64_t id) {

    column_id = id;

}

uint64_t FieldSchema::get_max_column_id() const {

    return max_column_id;

}

} // namespace doris