blob: fd12cbcd9861af45af8e75b98c493aa7ca4c952c [file]
/*
* 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