/*
 * 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 "paimon/common/data/blob_utils.h"

#include <cstddef>
#include <set>
#include <string_view>
#include <vector>

#include "arrow/api.h"
#include "arrow/array/array_nested.h"
#include "arrow/type.h"
#include "fmt/format.h"
#include "paimon/common/data/blob_defs.h"
#include "paimon/common/data/blob_view_struct.h"
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/string_utils.h"
#include "paimon/data/blob_descriptor.h"
namespace arrow {
class Array;
}

namespace paimon {
BlobUtils::SeparatedSchemas BlobUtils::SeparateBlobSchema(
    const std::shared_ptr<arrow::Schema>& schema, const std::set<std::string>& inline_fields) {
    std::vector<std::shared_ptr<arrow::Field>> main_fields;
    std::vector<std::shared_ptr<arrow::Field>> blob_fields;
    for (int32_t i = 0; i < schema->num_fields(); i++) {
        auto field = schema->field(i);
        if (IsAnyBlobField(field) && inline_fields.count(field->name()) == 0) {
            // Non-inline BLOB, ARRAY<BLOB> or MAP<..., BLOB> -> goes to blob file
            blob_fields.emplace_back(field);
        } else {
            // Non-blob fields OR inline BLOB fields -> stay in main
            main_fields.emplace_back(field);
        }
    }
    SeparatedSchemas result;
    result.main_schema = arrow::schema(main_fields);
    result.blob_schema = arrow::schema(blob_fields);
    return result;
}

Result<BlobUtils::SeparatedStructArrays> BlobUtils::SeparateBlobArray(
    const std::shared_ptr<arrow::StructArray>& struct_array,
    const std::set<std::string>& inline_fields) {
    std::shared_ptr<arrow::StructType> old_type =
        checked_pointer_cast<arrow::StructType>(struct_array->type());
    const auto& old_fields = old_type->fields();
    const auto& old_arrays = struct_array->fields();

    arrow::ArrayVector main_arrays;
    arrow::ArrayVector blob_arrays;
    arrow::FieldVector main_fields;
    arrow::FieldVector blob_fields;

    for (size_t i = 0; i < old_fields.size(); i++) {
        if (IsAnyBlobField(old_fields[i]) && inline_fields.count(old_fields[i]->name()) == 0) {
            blob_fields.push_back(old_fields[i]);
            blob_arrays.push_back(old_arrays[i]);
        } else {
            main_fields.push_back(old_fields[i]);
            main_arrays.push_back(old_arrays[i]);
        }
    }

    if (blob_fields.empty()) {
        return Status::Invalid(
            "SeparateBlobArray expects at least one non-inline blob field, but got none.");
    }

    SeparatedStructArrays result;
    if (!main_fields.empty()) {
        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(result.main_array,
                                          arrow::StructArray::Make(main_arrays, main_fields));
    }
    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(result.blob_array,
                                      arrow::StructArray::Make(blob_arrays, blob_fields));
    return result;
}

bool BlobUtils::IsBlobField(const std::shared_ptr<arrow::Field>& field) {
    if (field == nullptr) {
        return false;
    }
    const auto& type = field->type();
    if (type->id() != arrow::Type::LARGE_BINARY) {
        return false;
    }
    if (!field->HasMetadata()) {
        return false;
    }
    return IsBlobMetadata(field->metadata());
}

bool BlobUtils::IsArrayBlobField(const std::shared_ptr<arrow::Field>& field) {
    if (field == nullptr || field->type()->id() != arrow::Type::LIST) {
        return false;
    }
    const auto& list_type = checked_cast<const arrow::ListType&>(*field->type());
    // Arrow's C schema importer passes MakeChildField(0) directly to ListType, retaining the
    // element field's metadata.
    return IsBlobField(list_type.value_field());
}

bool BlobUtils::IsMapBlobField(const std::shared_ptr<arrow::Field>& field) {
    if (field == nullptr || field->type()->id() != arrow::Type::MAP) {
        return false;
    }
    const auto& map_type = checked_cast<const arrow::MapType&>(*field->type());
    // Arrow's C schema bridge does not retain nested field metadata for MapType. Paimon's
    // ordinary binary type is BINARY, so LARGE_BINARY uniquely identifies a BLOB value here.
    return map_type.item_type()->id() == arrow::Type::LARGE_BINARY;
}

bool BlobUtils::IsAnyBlobField(const std::shared_ptr<arrow::Field>& field) {
    return IsBlobField(field) || IsArrayBlobField(field) || IsMapBlobField(field);
}

bool BlobUtils::IsArrayBlobPlaceholder(const arrow::ListArray& array, int64_t row) {
    if (array.IsNull(row) || array.value_length(row) != 1) {
        return false;
    }
    const std::shared_ptr<arrow::Array>& values = array.values();
    if (values->type_id() != arrow::Type::LARGE_BINARY) {
        return false;
    }
    const int64_t value_index = array.value_offset(row);
    if (values->IsNull(value_index)) {
        return false;
    }
    const auto& binary_values = checked_cast<const arrow::LargeBinaryArray&>(*values);
    const std::string_view value = binary_values.GetView(value_index);
    return BlobDefs::IsPlaceholderSentinel(value.data(), value.size());
}

bool BlobUtils::IsMapBlobPlaceholder(const arrow::MapArray& array, int64_t row) {
    if (array.IsNull(row) || array.value_length(row) != 2) {
        return false;
    }
    const int64_t entry_index = array.value_offset(row);
    const std::shared_ptr<arrow::Array>& items = array.items();
    if (!items->IsNull(entry_index) || !items->IsNull(entry_index + 1)) {
        return false;
    }
    const std::shared_ptr<arrow::Array>& keys = array.keys();
    return keys->RangeEquals(entry_index, entry_index + 1, entry_index + 1, *keys);
}

bool BlobUtils::IsSupportedMapBlobKeyType(const arrow::DataType& key_type) {
    switch (key_type.id()) {
        case arrow::Type::BOOL:
        case arrow::Type::INT8:
        case arrow::Type::INT16:
        case arrow::Type::INT32:
        case arrow::Type::INT64:
        case arrow::Type::DATE32:
        case arrow::Type::DECIMAL128:
        case arrow::Type::STRING:
        case arrow::Type::BINARY:
            return true;
        default:
            return false;
    }
}

bool BlobUtils::IsBlobMetadata(const std::shared_ptr<const arrow::KeyValueMetadata>& metadata) {
    if (!metadata) {
        return false;
    }
    auto extension_name = metadata->Get(BlobDefs::kExtensionTypeKey);
    if (!extension_name.ok()) {
        return false;
    }
    return extension_name.ValueUnsafe() == BlobDefs::kExtensionTypeValue;
}

bool BlobUtils::IsBlobFile(const std::string& file_name) {
    return StringUtils::EndsWith(file_name, ".blob");
}

std::shared_ptr<arrow::Field> BlobUtils::ToArrowField(
    const std::string& field_name, bool nullable,
    std::unordered_map<std::string, std::string> metadata) {
    metadata[BlobDefs::kExtensionTypeKey] = BlobDefs::kExtensionTypeValue;
    return arrow::field(field_name, arrow::large_binary(), nullable,
                        std::make_shared<arrow::KeyValueMetadata>(metadata));
}

Status BlobUtils::ValidateBlobInlineFields(const std::shared_ptr<arrow::StructArray>& struct_array,
                                           const std::set<std::string>& field_names,
                                           const std::string& config_label) {
    if (field_names.empty()) {
        return Status::OK();
    }
    if (!struct_array) {
        return Status::Invalid("array in ValidateBlobInlineFields must be a struct_array");
    }

    bool is_descriptor = (config_label == "blob-descriptor-field");
    for (const auto& field_name : field_names) {
        auto field_array = struct_array->GetFieldByName(field_name);
        if (!field_array) {
            continue;
        }
        if (field_array->type_id() != arrow::Type::LARGE_BINARY) {
            return Status::Invalid(
                fmt::format("cannot cast array for field {} to LargeBinaryArray", field_name));
        }
        const auto* binary_array = checked_cast<const arrow::LargeBinaryArray*>(field_array.get());
        for (int64_t row = 0; row < binary_array->length(); ++row) {
            if (binary_array->IsNull(row)) {
                continue;
            }
            auto value = binary_array->GetView(row);
            Result<bool> valid = is_descriptor
                                     ? BlobDescriptor::IsBlobDescriptor(value.data(), value.size())
                                     : BlobViewStruct::IsBlobViewStruct(value.data(), value.size());
            PAIMON_ASSIGN_OR_RAISE(bool is_valid, std::move(valid));
            if (!is_valid) {
                return Status::Invalid(fmt::format(
                    "BLOB inline field {} require values to be set as corresponding type.",
                    field_name));
            }
        }
    }
    return Status::OK();
}

std::vector<DataField> BlobUtils::ConvertBlobInlineDataFields(
    const std::vector<DataField>& data_fields, const std::vector<std::string>& blob_inline_fields) {
    if (blob_inline_fields.empty()) {
        return data_fields;
    }

    std::set<std::string> blob_inline_field_set(blob_inline_fields.begin(),
                                                blob_inline_fields.end());
    std::vector<DataField> converted_fields;
    converted_fields.reserve(data_fields.size());
    for (const auto& data_field : data_fields) {
        if (blob_inline_field_set.find(data_field.Name()) == blob_inline_field_set.end()) {
            converted_fields.push_back(data_field);
            continue;
        }

        auto binary_field = arrow::field(data_field.Name(), arrow::binary(), data_field.Nullable(),
                                         data_field.ArrowField()->metadata());
        converted_fields.emplace_back(data_field.Id(), binary_field, data_field.Description());
    }
    return converted_fields;
}

}  // namespace paimon