| /* | |
| * 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/core/utils/nested_projection_utils.h" | |
| #include <set> | |
| #include <string> | |
| #include <unordered_set> | |
| #include <utility> | |
| #include <vector> | |
| #include "arrow/array/array_nested.h" | |
| #include "arrow/array/array_primitive.h" | |
| #include "arrow/array/builder_primitive.h" | |
| #include "arrow/array/concatenate.h" | |
| #include "arrow/array/util.h" | |
| #include "arrow/compute/cast.h" | |
| #include "arrow/type.h" | |
| #include "fmt/format.h" | |
| #include "paimon/common/data/blob_utils.h" | |
| #include "paimon/common/data/variant/variant_access_utils.h" | |
| #include "paimon/common/data/variant/variant_type_utils.h" | |
| #include "paimon/common/utils/arrow/vector_utils.h" | |
| #include "paimon/common/utils/checked_cast.h" | |
| #include "paimon/common/utils/string_utils.h" | |
| #include "paimon/core/casting/casting_utils.h" | |
| #include "paimon/status.h" | |
| namespace paimon { | |
| std::shared_ptr<arrow::Field> NestedProjectionUtils::FindFieldByName( | |
| const arrow::FieldVector& fields, const std::string& name) { | |
| for (const auto& field : fields) { | |
| if (field->name() == name) { | |
| return field; | |
| } | |
| } | |
| return nullptr; | |
| } | |
| Result<int32_t> NestedProjectionUtils::GetPaimonFieldId( | |
| const std::shared_ptr<arrow::Field>& field) { | |
| if (!field->HasMetadata() || !field->metadata()) { | |
| return Status::Invalid(fmt::format( | |
| "GetPaimonFieldId failed, do not exist metadata in field {}", field->name())); | |
| } | |
| auto result = field->metadata()->Get(DataField::FIELD_ID); | |
| if (!result.ok()) { | |
| return Status::Invalid( | |
| fmt::format("GetPaimonFieldId failed, cannot find field_id in metadata in field {}", | |
| field->name())); | |
| } | |
| std::optional<int32_t> field_id = StringUtils::StringToValue<int32_t>(result.ValueUnsafe()); | |
| if (!field_id) { | |
| return Status::Invalid( | |
| fmt::format("GetPaimonFieldId failed, cannot find convert field_id {} to int32", | |
| result.ValueUnsafe())); | |
| } | |
| return field_id.value(); | |
| } | |
| Result<std::shared_ptr<arrow::Field>> NestedProjectionUtils::FindFieldByPaimonId( | |
| const std::shared_ptr<arrow::DataType>& struct_type, int32_t field_id) { | |
| for (const auto& child : struct_type->fields()) { | |
| PAIMON_ASSIGN_OR_RAISE(int32_t paimon_field_id, GetPaimonFieldId(child)); | |
| if (paimon_field_id == field_id) { | |
| return child; | |
| } | |
| } | |
| return Status::Invalid( | |
| fmt::format("cannot find field {} in struct type {}", field_id, struct_type->ToString())); | |
| } | |
| Result<bool> NestedProjectionUtils::HasNestedSubfieldProjectionType( | |
| const std::shared_ptr<arrow::DataType>& file_type, | |
| const std::shared_ptr<arrow::DataType>& read_type) { | |
| switch (file_type->id()) { | |
| case arrow::Type::STRUCT: { | |
| if (VariantAccessUtils::IsVariantAccessType(read_type)) { | |
| // A variant-access projection is resolved by the variant read plans, not by | |
| // nested subfield projection. | |
| return false; | |
| } | |
| if (read_type->id() != arrow::Type::STRUCT) { | |
| return Status::Invalid(fmt::format( | |
| "HasNestedSubfieldProjectionType requires same nested type kind, but file " | |
| "type is {} and read type is {}", | |
| file_type->ToString(), read_type->ToString())); | |
| } | |
| auto file_struct = checked_pointer_cast<arrow::StructType>(file_type); | |
| auto read_struct = checked_pointer_cast<arrow::StructType>(read_type); | |
| bool field_count_diff = read_struct->num_fields() != file_struct->num_fields(); | |
| for (const auto& read_child : read_struct->fields()) { | |
| auto file_child = FindFieldByName(file_struct->fields(), read_child->name()); | |
| if (!file_child) { | |
| return Status::Invalid(fmt::format( | |
| "HasNestedSubfieldProjectionType found requested struct child '{}' " | |
| "missing in file type {}", | |
| read_child->name(), file_type->ToString())); | |
| } | |
| PAIMON_ASSIGN_OR_RAISE( | |
| bool child_has_nested_projection, | |
| HasNestedSubfieldProjectionType(file_child->type(), read_child->type())); | |
| if (child_has_nested_projection) { | |
| return true; | |
| } | |
| } | |
| return field_count_diff; | |
| } | |
| case arrow::Type::LIST: { | |
| if (read_type->id() != arrow::Type::LIST) { | |
| return Status::Invalid(fmt::format( | |
| "HasNestedSubfieldProjectionType requires same nested type kind, but file " | |
| "type is {} and read type is {}", | |
| file_type->ToString(), read_type->ToString())); | |
| } | |
| auto file_list = checked_pointer_cast<arrow::ListType>(file_type); | |
| auto read_list = checked_pointer_cast<arrow::ListType>(read_type); | |
| return HasNestedSubfieldProjectionType(file_list->value_type(), | |
| read_list->value_type()); | |
| } | |
| case arrow::Type::MAP: { | |
| if (read_type->id() != arrow::Type::MAP) { | |
| return Status::Invalid(fmt::format( | |
| "HasNestedSubfieldProjectionType requires same nested type kind, but file " | |
| "type is {} and read type is {}", | |
| file_type->ToString(), read_type->ToString())); | |
| } | |
| auto file_map = checked_pointer_cast<arrow::MapType>(file_type); | |
| auto read_map = checked_pointer_cast<arrow::MapType>(read_type); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| bool key_has_nested_projection, | |
| HasNestedSubfieldProjectionType(file_map->key_type(), read_map->key_type())); | |
| if (key_has_nested_projection) { | |
| return true; | |
| } | |
| return HasNestedSubfieldProjectionType(file_map->item_type(), read_map->item_type()); | |
| } | |
| default: | |
| return false; | |
| } | |
| } | |
| namespace { | |
| // Structural equality that also compares paimon field IDs on STRUCT children, so a | |
| // drop+add of a same-name/same-type field (new ID) is not treated as a no-op. | |
| Result<bool> EqualWithFieldIds(const std::shared_ptr<arrow::DataType>& a, | |
| const std::shared_ptr<arrow::DataType>& b) { | |
| if (a->id() != b->id() || a->num_fields() != b->num_fields()) { | |
| return false; | |
| } | |
| if (a->id() == arrow::Type::FIXED_SIZE_LIST) { | |
| const auto& vector_a = checked_cast<const arrow::FixedSizeListType&>(*a); | |
| const auto& vector_b = checked_cast<const arrow::FixedSizeListType&>(*b); | |
| if (vector_a.list_size() != vector_b.list_size()) { | |
| return false; | |
| } | |
| } | |
| if (a->num_fields() == 0) { | |
| return a->Equals(*b); | |
| } | |
| for (int32_t i = 0; i < a->num_fields(); ++i) { | |
| const auto& fa = a->field(i); | |
| const auto& fb = b->field(i); | |
| if (fa->nullable() != fb->nullable()) { | |
| return false; | |
| } | |
| if (a->id() == arrow::Type::STRUCT) { | |
| if (fa->name() != fb->name()) { | |
| return false; | |
| } | |
| // Compare IDs only when present (a map entry's key/value carry none). | |
| auto id_a = NestedProjectionUtils::GetPaimonFieldId(fa); | |
| auto id_b = NestedProjectionUtils::GetPaimonFieldId(fb); | |
| if (id_a.ok() && id_b.ok() && id_a.value() != id_b.value()) { | |
| return false; | |
| } | |
| } | |
| PAIMON_ASSIGN_OR_RAISE(bool child_equal, EqualWithFieldIds(fa->type(), fb->type())); | |
| if (!child_equal) { | |
| return false; | |
| } | |
| } | |
| return true; | |
| } | |
| /// Whether `read_type` is `data_type` with variant columns replaced by their variant-access | |
| /// projections and nothing else changed (matching paimon field IDs). | |
| Result<bool> IsVariantAccessSubstitution(const std::shared_ptr<arrow::DataType>& read_type, | |
| const std::shared_ptr<arrow::DataType>& data_type) { | |
| PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorEvolution(read_type, data_type)); | |
| PAIMON_ASSIGN_OR_RAISE(bool equal, EqualWithFieldIds(read_type, data_type)); | |
| if (equal) { | |
| return true; | |
| } | |
| if (VariantAccessUtils::IsVariantAccessType(read_type) && | |
| VariantTypeUtils::IsUnshreddedVariantType(data_type)) { | |
| return true; | |
| } | |
| // Any other difference in shape, including a dropped field, is a real projection. | |
| if (read_type->id() != data_type->id() || read_type->num_fields() != data_type->num_fields()) { | |
| return false; | |
| } | |
| for (int32_t i = 0; i < read_type->num_fields(); ++i) { | |
| const std::shared_ptr<arrow::Field>& read_child = read_type->field(i); | |
| const std::shared_ptr<arrow::Field>& data_child = data_type->field(i); | |
| // LIST and MAP name their children by format convention, so only STRUCT is | |
| // matched by name and field ID. | |
| if (read_type->id() == arrow::Type::STRUCT) { | |
| if (read_child->name() != data_child->name()) { | |
| return false; | |
| } | |
| auto id_r = NestedProjectionUtils::GetPaimonFieldId(read_child); | |
| auto id_d = NestedProjectionUtils::GetPaimonFieldId(data_child); | |
| if (id_r.ok() && id_d.ok() && id_r.value() != id_d.value()) { | |
| return false; | |
| } | |
| } | |
| PAIMON_ASSIGN_OR_RAISE(bool sub, | |
| IsVariantAccessSubstitution(read_child->type(), data_child->type())); | |
| if (!sub) { | |
| return false; | |
| } | |
| } | |
| return true; | |
| } | |
| // Reconcile a LIST/MAP item: read may ADD fields (evolution, null-filled | |
| // downstream) but must not DROP one. Returns the file-readable item type; | |
| // `container` names the container ("list"/"map") for the error message. | |
| Result<std::shared_ptr<arrow::DataType>> PruneRepeatedItemType( | |
| const std::shared_ptr<arrow::DataType>& read_type, | |
| const std::shared_ptr<arrow::DataType>& data_type, const char* container) { | |
| PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorEvolution(read_type, data_type)); | |
| PAIMON_ASSIGN_OR_RAISE(bool same, EqualWithFieldIds(read_type, data_type)); | |
| if (same) { | |
| return data_type; | |
| } | |
| PAIMON_ASSIGN_OR_RAISE(bool substitution, IsVariantAccessSubstitution(read_type, data_type)); | |
| if (substitution) { | |
| return read_type; | |
| } | |
| if (read_type->id() != data_type->id()) { | |
| return Status::Invalid( | |
| fmt::format("PruneDataType nested item type mismatch inside {}: read {} vs data {}", | |
| container, read_type->ToString(), data_type->ToString())); | |
| } | |
| switch (data_type->id()) { | |
| case arrow::Type::STRUCT: { | |
| arrow::FieldVector item_fields; | |
| for (const auto& data_child : data_type->fields()) { | |
| PAIMON_ASSIGN_OR_RAISE(int32_t data_child_id, | |
| NestedProjectionUtils::GetPaimonFieldId(data_child)); | |
| std::shared_ptr<arrow::Field> read_child; | |
| for (const auto& candidate : read_type->fields()) { | |
| PAIMON_ASSIGN_OR_RAISE(int32_t candidate_id, | |
| NestedProjectionUtils::GetPaimonFieldId(candidate)); | |
| if (candidate_id == data_child_id) { | |
| read_child = candidate; | |
| break; | |
| } | |
| } | |
| if (!read_child) { | |
| // A file field is dropped -- a real partial projection. | |
| return Status::Invalid(fmt::format( | |
| "PruneDataType does not support partial projection inside {}: src {} vs " | |
| "target {}", | |
| container, data_type->ToString(), read_type->ToString())); | |
| } | |
| if (read_child->name() != data_child->name()) { | |
| return Status::Invalid(fmt::format( | |
| "PruneDataType does not support renaming inside {}: field id {} read '{}' " | |
| "vs data '{}'", | |
| container, data_child_id, read_child->name(), data_child->name())); | |
| } | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<arrow::DataType> item_child_type, | |
| PruneRepeatedItemType(read_child->type(), data_child->type(), container)); | |
| item_fields.push_back(data_child->WithType(item_child_type)); | |
| } | |
| return arrow::struct_(item_fields); | |
| } | |
| case arrow::Type::LIST: { | |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::DataType> item, | |
| PruneRepeatedItemType(read_type->field(0)->type(), | |
| data_type->field(0)->type(), container)); | |
| return arrow::list(data_type->field(0)->WithType(item)); | |
| } | |
| case arrow::Type::MAP: { | |
| auto read_map = checked_pointer_cast<arrow::MapType>(read_type); | |
| auto data_map = checked_pointer_cast<arrow::MapType>(data_type); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<arrow::DataType> key, | |
| PruneRepeatedItemType(read_map->key_type(), data_map->key_type(), container)); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<arrow::DataType> item, | |
| PruneRepeatedItemType(read_map->item_type(), data_map->item_type(), container)); | |
| return checked_pointer_cast<arrow::DataType>(std::make_shared<arrow::MapType>( | |
| data_map->key_field()->WithType(key), data_map->item_field()->WithType(item), | |
| data_map->keys_sorted())); | |
| } | |
| default: | |
| return data_type; | |
| } | |
| } | |
| } // namespace | |
| Result<std::optional<std::shared_ptr<arrow::DataType>>> NestedProjectionUtils::PruneDataType( | |
| const std::shared_ptr<arrow::DataType>& read_type, | |
| const std::shared_ptr<arrow::DataType>& data_type) { | |
| PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorEvolution(read_type, data_type)); | |
| // Identical types (including paimon field IDs) need no pruning. | |
| PAIMON_ASSIGN_OR_RAISE(bool same, EqualWithFieldIds(read_type, data_type)); | |
| if (same) { | |
| return std::optional<std::shared_ptr<arrow::DataType>>(data_type); | |
| } | |
| switch (read_type->id()) { | |
| case arrow::Type::STRUCT: { | |
| if (VariantAccessUtils::IsVariantAccessType(read_type) && | |
| VariantTypeUtils::IsUnshreddedVariantType(data_type)) { | |
| // A variant-access projection replaces the variant column type; pass it through | |
| // so the read path extracts the described paths. | |
| return std::optional<std::shared_ptr<arrow::DataType>>(read_type); | |
| } | |
| arrow::FieldVector pruned_fields; | |
| for (const auto& read_child : read_type->fields()) { | |
| PAIMON_ASSIGN_OR_RAISE(int32_t read_child_id, GetPaimonFieldId(read_child)); | |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Field> data_child, | |
| FindFieldByPaimonId(data_type, read_child_id)); | |
| if (read_child->name() != data_child->name()) { | |
| return Status::Invalid(fmt::format( | |
| "PruneDataType does not support schema evolution inside struct: nested " | |
| "field id {} name mismatch: read '{}' vs data '{}'", | |
| read_child_id, read_child->name(), data_child->name())); | |
| } | |
| if (read_child->type()->id() != data_child->type()->id()) { | |
| return Status::Invalid(fmt::format( | |
| "PruneDataType nested field type mismatch for '{}': read {} vs data {}", | |
| read_child->name(), read_child->type()->ToString(), | |
| data_child->type()->ToString())); | |
| } | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::optional<std::shared_ptr<arrow::DataType>> pruned_child_type, | |
| PruneDataType(read_child->type(), data_child->type())); | |
| if (!pruned_child_type.has_value()) { | |
| // All sub-fields of this child were pruned away; skip it. | |
| continue; | |
| } | |
| pruned_fields.push_back(data_child->WithType(pruned_child_type.value())); | |
| } | |
| if (pruned_fields.empty()) { | |
| // All fields pruned — return nullopt so the caller can skip this field. | |
| return std::optional<std::shared_ptr<arrow::DataType>>(std::nullopt); | |
| } | |
| return std::optional<std::shared_ptr<arrow::DataType>>(arrow::struct_(pruned_fields)); | |
| } | |
| case arrow::Type::LIST: { | |
| PAIMON_ASSIGN_OR_RAISE(bool list_substitution, | |
| IsVariantAccessSubstitution(read_type, data_type)); | |
| if (list_substitution) { | |
| return std::optional<std::shared_ptr<arrow::DataType>>(read_type); | |
| } | |
| // Added fields (schema evolution) are allowed; dropped fields still fail. | |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::DataType> item, | |
| PruneRepeatedItemType(read_type->field(0)->type(), | |
| data_type->field(0)->type(), "list")); | |
| return std::optional<std::shared_ptr<arrow::DataType>>( | |
| arrow::list(data_type->field(0)->WithType(item))); | |
| } | |
| case arrow::Type::MAP: { | |
| PAIMON_ASSIGN_OR_RAISE(bool map_substitution, | |
| IsVariantAccessSubstitution(read_type, data_type)); | |
| if (map_substitution) { | |
| return std::optional<std::shared_ptr<arrow::DataType>>(read_type); | |
| } | |
| auto read_map = checked_pointer_cast<arrow::MapType>(read_type); | |
| auto data_map = checked_pointer_cast<arrow::MapType>(data_type); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<arrow::DataType> key, | |
| PruneRepeatedItemType(read_map->key_type(), data_map->key_type(), "map")); | |
| PAIMON_ASSIGN_OR_RAISE( | |
| std::shared_ptr<arrow::DataType> item, | |
| PruneRepeatedItemType(read_map->item_type(), data_map->item_type(), "map")); | |
| return std::optional<std::shared_ptr<arrow::DataType>>(std::make_shared<arrow::MapType>( | |
| data_map->key_field()->WithType(key), data_map->item_field()->WithType(item), | |
| data_map->keys_sorted())); | |
| } | |
| default: | |
| // Atomic type: return data_type as-is (type evolution is handled | |
| // separately by CastExecutor). | |
| return std::optional<std::shared_ptr<arrow::DataType>>(data_type); | |
| } | |
| } | |
| Result<bool> NestedProjectionUtils::HasNestedSubfieldProjection( | |
| const std::shared_ptr<arrow::Schema>& file_schema, | |
| const std::shared_ptr<arrow::Schema>& read_schema) { | |
| for (const auto& read_field : read_schema->fields()) { | |
| auto file_field = file_schema->GetFieldByName(read_field->name()); | |
| if (!file_field) { | |
| return Status::Invalid(fmt::format( | |
| "HasNestedSubfieldProjection found read field '{}' missing in file schema {}", | |
| read_field->name(), file_schema->ToString())); | |
| } | |
| if (read_field->type()->id() == arrow::Type::STRUCT || | |
| read_field->type()->id() == arrow::Type::LIST || | |
| read_field->type()->id() == arrow::Type::MAP) { | |
| PAIMON_ASSIGN_OR_RAISE( | |
| bool has_nested_projection, | |
| HasNestedSubfieldProjectionType(file_field->type(), read_field->type())); | |
| if (has_nested_projection) { | |
| return true; | |
| } | |
| } | |
| } | |
| return false; | |
| } | |
| // Map selected-keys support | |
| Result<std::vector<std::string>> NestedProjectionUtils::GetMapSelectedKeys( | |
| const std::shared_ptr<arrow::Field>& field) { | |
| std::vector<std::string> result; | |
| if (!field->HasMetadata() || !field->metadata()) { | |
| return result; | |
| } | |
| auto get_result = field->metadata()->Get(DataField::MAP_SELECTED_KEYS); | |
| if (!get_result.ok()) { | |
| return result; | |
| } | |
| if (BlobUtils::IsMapBlobField(field)) { | |
| return Status::NotImplemented( | |
| "paimon.map.selected-keys is not supported for MAP<..., BLOB>"); | |
| } | |
| auto tokens = StringUtils::Split(get_result.ValueUnsafe(), ",", /*ignore_empty=*/false); | |
| std::unordered_set<std::string> deduplicated; | |
| deduplicated.reserve(tokens.size()); | |
| for (const auto& token : tokens) { | |
| if (!deduplicated.insert(token).second) { | |
| return Status::Invalid(fmt::format("Duplicate selected key '{}' in {} metadata", token, | |
| DataField::MAP_SELECTED_KEYS)); | |
| } | |
| result.push_back(token); | |
| } | |
| return result; | |
| } | |
| bool NestedProjectionUtils::IsMapSharedShreddingAccessField( | |
| const std::shared_ptr<arrow::Field>& field) { | |
| if (field->type()->id() != arrow::Type::STRUCT || !field->HasMetadata() || !field->metadata()) { | |
| return false; | |
| } | |
| return field->metadata()->Contains(DataField::MAP_SELECTED_KEYS); | |
| } | |
| Result<std::vector<std::string>> NestedProjectionUtils::ValidateMapSharedShreddingAccessField( | |
| const std::shared_ptr<arrow::Field>& field) { | |
| if (field->type()->id() != arrow::Type::STRUCT) { | |
| return Status::Invalid( | |
| fmt::format("selected-key MAP field {} is not a STRUCT", field->name())); | |
| } | |
| auto struct_type = checked_pointer_cast<arrow::StructType>(field->type()); | |
| PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> selected_keys, GetMapSelectedKeys(field)); | |
| if (struct_type->num_fields() == 0 || | |
| selected_keys.size() != static_cast<size_t>(struct_type->num_fields())) { | |
| return Status::Invalid( | |
| fmt::format("selected-key metadata size {} does not match STRUCT field count {} for {}", | |
| selected_keys.size(), struct_type->num_fields(), field->name())); | |
| } | |
| const auto& value_type = struct_type->field(0)->type(); | |
| for (int32_t i = 1; i < struct_type->num_fields(); ++i) { | |
| if (!struct_type->field(i)->type()->Equals(value_type)) { | |
| return Status::Invalid(fmt::format( | |
| "selected-key MAP fields must have the same value type, but {} and {} differ", | |
| value_type->ToString(), struct_type->field(i)->type()->ToString())); | |
| } | |
| } | |
| return selected_keys; | |
| } | |
| Result<std::shared_ptr<arrow::DataType>> | |
| NestedProjectionUtils::BuildMapSharedShreddingAccessDataType( | |
| const std::shared_ptr<arrow::Field>& read_field, | |
| const std::shared_ptr<arrow::DataType>& data_type) { | |
| if (!IsMapSharedShreddingAccessField(read_field)) { | |
| return Status::Invalid( | |
| fmt::format("field {} is not a selected-key MAP projection", read_field->name())); | |
| } | |
| if (data_type->id() != arrow::Type::MAP) { | |
| return Status::Invalid( | |
| fmt::format("selected-key MAP projection {} requires MAP data type, got {}", | |
| read_field->name(), data_type->ToString())); | |
| } | |
| PAIMON_RETURN_NOT_OK(ValidateMapSharedShreddingAccessField(read_field).status()); | |
| auto read_struct = checked_pointer_cast<arrow::StructType>(read_field->type()); | |
| auto data_map = checked_pointer_cast<arrow::MapType>(data_type); | |
| arrow::FieldVector data_children; | |
| data_children.reserve(read_struct->num_fields()); | |
| for (const auto& read_child : read_struct->fields()) { | |
| data_children.push_back(read_child->WithType(data_map->item_type())); | |
| } | |
| return arrow::struct_(std::move(data_children)); | |
| } | |
| Result<std::string_view> NestedProjectionUtils::GetMapKeyViewAt( | |
| const std::shared_ptr<arrow::Array>& key_array, int64_t entry_idx) { | |
| if (key_array->IsNull(entry_idx)) { | |
| return Status::Invalid("selected-key MAP read found null MAP key at entry " + | |
| std::to_string(entry_idx)); | |
| } | |
| if (key_array->type_id() == arrow::Type::STRING) { | |
| return checked_pointer_cast<arrow::StringArray>(key_array)->GetView(entry_idx); | |
| } | |
| if (key_array->type_id() == arrow::Type::DICTIONARY) { | |
| auto dict_type = checked_pointer_cast<arrow::DictionaryType>(key_array->type()); | |
| if (dict_type->value_type()->id() != arrow::Type::STRING && | |
| dict_type->value_type()->id() != arrow::Type::LARGE_STRING) { | |
| return Status::Invalid( | |
| fmt::format("selected-key MAP read only supports string keys or " | |
| "dictionary<string|large_string> keys, got {}", | |
| key_array->type()->ToString())); | |
| } | |
| auto dict_keys = checked_pointer_cast<arrow::DictionaryArray>(key_array); | |
| int64_t dict_idx = dict_keys->GetValueIndex(entry_idx); | |
| const auto& dictionary = dict_keys->dictionary(); | |
| if (dictionary->IsNull(dict_idx)) { | |
| return Status::Invalid( | |
| "selected-key MAP read found null dictionary MAP key at dictionary index " + | |
| std::to_string(dict_idx)); | |
| } | |
| if (dict_type->value_type()->id() == arrow::Type::STRING) { | |
| return checked_pointer_cast<arrow::StringArray>(dictionary)->GetView(dict_idx); | |
| } | |
| return checked_pointer_cast<arrow::LargeStringArray>(dictionary)->GetView(dict_idx); | |
| } | |
| return Status::Invalid( | |
| fmt::format("selected-key MAP read only supports string keys or " | |
| "dictionary<string|large_string> keys, got {}", | |
| key_array->type()->ToString())); | |
| } | |
| Result<std::shared_ptr<arrow::Array>> NestedProjectionUtils::FilterMapArrayBySelectedKeys( | |
| const std::shared_ptr<arrow::Array>& array, const std::vector<std::string>& selected_keys, | |
| arrow::MemoryPool* pool) { | |
| if (selected_keys.empty() || !array || array->length() == 0) { | |
| return array; | |
| } | |
| if (pool == nullptr) { | |
| return Status::Invalid("FilterMapArrayBySelectedKeys requires a non-null memory pool"); | |
| } | |
| if (array->type_id() != arrow::Type::MAP) { | |
| return Status::Invalid(fmt::format( | |
| "FilterMapArrayBySelectedKeys requires map array, got {}", array->type()->ToString())); | |
| } | |
| auto map_array = checked_pointer_cast<arrow::MapArray>(array); | |
| auto map_type = checked_pointer_cast<arrow::MapType>(array->type()); | |
| auto key_array = map_array->keys(); | |
| auto values_array = map_array->items(); | |
| int64_t num_maps = map_array->length(); | |
| std::unordered_set<std::string> deduplicated; | |
| deduplicated.reserve(selected_keys.size()); | |
| for (const auto& selected_key : selected_keys) { | |
| if (!deduplicated.insert(selected_key).second) { | |
| return Status::Invalid(fmt::format("Duplicate selected key '{}' in {} metadata", | |
| selected_key, DataField::MAP_SELECTED_KEYS)); | |
| } | |
| } | |
| PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::unique_ptr<arrow::ArrayBuilder> key_builder_u, | |
| arrow::MakeBuilder(arrow::utf8(), pool)); | |
| PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::unique_ptr<arrow::ArrayBuilder> value_builder_u, | |
| arrow::MakeBuilder(values_array->type(), pool)); | |
| arrow::MapBuilder map_builder(pool, std::move(key_builder_u), std::move(value_builder_u)); | |
| auto* key_builder = checked_cast<arrow::StringBuilder*>(map_builder.key_builder()); | |
| auto* value_builder = map_builder.item_builder(); | |
| PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.Reserve(num_maps)); | |
| for (int64_t map_idx = 0; map_idx < num_maps; ++map_idx) { | |
| if (map_array->IsNull(map_idx)) { | |
| PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.AppendNull()); | |
| continue; | |
| } | |
| PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.Append()); | |
| int64_t start = map_array->value_offset(map_idx); | |
| int64_t end = map_array->value_offset(map_idx + 1); | |
| // Keep selected keys in the exact selected_keys order. | |
| for (const auto& selected_key : selected_keys) { | |
| for (int64_t entry_idx = start; entry_idx < end; ++entry_idx) { | |
| PAIMON_ASSIGN_OR_RAISE(std::string_view key_view, | |
| GetMapKeyViewAt(key_array, entry_idx)); | |
| if (key_view == selected_key) { | |
| PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->Append( | |
| key_view.data(), static_cast<int32_t>(key_view.size()))); | |
| PAIMON_RETURN_NOT_OK_FROM_ARROW( | |
| value_builder->AppendArraySlice(*values_array->data(), entry_idx, 1)); | |
| } | |
| } | |
| } | |
| } | |
| std::shared_ptr<arrow::Array> result_map; | |
| PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.Finish(&result_map)); | |
| return result_map; | |
| } | |
| Result<bool> NestedProjectionUtils::HasMapSelectedKeysRecursively( | |
| const std::shared_ptr<arrow::Field>& read_field) { | |
| if (!read_field) { | |
| return false; | |
| } | |
| if (IsMapSharedShreddingAccessField(read_field)) { | |
| PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> selected_keys, | |
| GetMapSelectedKeys(read_field)); | |
| auto read_struct = checked_pointer_cast<arrow::StructType>(read_field->type()); | |
| if (selected_keys.size() != static_cast<size_t>(read_struct->num_fields())) { | |
| return Status::Invalid(fmt::format( | |
| "selected-key metadata size {} does not match STRUCT field count {} for {}", | |
| selected_keys.size(), read_struct->num_fields(), read_field->name())); | |
| } | |
| return true; | |
| } | |
| if (read_field->type()->id() == arrow::Type::MAP) { | |
| PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> selected_keys, | |
| GetMapSelectedKeys(read_field)); | |
| return !selected_keys.empty(); | |
| } | |
| if (read_field->type()->id() == arrow::Type::STRUCT) { | |
| for (const auto& child : read_field->type()->fields()) { | |
| PAIMON_ASSIGN_OR_RAISE(bool has_selected_keys, HasMapSelectedKeysRecursively(child)); | |
| if (has_selected_keys) { | |
| return true; | |
| } | |
| } | |
| } | |
| return false; | |
| } | |
| Result<std::shared_ptr<arrow::Array>> | |
| NestedProjectionUtils::FilterMapArrayBySelectedKeysRecursively( | |
| const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::Field>& read_field, | |
| arrow::MemoryPool* pool) { | |
| if (!array || !read_field) { | |
| return array; | |
| } | |
| if (IsMapSharedShreddingAccessField(read_field)) { | |
| return array; | |
| } | |
| if (read_field->type()->id() == arrow::Type::MAP) { | |
| PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> selected_keys, | |
| GetMapSelectedKeys(read_field)); | |
| if (selected_keys.empty()) { | |
| return array; | |
| } | |
| return FilterMapArrayBySelectedKeys(array, selected_keys, pool); | |
| } | |
| if (read_field->type()->id() != arrow::Type::STRUCT) { | |
| return array; | |
| } | |
| if (array->type_id() != arrow::Type::STRUCT) { | |
| return Status::Invalid(fmt::format( | |
| "FilterMapArrayBySelectedKeysRecursively requires struct array for read field '{}', " | |
| "got {}", | |
| read_field->name(), array->type()->ToString())); | |
| } | |
| auto struct_array = checked_pointer_cast<arrow::StructArray>(array); | |
| auto read_struct_type = checked_pointer_cast<arrow::StructType>(read_field->type()); | |
| if (struct_array->num_fields() != read_struct_type->num_fields()) { | |
| return Status::Invalid(fmt::format( | |
| "FilterMapArrayBySelectedKeysRecursively struct field count mismatch for '{}': " | |
| "array {} vs read {}", | |
| read_field->name(), struct_array->num_fields(), read_struct_type->num_fields())); | |
| } | |
| std::vector<std::shared_ptr<arrow::ArrayData>> filtered_child_data; | |
| filtered_child_data.reserve(struct_array->num_fields()); | |
| for (int32_t i = 0; i < struct_array->num_fields(); ++i) { | |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> filtered_child, | |
| FilterMapArrayBySelectedKeysRecursively( | |
| struct_array->field(i), read_struct_type->field(i), pool)); | |
| filtered_child_data.push_back(filtered_child->data()); | |
| } | |
| auto filtered_struct_data = arrow::ArrayData::Make( | |
| read_struct_type, struct_array->length(), {struct_array->null_bitmap()}, | |
| std::move(filtered_child_data), struct_array->null_count(), struct_array->offset()); | |
| return arrow::MakeArray(std::move(filtered_struct_data)); | |
| } | |
| namespace { | |
| // Strips physical-only differences from a leaf type: ORC lazy decoding wraps | |
| // strings in a dictionary and may widen them to large_string. Unwrap dictionaries | |
| // for every value type, but preserve large_binary because it is blob's real type. | |
| // Two leaves with equal normalized types hold the same logical values. | |
| std::shared_ptr<arrow::DataType> NormalizeLeafRepresentation( | |
| const std::shared_ptr<arrow::DataType>& type) { | |
| auto t = type; | |
| if (t->id() == arrow::Type::DICTIONARY) { | |
| t = checked_pointer_cast<arrow::DictionaryType>(t)->value_type(); | |
| } | |
| if (t->id() == arrow::Type::LARGE_STRING) { | |
| return arrow::utf8(); | |
| } | |
| return t; | |
| } | |
| } // namespace | |
| Result<std::shared_ptr<arrow::Array>> NestedProjectionUtils::AlignArrayToReadType( | |
| const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::DataType>& read_type, | |
| arrow::MemoryPool* pool) { | |
| if (array->type()->Equals(read_type, /*check_metadata=*/true)) { | |
| return array; | |
| } | |
| // Produce exactly `read_type` so every file yields the same output type: rebuild | |
| // nested arrays with read-side fields/nullability and cast a leaf (decodes dict). | |
| const auto& data = array->data(); | |
| switch (read_type->id()) { | |
| case arrow::Type::STRUCT: { | |
| if (array->type()->id() != arrow::Type::STRUCT) { | |
| return Status::Invalid(fmt::format("AlignArrayToReadType cannot reconcile {} to {}", | |
| array->type()->ToString(), | |
| read_type->ToString())); | |
| } | |
| const auto& array_type = array->type(); | |
| std::vector<std::shared_ptr<arrow::ArrayData>> children; | |
| children.reserve(read_type->num_fields()); | |
| for (const auto& read_field : read_type->fields()) { | |
| // Match by name (parquet drops nested field-id metadata); if both | |
| // carry IDs they must agree, so a drop+add same-name field won't match. | |
| auto read_id = GetPaimonFieldId(read_field); | |
| int32_t match = -1; | |
| for (int32_t j = 0; j < array_type->num_fields(); j++) { | |
| const auto& array_field = array_type->field(j); | |
| if (array_field->name() != read_field->name()) { | |
| continue; | |
| } | |
| auto data_id = GetPaimonFieldId(array_field); | |
| if (read_id.ok() && data_id.ok() && read_id.value() != data_id.value()) { | |
| continue; | |
| } | |
| match = j; | |
| break; | |
| } | |
| if (match >= 0) { | |
| auto child = arrow::MakeArray(data->child_data[match]); | |
| PAIMON_ASSIGN_OR_RAISE(child, | |
| AlignArrayToReadType(child, read_field->type(), pool)); | |
| children.push_back(child->data()); | |
| } else { | |
| PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( | |
| std::shared_ptr<arrow::Array> null_child, | |
| arrow::MakeArrayOfNull(read_field->type(), data->offset + data->length, | |
| pool)); | |
| children.push_back(null_child->data()); | |
| } | |
| } | |
| auto new_data = data->Copy(); | |
| new_data->type = read_type; | |
| new_data->child_data = std::move(children); | |
| return arrow::MakeArray(new_data); | |
| } | |
| case arrow::Type::LIST: | |
| case arrow::Type::FIXED_SIZE_LIST: { | |
| if (array->type()->id() != read_type->id()) { | |
| return Status::Invalid(fmt::format("AlignArrayToReadType cannot reconcile {} to {}", | |
| array->type()->ToString(), | |
| read_type->ToString())); | |
| } | |
| if (read_type->id() == arrow::Type::FIXED_SIZE_LIST && | |
| checked_cast<const arrow::FixedSizeListType&>(*array->type()).list_size() != | |
| checked_cast<const arrow::FixedSizeListType&>(*read_type).list_size()) { | |
| return Status::Invalid("AlignArrayToReadType cannot change fixed-size list size"); | |
| } | |
| auto values = arrow::MakeArray(data->child_data[0]); | |
| PAIMON_ASSIGN_OR_RAISE(values, | |
| AlignArrayToReadType(values, read_type->field(0)->type(), pool)); | |
| auto new_data = data->Copy(); | |
| new_data->type = read_type; | |
| new_data->child_data = {values->data()}; | |
| return arrow::MakeArray(new_data); | |
| } | |
| case arrow::Type::MAP: { | |
| if (array->type()->id() != arrow::Type::MAP) { | |
| return Status::Invalid(fmt::format("AlignArrayToReadType cannot reconcile {} to {}", | |
| array->type()->ToString(), | |
| read_type->ToString())); | |
| } | |
| auto read_map = checked_pointer_cast<arrow::MapType>(read_type); | |
| const auto& entries_data = data->child_data[0]; | |
| auto key = arrow::MakeArray(entries_data->child_data[0]); | |
| auto value = arrow::MakeArray(entries_data->child_data[1]); | |
| PAIMON_ASSIGN_OR_RAISE(key, AlignArrayToReadType(key, read_map->key_type(), pool)); | |
| PAIMON_ASSIGN_OR_RAISE(value, AlignArrayToReadType(value, read_map->item_type(), pool)); | |
| auto new_entries = entries_data->Copy(); | |
| new_entries->type = arrow::struct_({read_map->key_field(), read_map->item_field()}); | |
| new_entries->child_data = {key->data(), value->data()}; | |
| auto new_data = data->Copy(); | |
| new_data->type = read_type; | |
| new_data->child_data = {new_entries}; | |
| return arrow::MakeArray(new_data); | |
| } | |
| default: { | |
| // Leaf: only physical-representation differences are valid here (ORC | |
| // dictionary encoding, string/binary offset width). Genuine type | |
| // evolution is handled by FieldMappingReader's cast executors and | |
| // rejected upstream in PruneDataType, so fail anything else. | |
| if (!NormalizeLeafRepresentation(array->type()) | |
| ->Equals(*NormalizeLeafRepresentation(read_type))) { | |
| return Status::Invalid( | |
| fmt::format("AlignArrayToReadType unsupported leaf type change: data {} vs " | |
| "read {}", | |
| array->type()->ToString(), read_type->ToString())); | |
| } | |
| return CastingUtils::Cast(array, read_type, arrow::compute::CastOptions::Safe(), pool); | |
| } | |
| } | |
| } | |
| } // namespace paimon |