| /* | |
| * 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 <algorithm> | |
| #include <bit> | |
| #include <span> | |
| #include <arrow/array/builder_binary.h> | |
| #include <arrow/array/builder_decimal.h> | |
| #include <arrow/array/builder_nested.h> | |
| #include <arrow/array/builder_primitive.h> | |
| #include <arrow/extension_type.h> | |
| #include <arrow/json/from_string.h> | |
| #include <arrow/scalar.h> | |
| #include <arrow/type.h> | |
| #include <arrow/util/decimal.h> | |
| #include <avro/Generic.hh> | |
| #include <avro/Node.hh> | |
| #include <avro/NodeImpl.hh> | |
| #include <avro/Types.hh> | |
| #include "iceberg/arrow/arrow_status_internal.h" | |
| #include "iceberg/arrow/literal_util_internal.h" | |
| #include "iceberg/avro/avro_data_util_internal.h" | |
| #include "iceberg/avro/avro_schema_util_internal.h" | |
| #include "iceberg/metadata_columns.h" | |
| #include "iceberg/schema.h" | |
| #include "iceberg/schema_internal.h" | |
| #include "iceberg/schema_util.h" | |
| #include "iceberg/util/checked_cast.h" | |
| #include "iceberg/util/macros.h" | |
| namespace iceberg::avro { | |
| using ::iceberg::arrow::ToErrorKind; | |
| namespace { | |
| /// \brief Forward declaration for mutual recursion. | |
| Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node, | |
| const ::avro::GenericDatum& avro_datum, | |
| const FieldProjection& projection, | |
| const SchemaField& projected_field, | |
| const arrow::MetadataColumnContext& metadata_context, | |
| ::arrow::ArrayBuilder* array_builder); | |
| /// \brief Append Avro record data to Arrow struct builder. | |
| Status AppendStructToBuilder(const ::avro::NodePtr& avro_node, | |
| const ::avro::GenericDatum& avro_datum, | |
| const std::span<const FieldProjection>& projections, | |
| const StructType& struct_type, | |
| const arrow::MetadataColumnContext& metadata_context, | |
| ::arrow::ArrayBuilder* array_builder) { | |
| if (avro_node->type() != ::avro::AVRO_RECORD) { | |
| return InvalidArgument("Expected Avro record, got type: {}", ToString(avro_node)); | |
| } | |
| const auto& avro_record = avro_datum.value<::avro::GenericRecord>(); | |
| auto* struct_builder = internal::checked_cast<::arrow::StructBuilder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(struct_builder->Append()); | |
| for (size_t i = 0; i < projections.size(); ++i) { | |
| const auto& field_projection = projections[i]; | |
| const auto& expected_field = struct_type.fields()[i]; | |
| auto* field_builder = struct_builder->field_builder(static_cast<int>(i)); | |
| if (field_projection.kind == FieldProjection::Kind::kProjected) { | |
| size_t avro_field_index = std::get<size_t>(field_projection.from); | |
| if (avro_field_index >= avro_record.fieldCount()) { | |
| return InvalidArgument("Avro field index {} out of bound {}", avro_field_index, | |
| avro_record.fieldCount()); | |
| } | |
| const auto& avro_field_node = avro_node->leafAt(avro_field_index); | |
| const auto& avro_field_datum = avro_record.fieldAt(avro_field_index); | |
| ICEBERG_RETURN_UNEXPECTED(AppendFieldToBuilder(avro_field_node, avro_field_datum, | |
| field_projection, expected_field, | |
| metadata_context, field_builder)); | |
| } else if (field_projection.kind == FieldProjection::Kind::kNull) { | |
| ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull()); | |
| } else if (field_projection.kind == FieldProjection::Kind::kDefault) { | |
| ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder(field_projection, field_builder)); | |
| } else if (field_projection.kind == FieldProjection::Kind::kMetadata) { | |
| int32_t field_id = expected_field.field_id(); | |
| if (field_id == MetadataColumns::kFilePathColumnId) { | |
| auto string_builder = | |
| internal::checked_cast<::arrow::StringBuilder*>(field_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(string_builder->Append(metadata_context.file_path)); | |
| } else if (field_id == MetadataColumns::kFilePositionColumnId) { | |
| auto int_builder = internal::checked_cast<::arrow::Int64Builder*>(field_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(int_builder->Append(metadata_context.next_file_pos)); | |
| } else if (MetadataColumns::IsRowLineageColumn(field_id)) { | |
| ICEBERG_RETURN_UNEXPECTED(arrow::AppendInheritedRowLineageValue( | |
| field_id, metadata_context, field_builder)); | |
| } else { | |
| return NotSupported("Unsupported metadata column field id: {}", field_id); | |
| } | |
| } else { | |
| return NotImplemented("Unsupported field projection kind: {}", | |
| ToString(field_projection.kind)); | |
| } | |
| } | |
| return {}; | |
| } | |
| /// \brief Append Avro array data to Arrow list builder. | |
| Status AppendListToBuilder(const ::avro::NodePtr& avro_node, | |
| const ::avro::GenericDatum& avro_datum, | |
| const FieldProjection& element_projection, | |
| const ListType& list_type, | |
| const arrow::MetadataColumnContext& metadata_context, | |
| ::arrow::ArrayBuilder* array_builder) { | |
| if (avro_node->type() != ::avro::AVRO_ARRAY) { | |
| return InvalidArgument("Expected Avro array, got type: {}", ToString(avro_node)); | |
| } | |
| const auto& avro_array = avro_datum.value<::avro::GenericArray>(); | |
| auto* list_builder = internal::checked_cast<::arrow::ListBuilder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(list_builder->Append()); | |
| auto* value_builder = list_builder->value_builder(); | |
| const auto& element_node = avro_node->leafAt(0); | |
| const auto& element_field = list_type.fields().back(); | |
| for (const auto& element : avro_array.value()) { | |
| ICEBERG_RETURN_UNEXPECTED(AppendFieldToBuilder(element_node, element, | |
| element_projection, element_field, | |
| metadata_context, value_builder)); | |
| } | |
| return {}; | |
| } | |
| /// \brief Append Avro map data to Arrow map builder. | |
| Status AppendMapToBuilder(const ::avro::NodePtr& avro_node, | |
| const ::avro::GenericDatum& avro_datum, | |
| const FieldProjection& key_projection, | |
| const FieldProjection& value_projection, | |
| const MapType& map_type, | |
| const arrow::MetadataColumnContext& metadata_context, | |
| ::arrow::ArrayBuilder* array_builder) { | |
| auto* map_builder = internal::checked_cast<::arrow::MapBuilder*>(array_builder); | |
| if (avro_node->type() == ::avro::AVRO_MAP) { | |
| // Handle regular Avro map: map<string, value> | |
| const auto& avro_map = avro_datum.value<::avro::GenericMap>(); | |
| const auto& map_entries = avro_map.value(); | |
| const auto& key_node = avro_node->leafAt(0); | |
| const auto& value_node = avro_node->leafAt(1); | |
| const auto& key_field = map_type.key(); | |
| const auto& value_field = map_type.value(); | |
| ICEBERG_ARROW_RETURN_NOT_OK(map_builder->Append()); | |
| auto* key_builder = map_builder->key_builder(); | |
| auto* item_builder = map_builder->item_builder(); | |
| for (const auto& entry : map_entries) { | |
| ICEBERG_RETURN_UNEXPECTED(AppendFieldToBuilder(key_node, entry.first, | |
| key_projection, key_field, | |
| metadata_context, key_builder)); | |
| ICEBERG_RETURN_UNEXPECTED(AppendFieldToBuilder(value_node, entry.second, | |
| value_projection, value_field, | |
| metadata_context, item_builder)); | |
| } | |
| return {}; | |
| } else if (avro_node->type() == ::avro::AVRO_ARRAY && HasMapLogicalType(avro_node)) { | |
| // Handle array-based map: list<struct<key, value>> | |
| const auto& avro_array = avro_datum.value<::avro::GenericArray>(); | |
| const auto& array_entries = avro_array.value(); | |
| const auto& key_field = map_type.key(); | |
| const auto& value_field = map_type.value(); | |
| ICEBERG_ARROW_RETURN_NOT_OK(map_builder->Append()); | |
| auto* key_builder = map_builder->key_builder(); | |
| auto* item_builder = map_builder->item_builder(); | |
| const auto& record_node = avro_node->leafAt(0); | |
| if (record_node->type() != ::avro::AVRO_RECORD || record_node->leaves() != 2) { | |
| return InvalidArgument( | |
| "Array-based map must contain records with exactly 2 fields, got: {}", | |
| ToString(record_node)); | |
| } | |
| const auto& key_node = record_node->leafAt(0); | |
| const auto& value_node = record_node->leafAt(1); | |
| for (const auto& entry : array_entries) { | |
| const auto& record = entry.value<::avro::GenericRecord>(); | |
| ICEBERG_RETURN_UNEXPECTED(AppendFieldToBuilder(key_node, record.fieldAt(0), | |
| key_projection, key_field, | |
| metadata_context, key_builder)); | |
| ICEBERG_RETURN_UNEXPECTED(AppendFieldToBuilder(value_node, record.fieldAt(1), | |
| value_projection, value_field, | |
| metadata_context, item_builder)); | |
| } | |
| return {}; | |
| } else { | |
| return InvalidArgument("Expected Avro map or array with map logical type, got: {}", | |
| ToString(avro_node)); | |
| } | |
| } | |
| /// \brief Append nested Avro data to Arrow array builder based on type. | |
| Status AppendNestedValueToBuilder(const ::avro::NodePtr& avro_node, | |
| const ::avro::GenericDatum& avro_datum, | |
| const std::span<const FieldProjection>& projections, | |
| const NestedType& projected_type, | |
| const arrow::MetadataColumnContext& metadata_context, | |
| ::arrow::ArrayBuilder* array_builder) { | |
| switch (projected_type.type_id()) { | |
| case TypeId::kStruct: { | |
| const auto& struct_type = internal::checked_cast<const StructType&>(projected_type); | |
| return AppendStructToBuilder(avro_node, avro_datum, projections, struct_type, | |
| metadata_context, array_builder); | |
| } | |
| case TypeId::kList: { | |
| if (projections.size() != 1) { | |
| return InvalidArgument("Expected 1 projection for list, got: {}", | |
| projections.size()); | |
| } | |
| const auto& list_type = internal::checked_cast<const ListType&>(projected_type); | |
| return AppendListToBuilder(avro_node, avro_datum, projections[0], list_type, | |
| metadata_context, array_builder); | |
| } | |
| case TypeId::kMap: { | |
| if (projections.size() != 2) { | |
| return InvalidArgument("Expected 2 projections for map, got: {}", | |
| projections.size()); | |
| } | |
| const auto& map_type = internal::checked_cast<const MapType&>(projected_type); | |
| return AppendMapToBuilder(avro_node, avro_datum, projections[0], projections[1], | |
| map_type, metadata_context, array_builder); | |
| } | |
| default: | |
| return InvalidArgument("Unsupported nested type: {}", projected_type.ToString()); | |
| } | |
| } | |
| Status AppendPrimitiveValueToBuilder(const ::avro::NodePtr& avro_node, | |
| const ::avro::GenericDatum& avro_datum, | |
| const SchemaField& projected_field, | |
| ::arrow::ArrayBuilder* array_builder) { | |
| const auto& projected_type = *projected_field.type(); | |
| if (!projected_type.is_primitive()) { | |
| return InvalidArgument("Expected primitive type, got: {}", projected_type.ToString()); | |
| } | |
| switch (projected_type.type_id()) { | |
| case TypeId::kBoolean: { | |
| if (avro_node->type() != ::avro::AVRO_BOOL) { | |
| return InvalidArgument("Expected Avro boolean for boolean field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = internal::checked_cast<::arrow::BooleanBuilder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<bool>())); | |
| return {}; | |
| } | |
| case TypeId::kInt: { | |
| if (avro_node->type() != ::avro::AVRO_INT) { | |
| return InvalidArgument("Expected Avro int for int field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = internal::checked_cast<::arrow::Int32Builder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<int32_t>())); | |
| return {}; | |
| } | |
| case TypeId::kLong: { | |
| auto* builder = internal::checked_cast<::arrow::Int64Builder*>(array_builder); | |
| if (avro_node->type() == ::avro::AVRO_LONG) { | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<int64_t>())); | |
| } else if (avro_node->type() == ::avro::AVRO_INT) { | |
| ICEBERG_ARROW_RETURN_NOT_OK( | |
| builder->Append(static_cast<int64_t>(avro_datum.value<int32_t>()))); | |
| } else { | |
| return InvalidArgument("Expected Avro int/long for long field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| return {}; | |
| } | |
| case TypeId::kFloat: { | |
| if (avro_node->type() != ::avro::AVRO_FLOAT) { | |
| return InvalidArgument("Expected Avro float for float field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = internal::checked_cast<::arrow::FloatBuilder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<float>())); | |
| return {}; | |
| } | |
| case TypeId::kDouble: { | |
| auto* builder = internal::checked_cast<::arrow::DoubleBuilder*>(array_builder); | |
| if (avro_node->type() == ::avro::AVRO_DOUBLE) { | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<double>())); | |
| } else if (avro_node->type() == ::avro::AVRO_FLOAT) { | |
| ICEBERG_ARROW_RETURN_NOT_OK( | |
| builder->Append(static_cast<double>(avro_datum.value<float>()))); | |
| } else { | |
| return InvalidArgument("Expected Avro float/double for double field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| return {}; | |
| } | |
| case TypeId::kString: { | |
| if (avro_node->type() != ::avro::AVRO_STRING) { | |
| return InvalidArgument("Expected Avro string for string field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = internal::checked_cast<::arrow::StringBuilder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<std::string>())); | |
| return {}; | |
| } | |
| case TypeId::kBinary: | |
| case TypeId::kGeometry: | |
| case TypeId::kGeography: { | |
| if (avro_node->type() != ::avro::AVRO_BYTES) { | |
| return InvalidArgument("Expected Avro bytes for binary field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = internal::checked_cast<::arrow::BinaryBuilder*>(array_builder); | |
| const auto& bytes = avro_datum.value<std::vector<uint8_t>>(); | |
| ICEBERG_ARROW_RETURN_NOT_OK( | |
| builder->Append(bytes.data(), static_cast<int32_t>(bytes.size()))); | |
| return {}; | |
| } | |
| case TypeId::kFixed: { | |
| if (avro_node->type() != ::avro::AVRO_FIXED) { | |
| return InvalidArgument("Expected Avro fixed for fixed field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| const auto& fixed = avro_datum.value<::avro::GenericFixed>(); | |
| const auto& fixed_type = internal::checked_cast<const FixedType&>(projected_type); | |
| if (static_cast<size_t>(fixed.value().size()) != fixed_type.length()) { | |
| return InvalidArgument("Expected Avro fixed[{}], got: {}", fixed_type.length(), | |
| ToString(avro_node)); | |
| } | |
| auto* builder = | |
| internal::checked_cast<::arrow::FixedSizeBinaryBuilder*>(array_builder); | |
| const auto& value = fixed.value(); | |
| ICEBERG_ARROW_RETURN_NOT_OK( | |
| builder->Append(reinterpret_cast<const uint8_t*>(value.data()))); | |
| return {}; | |
| } | |
| case TypeId::kUuid: { | |
| if (avro_node->type() != ::avro::AVRO_FIXED || | |
| avro_node->logicalType().type() != ::avro::LogicalType::UUID) { | |
| return InvalidArgument("Expected Avro fixed for uuid field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = | |
| internal::checked_cast<::arrow::FixedSizeBinaryBuilder*>(array_builder); | |
| const auto& fixed = avro_datum.value<::avro::GenericFixed>(); | |
| if (fixed.value().size() != 16) { | |
| return InvalidArgument("Expected UUID fixed length 16, got: {}", | |
| fixed.value().size()); | |
| } | |
| const auto& value = fixed.value(); | |
| ICEBERG_ARROW_RETURN_NOT_OK( | |
| builder->Append(reinterpret_cast<const uint8_t*>(value.data()))); | |
| return {}; | |
| } | |
| case TypeId::kDecimal: { | |
| if (avro_node->type() != ::avro::AVRO_FIXED || | |
| avro_node->logicalType().type() != ::avro::LogicalType::DECIMAL) { | |
| return InvalidArgument( | |
| "Expected Avro fixed with decimal logical type for decimal field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| const auto& fixed = avro_datum.value<::avro::GenericFixed>(); | |
| const auto& value = fixed.value(); | |
| ICEBERG_ARROW_ASSIGN_OR_RETURN( | |
| auto decimal, ::arrow::Decimal128::FromBigEndian(value.data(), value.size())); | |
| auto* builder = internal::checked_cast<::arrow::Decimal128Builder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(decimal)); | |
| return {}; | |
| } | |
| case TypeId::kDate: { | |
| if (!IsAvroDateOrPlainInt(avro_node)) { | |
| return InvalidArgument( | |
| "Expected Avro int with DATE logical type or plain int for date field, got: " | |
| "{}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = internal::checked_cast<::arrow::Date32Builder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<int32_t>())); | |
| return {}; | |
| } | |
| case TypeId::kTime: { | |
| if (avro_node->type() != ::avro::AVRO_LONG || | |
| avro_node->logicalType().type() != ::avro::LogicalType::TIME_MICROS) { | |
| return InvalidArgument( | |
| "Expected Avro long with TIME_MICROS for time field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = internal::checked_cast<::arrow::Time64Builder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<int64_t>())); | |
| return {}; | |
| } | |
| case TypeId::kTimestamp: | |
| case TypeId::kTimestampTz: { | |
| if (avro_node->type() != ::avro::AVRO_LONG || | |
| avro_node->logicalType().type() != ::avro::LogicalType::TIMESTAMP_MICROS) { | |
| return InvalidArgument( | |
| "Expected Avro long with TIMESTAMP_MICROS for timestamp field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = internal::checked_cast<::arrow::TimestampBuilder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<int64_t>())); | |
| return {}; | |
| } | |
| case TypeId::kTimestampNs: | |
| case TypeId::kTimestampTzNs: { | |
| if (avro_node->type() != ::avro::AVRO_LONG || | |
| avro_node->logicalType().type() != ::avro::LogicalType::TIMESTAMP_NANOS) { | |
| return InvalidArgument( | |
| "Expected Avro long with TIMESTAMP_NANOS for timestamp field, got: {}", | |
| ToString(avro_node)); | |
| } | |
| auto* builder = internal::checked_cast<::arrow::TimestampBuilder*>(array_builder); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->Append(avro_datum.value<int64_t>())); | |
| return {}; | |
| } | |
| default: | |
| return InvalidArgument("Unsupported primitive type {} to append avro node {}", | |
| projected_field.type()->ToString(), ToString(avro_node)); | |
| } | |
| } | |
| /// \brief Dispatch to appropriate handlers based on the projection kind. | |
| Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node, | |
| const ::avro::GenericDatum& avro_datum, | |
| const FieldProjection& projection, | |
| const SchemaField& projected_field, | |
| const arrow::MetadataColumnContext& metadata_context, | |
| ::arrow::ArrayBuilder* array_builder) { | |
| if (projection.kind == FieldProjection::Kind::kNull) { | |
| ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull()); | |
| return {}; | |
| } | |
| if (projection.kind == FieldProjection::Kind::kDefault) { | |
| return AppendDefaultToBuilder(projection, array_builder); | |
| } | |
| const bool is_row_lineage = | |
| MetadataColumns::IsRowLineageColumn(projected_field.field_id()); | |
| if (avro_node->type() == ::avro::AVRO_UNION) { | |
| size_t branch = avro_datum.unionBranch(); | |
| if (avro_node->leafAt(branch)->type() == ::avro::AVRO_NULL) { | |
| if (is_row_lineage) { | |
| return arrow::AppendInheritedRowLineageValue(projected_field.field_id(), | |
| metadata_context, array_builder); | |
| } | |
| ICEBERG_ARROW_RETURN_NOT_OK(array_builder->AppendNull()); | |
| return {}; | |
| } else { | |
| return AppendFieldToBuilder(avro_node->leafAt(branch), avro_datum, projection, | |
| projected_field, metadata_context, array_builder); | |
| } | |
| } | |
| const auto& projected_type = *projected_field.type(); | |
| if (projected_type.is_primitive()) { | |
| return AppendPrimitiveValueToBuilder(avro_node, avro_datum, projected_field, | |
| array_builder); | |
| } else { | |
| const auto& nested_type = internal::checked_cast<const NestedType&>(projected_type); | |
| return AppendNestedValueToBuilder(avro_node, avro_datum, projection.children, | |
| nested_type, metadata_context, array_builder); | |
| } | |
| } | |
| } // namespace | |
| namespace { | |
| Result<std::shared_ptr<::arrow::Scalar>> MakeDefaultScalar( | |
| const Literal& literal, const std::shared_ptr<::arrow::DataType>& builder_type) { | |
| // The builder's own memory pool is not exposed, so the small scalar buffer uses the | |
| // default pool. | |
| ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Scalar> scalar, | |
| arrow::ToArrowScalar(literal, ::arrow::default_memory_pool())); | |
| // For an extension builder (e.g. `arrow.uuid`) target its storage type: ToArrowScalar | |
| // yields the storage scalar (fixed_size_binary(16) for uuid) and Scalar::CastTo has no | |
| // kernel that targets an extension type. This mirrors MakeDefaultArray's extension | |
| // handling. | |
| std::shared_ptr<::arrow::DataType> target_type = builder_type; | |
| if (target_type->id() == ::arrow::Type::EXTENSION) { | |
| target_type = internal::checked_cast<const ::arrow::ExtensionType&>(*target_type) | |
| .storage_type(); | |
| } | |
| if (!scalar->type->Equals(*target_type)) { | |
| ICEBERG_ARROW_ASSIGN_OR_RETURN(scalar, scalar->CastTo(target_type)); | |
| } | |
| return scalar; | |
| } | |
| Status PrepareStructDefaultScalars(std::span<FieldProjection> projections, | |
| ::arrow::ArrayBuilder* builder); | |
| // Recurse into whatever nested builder this projection describes, so a default cached for | |
| // a struct field is found at any nesting depth (e.g. `list<list<struct<...>>>`) instead | |
| // of only when the collection's child is an immediate struct. | |
| Status PrepareNestedDefaultScalars(FieldProjection& projection, | |
| ::arrow::ArrayBuilder* builder) { | |
| if (projection.kind != FieldProjection::Kind::kProjected || | |
| projection.children.empty()) { | |
| return {}; | |
| } | |
| switch (builder->type()->id()) { | |
| case ::arrow::Type::STRUCT: | |
| return PrepareStructDefaultScalars(projection.children, builder); | |
| case ::arrow::Type::LIST: { | |
| // List projections store a single child for the element. | |
| auto* list_builder = internal::checked_cast<::arrow::ListBuilder*>(builder); | |
| return PrepareNestedDefaultScalars(projection.children[0], | |
| list_builder->value_builder()); | |
| } | |
| case ::arrow::Type::LARGE_LIST: { | |
| auto* list_builder = internal::checked_cast<::arrow::LargeListBuilder*>(builder); | |
| return PrepareNestedDefaultScalars(projection.children[0], | |
| list_builder->value_builder()); | |
| } | |
| case ::arrow::Type::MAP: { | |
| auto* map_builder = internal::checked_cast<::arrow::MapBuilder*>(builder); | |
| if (projection.children.size() >= 1) { | |
| ICEBERG_RETURN_UNEXPECTED(PrepareNestedDefaultScalars( | |
| projection.children[0], map_builder->key_builder())); | |
| } | |
| if (projection.children.size() >= 2) { | |
| ICEBERG_RETURN_UNEXPECTED(PrepareNestedDefaultScalars( | |
| projection.children[1], map_builder->item_builder())); | |
| } | |
| return {}; | |
| } | |
| default: | |
| return {}; | |
| } | |
| } | |
| Status PrepareStructDefaultScalars(std::span<FieldProjection> projections, | |
| ::arrow::ArrayBuilder* builder) { | |
| auto* struct_builder = internal::checked_cast<::arrow::StructBuilder*>(builder); | |
| if (static_cast<size_t>(struct_builder->num_fields()) != projections.size()) { | |
| return InvalidArgument( | |
| "Inconsistent number of struct builder fields ({}) and projections ({})", | |
| struct_builder->num_fields(), projections.size()); | |
| } | |
| for (size_t i = 0; i < projections.size(); ++i) { | |
| auto& field_projection = projections[i]; | |
| auto* field_builder = struct_builder->field_builder(static_cast<int>(i)); | |
| if (field_projection.kind == FieldProjection::Kind::kDefault) { | |
| // Get-or-create the single Avro attributes container: another Avro attribute may | |
| // have created it already, so guard on `default_scalar` being unset rather than on | |
| // the container's presence (otherwise a pre-existing container would skip | |
| // preparation and silently fall back to per-row scalar rebuilding). | |
| std::shared_ptr<AvroExtraAttributes> attrs; | |
| if (field_projection.attributes == nullptr) { | |
| attrs = std::make_shared<AvroExtraAttributes>(); | |
| field_projection.attributes = attrs; | |
| } else { | |
| // Avro attaches only AvroExtraAttributes; checked_pointer_cast asserts that | |
| // invariant in debug rather than silently reinterpreting a foreign attributes | |
| // type. | |
| attrs = internal::checked_pointer_cast<AvroExtraAttributes>( | |
| field_projection.attributes); | |
| } | |
| if (attrs->default_scalar == nullptr) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| attrs->default_scalar, | |
| MakeDefaultScalar(std::get<Literal>(field_projection.from), | |
| field_builder->type())); | |
| } | |
| continue; | |
| } | |
| ICEBERG_RETURN_UNEXPECTED( | |
| PrepareNestedDefaultScalars(field_projection, field_builder)); | |
| } | |
| return {}; | |
| } | |
| } // namespace | |
| Status PrepareDefaultScalars(SchemaProjection& projection, | |
| ::arrow::ArrayBuilder* root_builder) { | |
| return PrepareStructDefaultScalars(projection.fields, root_builder); | |
| } | |
| Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder) { | |
| ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Scalar> scalar, | |
| MakeDefaultScalar(literal, builder->type())); | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendScalar(*scalar)); | |
| return {}; | |
| } | |
| Status AppendDefaultToBuilder(const FieldProjection& projection, | |
| ::arrow::ArrayBuilder* builder) { | |
| // Avro projections carry a single attributes type, so once one is attached it is an | |
| // AvroExtraAttributes; use checked_cast instead of a per-row dynamic_cast. | |
| if (projection.attributes != nullptr) { | |
| const auto& attrs = | |
| internal::checked_cast<const AvroExtraAttributes&>(*projection.attributes); | |
| if (attrs.default_scalar != nullptr) { | |
| ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendScalar(*attrs.default_scalar)); | |
| return {}; | |
| } | |
| } | |
| return AppendDefaultToBuilder(std::get<Literal>(projection.from), builder); | |
| } | |
| Status AppendDatumToBuilder(const ::avro::NodePtr& avro_node, | |
| const ::avro::GenericDatum& avro_datum, | |
| const SchemaProjection& projection, | |
| const Schema& projected_schema, | |
| const arrow::MetadataColumnContext& metadata_context, | |
| ::arrow::ArrayBuilder* array_builder) { | |
| return AppendNestedValueToBuilder(avro_node, avro_datum, projection.fields, | |
| projected_schema, metadata_context, array_builder); | |
| } | |
| namespace { | |
| // ToAvroNodeVisitor uses 0 for null branch and 1 for value branch. | |
| constexpr int64_t kNullBranch = 0; | |
| constexpr int64_t kValueBranch = 1; | |
| } // namespace | |
| Status EncodeDecimalToFixed(const ::arrow::Decimal128& value, size_t fixed_size, | |
| std::vector<uint8_t>* bytes) { | |
| ICEBERG_PRECHECK(fixed_size > 0, "Avro decimal fixed size must be positive"); | |
| // Arrow's bytes are native-endian; Avro fixed decimals are big-endian. | |
| auto big_endian = value.ToBytes(); | |
| if constexpr (std::endian::native == std::endian::little) { | |
| std::ranges::reverse(big_endian); | |
| } | |
| const uint8_t padding = (big_endian.front() & 0x80) == 0 ? 0x00 : 0xff; | |
| const auto value_size = std::min(fixed_size, big_endian.size()); | |
| const auto first = big_endian.end() - value_size; | |
| // Discard only sign extension, without changing the retained sign bit. | |
| ICEBERG_PRECHECK(std::all_of(big_endian.begin(), first, | |
| [padding](uint8_t byte) { return byte == padding; }) && | |
| ((*first ^ padding) & 0x80) == 0, | |
| "Decimal does not fit Avro fixed size {}", fixed_size); | |
| bytes->assign(fixed_size - value_size, padding); | |
| bytes->insert(bytes->end(), first, big_endian.end()); | |
| return {}; | |
| } | |
| Status ExtractDatumFromArray(const ::arrow::Array& array, int64_t index, | |
| ::avro::GenericDatum* datum) { | |
| if (index < 0 || index >= array.length()) { | |
| return InvalidArgument("Cannot extract datum from array at index {} of length {}", | |
| index, array.length()); | |
| } | |
| if (array.IsNull(index)) { | |
| if (datum->type() == ::avro::AVRO_NULL) { | |
| return {}; | |
| } | |
| if (!datum->isUnion()) [[unlikely]] { | |
| return InvalidSchema("Cannot extract null to non-union type: {}", | |
| ::avro::toString(datum->type())); | |
| } | |
| datum->selectBranch(kNullBranch); | |
| return {}; | |
| } | |
| if (datum->isUnion()) { | |
| datum->selectBranch(kValueBranch); | |
| } | |
| switch (array.type()->id()) { | |
| case ::arrow::Type::BOOL: { | |
| const auto& bool_array = | |
| internal::checked_cast<const ::arrow::BooleanArray&>(array); | |
| datum->value<bool>() = bool_array.Value(index); | |
| return {}; | |
| } | |
| case ::arrow::Type::INT32: { | |
| const auto& int32_array = internal::checked_cast<const ::arrow::Int32Array&>(array); | |
| datum->value<int32_t>() = int32_array.Value(index); | |
| return {}; | |
| } | |
| case ::arrow::Type::INT64: { | |
| const auto& int64_array = internal::checked_cast<const ::arrow::Int64Array&>(array); | |
| datum->value<int64_t>() = int64_array.Value(index); | |
| return {}; | |
| } | |
| case ::arrow::Type::FLOAT: { | |
| const auto& float_array = internal::checked_cast<const ::arrow::FloatArray&>(array); | |
| datum->value<float>() = float_array.Value(index); | |
| return {}; | |
| } | |
| case ::arrow::Type::DOUBLE: { | |
| const auto& double_array = | |
| internal::checked_cast<const ::arrow::DoubleArray&>(array); | |
| datum->value<double>() = double_array.Value(index); | |
| return {}; | |
| } | |
| // TODO(gangwu): support LARGE_STRING. | |
| case ::arrow::Type::STRING: { | |
| const auto& string_array = | |
| internal::checked_cast<const ::arrow::StringArray&>(array); | |
| datum->value<std::string>() = string_array.GetString(index); | |
| return {}; | |
| } | |
| // TODO(gangwu): support LARGE_BINARY. | |
| case ::arrow::Type::BINARY: { | |
| const auto& binary_array = | |
| internal::checked_cast<const ::arrow::BinaryArray&>(array); | |
| std::string_view value = binary_array.GetView(index); | |
| datum->value<std::vector<uint8_t>>().assign( | |
| reinterpret_cast<const uint8_t*>(value.data()), | |
| reinterpret_cast<const uint8_t*>(value.data()) + value.size()); | |
| return {}; | |
| } | |
| case ::arrow::Type::FIXED_SIZE_BINARY: { | |
| const auto& fixed_array = | |
| internal::checked_cast<const ::arrow::FixedSizeBinaryArray&>(array); | |
| std::string_view value = fixed_array.GetView(index); | |
| auto& fixed_datum = datum->value<::avro::GenericFixed>(); | |
| fixed_datum.value().assign(value.begin(), value.end()); | |
| return {}; | |
| } | |
| case ::arrow::Type::DECIMAL128: { | |
| const auto& decimal_array = | |
| internal::checked_cast<const ::arrow::Decimal128Array&>(array); | |
| auto& fixed_datum = datum->value<::avro::GenericFixed>(); | |
| return EncodeDecimalToFixed(::arrow::Decimal128(decimal_array.GetValue(index)), | |
| fixed_datum.schema()->fixedSize(), | |
| &fixed_datum.value()); | |
| } | |
| case ::arrow::Type::DATE32: { | |
| const auto& date_array = internal::checked_cast<const ::arrow::Date32Array&>(array); | |
| datum->value<int32_t>() = date_array.Value(index); | |
| return {}; | |
| } | |
| case ::arrow::Type::TIME64: { | |
| const auto& time_array = internal::checked_cast<const ::arrow::Time64Array&>(array); | |
| datum->value<int64_t>() = time_array.Value(index); | |
| return {}; | |
| } | |
| // For both timestamp and timestamp_tz with time unit as microsecond. | |
| case ::arrow::Type::TIMESTAMP: { | |
| const auto& timestamp_array = | |
| internal::checked_cast<const ::arrow::TimestampArray&>(array); | |
| datum->value<int64_t>() = timestamp_array.Value(index); | |
| return {}; | |
| } | |
| case ::arrow::Type::EXTENSION: { | |
| const auto& extension_type = | |
| internal::checked_cast<const ::arrow::ExtensionType&>(*array.type()); | |
| if (extension_type.extension_name() == kArrowUuidExtensionName) { | |
| const auto& extension_array = | |
| internal::checked_cast<const ::arrow::ExtensionArray&>(array); | |
| const auto& fixed_array = | |
| internal::checked_cast<const ::arrow::FixedSizeBinaryArray&>( | |
| *extension_array.storage()); | |
| std::string_view value = fixed_array.GetView(index); | |
| auto& fixed_datum = datum->value<::avro::GenericFixed>(); | |
| fixed_datum.value().assign(value.begin(), value.end()); | |
| return {}; | |
| } | |
| return NotSupported("Unsupported Arrow extension type: {}", | |
| extension_type.extension_name()); | |
| } | |
| case ::arrow::Type::STRUCT: { | |
| const auto& struct_array = | |
| internal::checked_cast<const ::arrow::StructArray&>(array); | |
| auto& record = datum->value<::avro::GenericRecord>(); | |
| for (int i = 0; i < struct_array.num_fields(); ++i) { | |
| ICEBERG_RETURN_UNEXPECTED( | |
| ExtractDatumFromArray(*struct_array.field(i), index, &record.fieldAt(i))); | |
| } | |
| return {}; | |
| } | |
| // TODO(gangwu): support LARGE_LIST. | |
| case ::arrow::Type::LIST: { | |
| const auto& list_array = internal::checked_cast<const ::arrow::ListArray&>(array); | |
| auto& avro_array = datum->value<::avro::GenericArray>(); | |
| auto& elements = avro_array.value(); | |
| auto start = list_array.value_offset(index); | |
| auto end = list_array.value_offset(index + 1); | |
| auto length = end - start; | |
| auto values = list_array.values(); | |
| elements.resize(length, ::avro::GenericDatum(avro_array.schema()->leafAt(0))); | |
| for (int64_t i = 0; i < length; ++i) { | |
| ICEBERG_RETURN_UNEXPECTED( | |
| ExtractDatumFromArray(*values, start + i, &elements[i])); | |
| } | |
| return {}; | |
| } | |
| case ::arrow::Type::MAP: { | |
| const auto& map_array = internal::checked_cast<const ::arrow::MapArray&>(array); | |
| auto start = map_array.value_offset(index); | |
| auto end = map_array.value_offset(index + 1); | |
| auto length = end - start; | |
| auto keys = map_array.keys(); | |
| auto items = map_array.items(); | |
| if (datum->type() == ::avro::AVRO_MAP) { | |
| // Handle regular Avro map | |
| auto& avro_map = datum->value<::avro::GenericMap>(); | |
| auto value_node = avro_map.schema()->leafAt(1); | |
| auto& map_entries = avro_map.value(); | |
| map_entries.resize( | |
| length, std::make_pair(std::string(), ::avro::GenericDatum(value_node))); | |
| const auto& key_array = | |
| internal::checked_cast<const ::arrow::StringArray&>(*keys); | |
| for (int64_t i = 0; i < length; ++i) { | |
| auto& map_entry = map_entries[i]; | |
| map_entry.first = key_array.GetString(start + i); | |
| ICEBERG_RETURN_UNEXPECTED( | |
| ExtractDatumFromArray(*items, start + i, &map_entry.second)); | |
| } | |
| } else if (datum->type() == ::avro::AVRO_ARRAY) { | |
| // Handle array-based map (list<struct<key, value>>) | |
| auto& avro_array = datum->value<::avro::GenericArray>(); | |
| auto record_node = avro_array.schema()->leafAt(0); | |
| if (record_node->type() != ::avro::AVRO_RECORD || record_node->leaves() != 2) { | |
| return InvalidArgument( | |
| "Expected Avro record with 2 fields for map value, got: {}", | |
| ToString(record_node)); | |
| } | |
| auto& elements = avro_array.value(); | |
| elements.resize(length, ::avro::GenericDatum(record_node)); | |
| for (int64_t i = 0; i < length; ++i) { | |
| auto& record = elements[i].value<::avro::GenericRecord>(); | |
| ICEBERG_RETURN_UNEXPECTED( | |
| ExtractDatumFromArray(*keys, start + i, &record.fieldAt(0))); | |
| ICEBERG_RETURN_UNEXPECTED( | |
| ExtractDatumFromArray(*items, start + i, &record.fieldAt(1))); | |
| } | |
| } else { | |
| return InvalidArgument("Unsupported Avro type for map: {}", | |
| static_cast<int>(datum->type())); | |
| } | |
| return {}; | |
| } | |
| default: | |
| return InvalidArgument("Unsupported Arrow array type: {}", | |
| array.type()->ToString()); | |
| } | |
| } | |
| } // namespace iceberg::avro |