// 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.
// This file is copied from
// https://github.com/ClickHouse/ClickHouse/blob/master/src/Functions/FunctionHelpers.cpp
// and modified by Doris

#include "exprs/function/function_helpers.h"

#include <fmt/format.h>
#include <glog/logging.h>

#include <algorithm>
#include <memory>
#include <ostream>
#include <string>
#include <vector>

#include "common/consts.h"
#include "common/status.h"
#include "core/block/column_with_type_and_name.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/data_type/data_type_nullable.h"
#include "exprs/function/function.h"
#include "util/string_util.h"

namespace doris {
std::tuple<Block, ColumnNumbers> create_block_with_nested_columns(const Block& block,
                                                                  const ColumnNumbers& args,
                                                                  const bool need_check_same) {
    Block res;
    ColumnNumbers res_args(args.size());
    res.reserve(args.size() + 1);

    // only build temp block by args column, if args[i] == args[j]
    // just keep one
    for (size_t i = 0; i < args.size(); ++i) {
        bool is_in_res = false;
        size_t pre_loc = 0;

        if (need_check_same) {
            for (int j = 0; j < i; ++j) {
                if (args[j] == args[i]) {
                    is_in_res = true;
                    pre_loc = res_args[j];
                    break;
                }
            }
        }

        if (!is_in_res) {
            const auto& col = block.get_by_position(args[i]);
            if (col.type->is_nullable()) {
                const DataTypePtr& nested_type =
                        static_cast<const DataTypeNullable&>(*col.type).get_nested_type();

                if (!col.column) {
                    res.insert({nullptr, nested_type, col.name});
                } else if (const auto* nullable =
                                   check_and_get_column<ColumnNullable>(*col.column)) {
                    const auto& nested_col = nullable->get_nested_column_ptr();
                    res.insert({nested_col, nested_type, col.name});
                } else if (const auto* const_column =
                                   check_and_get_column<ColumnConst>(*col.column)) {
                    const auto& nested_col =
                            check_and_get_column<ColumnNullable>(const_column->get_data_column())
                                    ->get_nested_column_ptr();
                    res.insert({ColumnConst::create(nested_col, col.column->size()), nested_type,
                                col.name});
                } else {
                    throw doris::Exception(
                            ErrorCode::INTERNAL_ERROR,
                            "Illegal column= {},  for DataTypeNullable" + col.column->get_name());
                }
            } else {
                res.insert(col);
            }

            res_args[i] = res.columns() - 1;
        } else {
            res_args[i] = (int)pre_loc;
        }
    }

    return {std::move(res), std::move(res_args)};
}

std::tuple<Block, ColumnNumbers, size_t> create_block_with_nested_columns(const Block& block,
                                                                          const ColumnNumbers& args,
                                                                          uint32_t result) {
    auto [res, res_args] = create_block_with_nested_columns(block, args, true);
    // insert result column in temp block
    res.insert(block.get_by_position(result));
    return {std::move(res), std::move(res_args), res.columns() - 1};
}

void validate_argument_type(const IFunction& func, const DataTypes& arguments,
                            size_t argument_index, bool (*validator_func)(const IDataType&),
                            const char* expected_type_description) {
    if (arguments.size() <= argument_index) {
        throw doris::Exception(ErrorCode::INVALID_ARGUMENT,
                               "Incorrect number of arguments of function {}" + func.get_name());
    }

    const auto& argument = arguments[argument_index];
    if (!validator_func(*argument)) {
        throw doris::Exception(ErrorCode::INVALID_ARGUMENT,
                               "Illegal type {} of {} argument of function {} expected {}",
                               argument->get_name(), argument_index, func.get_name(),
                               expected_type_description);
    }
}

const ColumnConst* check_and_get_column_const_string_or_fixedstring(const IColumn* column) {
    if (!is_column_const(*column)) return {};

    const ColumnConst* res = assert_cast<const ColumnConst*, TypeCheckOnRelease::DISABLE>(column);

    if (is_column<ColumnString>(&res->get_data_column())) return res;

    return {};
}
} // namespace doris