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

#include <gen_cpp/PlanNodes_types.h>

#include <gen_cpp/Types_types.h>

#include "runtime/descriptors.h"

#include "util/string_util.h"

namespace doris {

/* static */

Status TableFormatReader::_extract_partition_values(

const TFileRangeDesc& range, const TupleDescriptor* tuple_descriptor,

std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>&

partition_values,

std::unordered_map<std::string, bool>* partition_value_is_null) {

partition_values.clear();

if (partition_value_is_null != nullptr) {

partition_value_is_null->clear();

}

if (range.__isset.columns_from_path_keys && tuple_descriptor != nullptr) {

DORIS_CHECK(range.__isset.columns_from_path);

DORIS_CHECK(range.columns_from_path.size() == range.columns_from_path_keys.size());

const bool has_null_flags = range.__isset.columns_from_path_is_null;

if (has_null_flags) {

DORIS_CHECK(range.columns_from_path_is_null.size() ==

range.columns_from_path_keys.size());

}

std::unordered_map<std::string, const SlotDescriptor*> name_to_slot;

for (auto* slot : tuple_descriptor->slots()) {

name_to_slot[slot->col_name()] = slot;

}

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

const auto& key = range.columns_from_path_keys[i];

const auto& value = range.columns_from_path[i];

auto slot_it = name_to_slot.find(key);

if (slot_it != name_to_slot.end()) {

partition_values.emplace(key, std::make_tuple(value, slot_it->second));

if (partition_value_is_null != nullptr) {

partition_value_is_null->emplace(

key, has_null_flags ? range.columns_from_path_is_null[i] : false);

}

}

}

}

return Status::OK();

}

Status TableFormatReader::on_before_init_reader(ReaderInitContext* ctx) {

_column_descs = ctx->column_descs;

_fill_col_name_to_block_idx = ctx->col_name_to_block_idx;

RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,

_fill_partition_values,

&_fill_partition_value_is_null));

for (const auto& desc : *ctx->column_descs) {

if (desc.category == ColumnCategory::REGULAR ||

desc.category == ColumnCategory::GENERATED) {

ctx->column_names.push_back(desc.name);

}

}

// Build default table_info_node from file column names (case-insensitive matching).

// Subclasses (OrcReader, ParquetReader, Hive, Iceberg, etc.) override on_before_init_reader

// and build their own table_info_node AFTER calling _extract_partition_values.

// For simple readers (CSV, JSON, etc.) that do NOT override, we build it here.

std::unordered_map<std::string, DataTypePtr> file_columns;

RETURN_IF_ERROR(get_columns(&file_columns));

// lowercase file name → original file name

std::unordered_map<std::string, std::string> lower_to_native;

for (const auto& [name, _] : file_columns) {

lower_to_native[doris::to_lower(name)] = name;

}

// Auto-compute missing columns for simple readers (CSV/JSON/Arrow/etc.).

// Parquet/ORC readers compute their own _fill_missing_defaults in _do_init_reader.

if (_column_descs) {

for (const auto& desc : *_column_descs) {

if (desc.category != ColumnCategory::REGULAR &&

desc.category != ColumnCategory::GENERATED) {

continue;

}

// Skip columns already handled as partition columns to avoid double-fill.

if (_fill_partition_values.contains(desc.name)) {

continue;

}

if (!lower_to_native.contains(doris::to_lower(desc.name))) {

_fill_missing_defaults[desc.name] = desc.default_expr;

_fill_missing_cols.insert(desc.name);

}

}

}

auto info_node = std::make_shared<TableSchemaChangeHelper::StructNode>();

for (const auto* slot : ctx->tuple_descriptor->slots()) {

auto it = lower_to_native.find(slot->col_name_lower_case());

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

info_node->add_children(slot->col_name(), it->second,

TableSchemaChangeHelper::ConstNode::get_instance());

} else {

info_node->add_not_exist_children(slot->col_name());

}

}

ctx->table_info_node = info_node;

return Status::OK();

}

} // namespace doris