diff --git a/src/paimon/common/reader/data_evolution_file_reader.cpp b/src/paimon/common/reader/data_evolution_file_reader.cpp index eefe900cc..a4e49ecbb 100644 --- a/src/paimon/common/reader/data_evolution_file_reader.cpp +++ b/src/paimon/common/reader/data_evolution_file_reader.cpp @@ -16,6 +16,8 @@ #include "paimon/common/reader/data_evolution_file_reader.h" +#include "arrow/array/array_nested.h" +#include "arrow/array/util.h" #include "arrow/c/abi.h" #include "arrow/c/bridge.h" #include "fmt/format.h" @@ -25,6 +27,7 @@ #include "paimon/common/utils/arrow/status_utils.h" namespace paimon { + Result> DataEvolutionFileReader::Create( std::vector>&& readers, const std::shared_ptr& read_schema, int32_t read_batch_size, @@ -82,6 +85,7 @@ Result DataEvolutionFileReader::NextBatchWithB } const auto& sub_array = array_for_each_reader[reader_offsets_[i]]; assert(sub_array->num_fields() > field_offsets_[i]); + // Each file is already aligned to its read schema by its FieldMappingReader. target_sub_array_vec.push_back(sub_array->field(field_offsets_[i])); } PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( diff --git a/src/paimon/core/io/field_mapping_reader.cpp b/src/paimon/core/io/field_mapping_reader.cpp index 3c225e7d6..efea20720 100644 --- a/src/paimon/core/io/field_mapping_reader.cpp +++ b/src/paimon/core/io/field_mapping_reader.cpp @@ -149,6 +149,11 @@ Result> FieldMappingReader::Create( if (mapping_reader->non_partition_info_.cast_executors[i] != nullptr) { mapping_reader->need_casting_ = true; } + // A differing nested type needs the AlignArrayToReadType reshape below. + if (!mapping_reader->non_partition_info_.non_partition_data_schema[i].Type()->Equals( + *mapping_reader->non_partition_info_.non_partition_read_schema[i].Type())) { + mapping_reader->need_casting_ = true; + } // Field name change (RENAME COLUMN) also requires mapping: data schema // carries the file's physical name while read schema carries the // post-rename logical name. If we skipped mapping, the inner reader's @@ -201,6 +206,7 @@ Result> FieldMappingReader::CastNonPartitionArrayI casted_array.reserve(field_count); casted_field_names.reserve(field_count); for (int32_t i = 0; i < field_count; i++) { + std::shared_ptr column; if (non_partition_info_.cast_executors[i] != nullptr) { auto single_column_array = struct_array->field(i); // if src array is dict, cast to string first @@ -213,18 +219,27 @@ Result> FieldMappingReader::CastNonPartitionArrayI arrow::compute::CastOptions::Safe(), arrow_pool_.get())); } PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr casted, + column, non_partition_info_.cast_executors[i]->Cast( single_column_array, non_partition_info_.non_partition_read_schema[i].Type(), arrow_pool_.get())); - casted_array.push_back(casted); - casted_field_names.push_back(non_partition_info_.non_partition_data_schema[i].Name()); } else { // read and data type may both be string type, but after adapter transform, type may be // dictionary, need reconstruct struct type - casted_array.push_back(struct_array->field(i)); - casted_field_names.push_back(non_partition_info_.non_partition_data_schema[i].Name()); + column = struct_array->field(i); + } + // Null-fill nested fields added by schema evolution. Only when the data and + // read types differ -- the reader may hand back a dictionary-encoded array + // for an unchanged type, which is not a reshape target. + if (!non_partition_info_.non_partition_data_schema[i].Type()->Equals( + *non_partition_info_.non_partition_read_schema[i].Type())) { + PAIMON_ASSIGN_OR_RAISE( + column, NestedProjectionUtils::AlignArrayToReadType( + column, non_partition_info_.non_partition_read_schema[i].Type(), + arrow_pool_.get())); } + casted_array.push_back(column); + casted_field_names.push_back(non_partition_info_.non_partition_data_schema[i].Name()); } PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr arrow_array, arrow::StructArray::Make(casted_array, casted_field_names)); diff --git a/src/paimon/core/io/field_mapping_reader_test.cpp b/src/paimon/core/io/field_mapping_reader_test.cpp index be9cc35a3..37f2be4e1 100644 --- a/src/paimon/core/io/field_mapping_reader_test.cpp +++ b/src/paimon/core/io/field_mapping_reader_test.cpp @@ -448,6 +448,91 @@ TEST_F(FieldMappingReaderTest, TestDictionaryTypeWithSchemaEvolution) { partition, expected_array); } +TEST_F(FieldMappingReaderTest, TestSchemaEvolutionAddedFieldInsideList) { + // A field `c`(id=12) was added inside the list's struct after the file was + // written. Reading the old file with the new schema must null-fill `c`. + auto id_field = [](const std::string& name, const std::shared_ptr& type, + int32_t id) { + return DataField::ConvertDataFieldToArrowField(DataField(id, arrow::field(name, type))); + }; + auto data_struct = + arrow::struct_({id_field("a", arrow::int32(), 10), id_field("b", arrow::utf8(), 11)}); + auto read_struct = + arrow::struct_({id_field("a", arrow::int32(), 10), id_field("b", arrow::utf8(), 11), + id_field("c", arrow::int32(), 12)}); + std::vector data_fields = { + DataField(100, arrow::field("items", arrow::list(arrow::field("item", data_struct))))}; + std::vector read_fields = { + DataField(100, arrow::field("items", arrow::list(arrow::field("item", read_struct))))}; + auto data_schema = DataField::ConvertDataFieldsToArrowSchema(data_fields); + auto read_schema = DataField::ConvertDataFieldsToArrowSchema(read_fields); + + auto data_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(data_schema->fields()), R"([ + [[[1, "x"], [2, "y"]]], + [[[3, "z"]]] + ])") + .ValueOrDie()); + + ASSERT_OK_AND_ASSIGN(auto mapping_builder, + FieldMappingBuilder::Create(read_schema, /*partition_keys=*/{}, + /*predicate=*/nullptr)); + ASSERT_OK_AND_ASSIGN(auto mapping, mapping_builder->CreateFieldMapping(data_fields)); + auto mock = std::make_unique( + data_array, arrow::struct_(data_schema->fields()), /*read_batch_size=*/8); + ASSERT_OK_AND_ASSIGN(auto reader, FieldMappingReader::Create( + read_schema->num_fields(), std::move(mock), + BinaryRow::EmptyRow(), std::move(mapping), + /*skip_map_selected_keys_filter_field_ids=*/{}, pool_)); + ASSERT_OK_AND_ASSIGN(auto result_array, ReadResultCollector::CollectResult(reader.get())); + + auto expect_array = + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(read_schema->fields()), R"([ + [[[1, "x", null], [2, "y", null]]], + [[[3, "z", null]]] + ])") + .ValueOrDie(); + auto expected_chunk = std::make_shared(arrow::ArrayVector({expect_array})); + ASSERT_TRUE(result_array->type()->Equals(expected_chunk->type())) + << result_array->type()->ToString() << " vs " << expected_chunk->type()->ToString(); + ASSERT_TRUE(result_array->Equals(expected_chunk)) + << result_array->ToString() << " vs " << expected_chunk->ToString(); +} + +TEST_F(FieldMappingReaderTest, TestSchemaEvolutionAddedFieldInsideListOrc) { + // ORC round-trip: added field inside a list's struct is null-filled. + auto id_field = [](const std::string& name, const std::shared_ptr& type, + int32_t id) { + return DataField::ConvertDataFieldToArrowField(DataField(id, arrow::field(name, type))); + }; + auto data_struct = + arrow::struct_({id_field("a", arrow::int32(), 10), id_field("b", arrow::int32(), 11)}); + auto read_struct = + arrow::struct_({id_field("a", arrow::int32(), 10), id_field("b", arrow::int32(), 11), + id_field("c", arrow::int32(), 12)}); + std::vector data_fields = { + DataField(100, arrow::field("items", arrow::list(arrow::field("item", data_struct))))}; + std::vector read_fields = { + DataField(100, arrow::field("items", arrow::list(arrow::field("item", read_struct))))}; + auto data_schema = DataField::ConvertDataFieldsToArrowSchema(data_fields); + auto read_schema = DataField::ConvertDataFieldsToArrowSchema(read_fields); + + auto data_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(data_schema->fields()), R"([ + [[[1, 2], [3, 4]]], + [[[5, 6]]] + ])") + .ValueOrDie()); + auto expect_array = + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(read_schema->fields()), R"([ + [[[1, 2, null], [3, 4, null]]], + [[[5, 6, null]]] + ])") + .ValueOrDie(); + CheckResult(data_schema, data_array, read_schema, /*predicate=*/nullptr, /*partition_keys=*/{}, + BinaryRow::EmptyRow(), expect_array); +} + TEST_F(FieldMappingReaderTest, TestSchemaEvolutionWithModifyType) { std::vector data_fields = {DataField(0, arrow::field("f0", arrow::utf8())), DataField(1, arrow::field("f1", arrow::float32())), diff --git a/src/paimon/core/utils/field_mapping.cpp b/src/paimon/core/utils/field_mapping.cpp index 0c90ae1b2..493d350c3 100644 --- a/src/paimon/core/utils/field_mapping.cpp +++ b/src/paimon/core/utils/field_mapping.cpp @@ -172,10 +172,11 @@ Result>> FieldMappingBuilder::CreateDa FieldTypeUtils::ConvertToFieldType(data_fields[i].Type()->id())); if (!read_fields[i].Type()->Equals(data_fields[i].Type())) { - if (read_type == FieldType::STRUCT) { - // STRUCT may still differ by nested pruning shape. No cast is - // needed — type pruning is handled by PruneDataType during - // field mapping construction. + auto read_type_id = read_fields[i].Type()->id(); + if (read_type_id == arrow::Type::STRUCT || read_type_id == arrow::Type::LIST || + read_type_id == arrow::Type::MAP) { + // Nested type differs by pruning/evolution; the reader's reshape + // handles it, no scalar cast. cast_executors.push_back(nullptr); continue; } diff --git a/src/paimon/core/utils/nested_projection_utils.cpp b/src/paimon/core/utils/nested_projection_utils.cpp index 5f82a22ab..0114cc701 100644 --- a/src/paimon/core/utils/nested_projection_utils.cpp +++ b/src/paimon/core/utils/nested_projection_utils.cpp @@ -26,11 +26,14 @@ #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/variant/variant_access_utils.h" #include "paimon/common/data/variant/variant_type_utils.h" #include "paimon/common/utils/string_utils.h" +#include "paimon/core/casting/casting_utils.h" #include "paimon/status.h" namespace paimon { @@ -150,12 +153,47 @@ Result NestedProjectionUtils::HasNestedSubfieldProjectionType( 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 EqualWithFieldIds(const std::shared_ptr& a, + const std::shared_ptr& b) { + if (a->id() != b->id() || a->num_fields() != b->num_fields()) { + 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. Such a read drops no field, so it is not a partial -/// projection of an enclosing repeated group and may pass through where a real one must fail. -bool IsVariantAccessSubstitution(const std::shared_ptr& read_type, - const std::shared_ptr& data_type) { - if (read_type->Equals(data_type)) { +/// projections and nothing else changed (matching paimon field IDs). +Result IsVariantAccessSubstitution(const std::shared_ptr& read_type, + const std::shared_ptr& data_type) { + PAIMON_ASSIGN_OR_RAISE(bool equal, EqualWithFieldIds(read_type, data_type)); + if (equal) { return true; } if (VariantAccessUtils::IsVariantAccessType(read_type) && @@ -169,25 +207,113 @@ bool IsVariantAccessSubstitution(const std::shared_ptr& read_ty for (int32_t i = 0; i < read_type->num_fields(); ++i) { const std::shared_ptr& read_child = read_type->field(i); const std::shared_ptr& data_child = data_type->field(i); - // LIST and MAP name their children by format convention, so only STRUCT is matched - // by name. - if (read_type->id() == arrow::Type::STRUCT && read_child->name() != data_child->name()) { - return false; + // 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; + } } - if (!IsVariantAccessSubstitution(read_child->type(), data_child->type())) { + 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> PruneRepeatedItemType( + const std::shared_ptr& read_type, + const std::shared_ptr& data_type, const char* container) { + 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 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 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 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 = std::static_pointer_cast(read_type); + auto data_map = std::static_pointer_cast(data_type); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr key, + PruneRepeatedItemType(read_map->key_type(), data_map->key_type(), container)); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr item, + PruneRepeatedItemType(read_map->item_type(), data_map->item_type(), container)); + return std::static_pointer_cast(std::make_shared( + data_map->key_field()->WithType(key), data_map->item_field()->WithType(item), + data_map->keys_sorted())); + } + default: + return data_type; + } +} + } // namespace Result>> NestedProjectionUtils::PruneDataType( const std::shared_ptr& read_type, const std::shared_ptr& data_type) { - // Identical types need no pruning. - if (read_type->Equals(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>(data_type); } @@ -232,25 +358,35 @@ Result>> NestedProjectionUtils::P return std::optional>(arrow::struct_(pruned_fields)); } case arrow::Type::LIST: { - if (IsVariantAccessSubstitution(read_type, data_type)) { + PAIMON_ASSIGN_OR_RAISE(bool list_substitution, + IsVariantAccessSubstitution(read_type, data_type)); + if (list_substitution) { return std::optional>(read_type); } - // Keep behavior aligned with format readers: partial projection inside - // LIST is unsupported and must fail fast. - return Status::Invalid( - fmt::format("PruneDataType does not support partial projection inside list: src {} " - "vs target {}", - data_type->ToString(), read_type->ToString())); + // Added fields (schema evolution) are allowed; dropped fields still fail. + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr item, + PruneRepeatedItemType(read_type->field(0)->type(), + data_type->field(0)->type(), "list")); + return std::optional>( + arrow::list(data_type->field(0)->WithType(item))); } case arrow::Type::MAP: { - if (IsVariantAccessSubstitution(read_type, data_type)) { + PAIMON_ASSIGN_OR_RAISE(bool map_substitution, + IsVariantAccessSubstitution(read_type, data_type)); + if (map_substitution) { return std::optional>(read_type); } - // Keep behavior aligned with format readers: partial projection inside - // MAP is unsupported and must fail fast. - return Status::Invalid(fmt::format( - "PruneDataType does not support partial projection inside map: src {} vs target {}", - data_type->ToString(), read_type->ToString())); + auto read_map = std::static_pointer_cast(read_type); + auto data_map = std::static_pointer_cast(data_type); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr key, + PruneRepeatedItemType(read_map->key_type(), data_map->key_type(), "map")); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr item, + PruneRepeatedItemType(read_map->item_type(), data_map->item_type(), "map")); + return std::optional>(std::make_shared( + 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 @@ -452,4 +588,130 @@ Result> NestedProjectionUtils::FilterMapArrayBySel return result_map; } +namespace { +// Strips physical-only differences from a leaf type: ORC lazy decoding wraps +// strings in a dictionary and may widen them to large_string. binary is not +// dictionary-encoded and large_binary is blob's real type, so neither is +// normalized. Two leaves with equal normalized types hold the same logical +// values. +std::shared_ptr NormalizeLeafRepresentation( + const std::shared_ptr& type) { + auto t = type; + if (t->id() == arrow::Type::DICTIONARY) { + t = std::static_pointer_cast(t)->value_type(); + } + if (t->id() == arrow::Type::LARGE_STRING) { + return arrow::utf8(); + } + return t; +} +} // namespace + +Result> NestedProjectionUtils::AlignArrayToReadType( + const std::shared_ptr& array, const std::shared_ptr& read_type, + arrow::MemoryPool* pool) { + PAIMON_ASSIGN_OR_RAISE(bool same, EqualWithFieldIds(array->type(), read_type)); + if (same) { + return array; + } + // Produce exactly `read_type` so every file yields the same output type: rebuild + // STRUCT/LIST/MAP with read-side types/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> 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 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: { + if (array->type()->id() != arrow::Type::LIST) { + return Status::Invalid(fmt::format("AlignArrayToReadType cannot reconcile {} to {}", + array->type()->ToString(), + read_type->ToString())); + } + auto read_list = std::static_pointer_cast(read_type); + auto values = arrow::MakeArray(data->child_data[0]); + PAIMON_ASSIGN_OR_RAISE(values, + AlignArrayToReadType(values, read_list->value_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 = std::static_pointer_cast(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 diff --git a/src/paimon/core/utils/nested_projection_utils.h b/src/paimon/core/utils/nested_projection_utils.h index 760b72176..2b4fd2838 100644 --- a/src/paimon/core/utils/nested_projection_utils.h +++ b/src/paimon/core/utils/nested_projection_utils.h @@ -85,6 +85,13 @@ class PAIMON_EXPORT NestedProjectionUtils { const std::shared_ptr& map_array, const std::vector& selected_keys, arrow::MemoryPool* pool); + /// Reshape `array` to `read_type`, null-filling nested fields added by schema + /// evolution. No-op when types match. STRUCT matches children by paimon field id; + /// LIST/MAP recurse into items, preserving offsets and validity. + static Result> AlignArrayToReadType( + const std::shared_ptr& array, + const std::shared_ptr& read_type, arrow::MemoryPool* pool); + private: static Result HasNestedSubfieldProjectionType( const std::shared_ptr& file_type, diff --git a/src/paimon/core/utils/nested_projection_utils_test.cpp b/src/paimon/core/utils/nested_projection_utils_test.cpp index 1fa8f6f72..979bd9293 100644 --- a/src/paimon/core/utils/nested_projection_utils_test.cpp +++ b/src/paimon/core/utils/nested_projection_utils_test.cpp @@ -16,11 +16,13 @@ #include "paimon/core/utils/nested_projection_utils.h" +#include "arrow/array/array_dict.h" #include "arrow/array/array_nested.h" #include "arrow/array/builder_binary.h" #include "arrow/array/builder_dict.h" #include "arrow/array/builder_nested.h" #include "arrow/array/builder_primitive.h" +#include "arrow/ipc/json_simple.h" #include "arrow/memory_pool.h" #include "arrow/type.h" #include "gtest/gtest.h" @@ -222,6 +224,177 @@ TEST(NestedProjectionUtilsTest, PruneDataTypeListDroppingSiblingOfVariantStillFa "partial projection inside list"); } +TEST(NestedProjectionUtilsTest, PruneDataTypeListStructSchemaEvolutionAddedField) { + // Added field (id=12) inside the list's struct: return the file struct. + auto data_inner = + arrow::struct_({MakeField("a", arrow::int32(), 10), MakeField("b", arrow::utf8(), 11)}); + auto read_inner = + arrow::struct_({MakeField("a", arrow::int32(), 10), MakeField("b", arrow::utf8(), 11), + MakeField("c", arrow::int32(), 12)}); + auto data_type = arrow::list(arrow::field("item", data_inner)); + auto read_type = arrow::list(arrow::field("item", read_inner)); + + ASSERT_OK_AND_ASSIGN(auto result, NestedProjectionUtils::PruneDataType(read_type, data_type)); + ASSERT_TRUE(result.has_value()); + ASSERT_TRUE(result.value()->Equals(*data_type)) << result.value()->ToString(); +} + +TEST(NestedProjectionUtilsTest, PruneDataTypeMapStructSchemaEvolutionAddedField) { + // Added field inside a MAP value. + auto data_inner = + arrow::struct_({MakeField("a", arrow::int32(), 10), MakeField("b", arrow::utf8(), 11)}); + auto read_inner = + arrow::struct_({MakeField("a", arrow::int32(), 10), MakeField("b", arrow::utf8(), 11), + MakeField("c", arrow::int32(), 12)}); + auto data_type = arrow::map(arrow::utf8(), data_inner); + auto read_type = arrow::map(arrow::utf8(), read_inner); + + ASSERT_OK_AND_ASSIGN(auto result, NestedProjectionUtils::PruneDataType(read_type, data_type)); + ASSERT_TRUE(result.has_value()); + ASSERT_TRUE(result.value()->Equals(*data_type)) << result.value()->ToString(); +} + +TEST(NestedProjectionUtilsTest, PruneDataTypeListStructDropAndAddStillFails) { + // Dropping a file field (b) is a real partial projection -- keep failing. + auto data_inner = + arrow::struct_({MakeField("a", arrow::int32(), 10), MakeField("b", arrow::utf8(), 11)}); + auto read_inner = + arrow::struct_({MakeField("a", arrow::int32(), 10), MakeField("c", arrow::int32(), 12)}); + auto data_type = arrow::list(arrow::field("item", data_inner)); + auto read_type = arrow::list(arrow::field("item", read_inner)); + + ASSERT_NOK_WITH_MSG(NestedProjectionUtils::PruneDataType(read_type, data_type), + "partial projection inside list"); +} + +TEST(NestedProjectionUtilsTest, AlignArrayToReadTypeNullFillsAddedListStructField) { + auto* pool = arrow::default_memory_pool(); + // file array: list> = [ [{1,"x"},{2,"y"}], [{3,"z"}] ] + arrow::Int32Builder ab(pool); + ASSERT_TRUE(ab.AppendValues({1, 2, 3}).ok()); + std::shared_ptr a; + ASSERT_TRUE(ab.Finish(&a).ok()); + arrow::StringBuilder bb(pool); + ASSERT_TRUE(bb.AppendValues({"x", "y", "z"}).ok()); + std::shared_ptr b; + ASSERT_TRUE(bb.Finish(&b).ok()); + auto data_struct_type = + arrow::struct_({MakeField("a", arrow::int32(), 10), MakeField("b", arrow::utf8(), 11)}); + std::shared_ptr struct_arr = + arrow::StructArray::Make({a, b}, data_struct_type->fields()).ValueOrDie(); + arrow::Int32Builder offb(pool); + ASSERT_TRUE(offb.AppendValues({0, 2, 3}).ok()); + std::shared_ptr offsets; + ASSERT_TRUE(offb.Finish(&offsets).ok()); + std::shared_ptr list_arr = + arrow::ListArray::FromArrays(*offsets, *struct_arr, pool).ValueOrDie(); + + // read type adds c:int(12) inside the struct. + auto read_struct = + arrow::struct_({MakeField("a", arrow::int32(), 10), MakeField("b", arrow::utf8(), 11), + MakeField("c", arrow::int32(), 12)}); + auto read_type = arrow::list(arrow::field("item", read_struct)); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr aligned, + NestedProjectionUtils::AlignArrayToReadType(list_arr, read_type, pool)); + ASSERT_TRUE(aligned->type()->Equals(*read_type)) << aligned->type()->ToString(); + auto out_struct = std::static_pointer_cast( + std::static_pointer_cast(aligned)->values()); + auto c_col = out_struct->GetFieldByName("c"); + ASSERT_NE(c_col, nullptr); + ASSERT_EQ(c_col->null_count(), c_col->length()); // added field is all null + auto a_col = std::static_pointer_cast(out_struct->GetFieldByName("a")); + ASSERT_EQ(a_col->Value(0), 1); + ASSERT_EQ(a_col->Value(2), 3); +} + +TEST(NestedProjectionUtilsTest, AlignArrayToReadTypeDecodesDictionaryLeafAndNullFills) { + // ORC lazy decoding returns dictionary; decode/cast to string. + auto* pool = arrow::default_memory_pool(); + auto dict_type = arrow::dictionary(arrow::int64(), arrow::large_utf8()); + auto a_dict = + arrow::ipc::internal::json::ArrayFromJSON(dict_type, R"(["x", "y", "x"])").ValueOrDie(); + auto data_struct = arrow::struct_({MakeField("a", dict_type, 10)}); + auto struct_arr = arrow::StructArray::Make({a_dict}, data_struct->fields()).ValueOrDie(); + + auto read_type = + arrow::struct_({MakeField("a", arrow::utf8(), 10), MakeField("b", arrow::int32(), 11)}); + ASSERT_OK_AND_ASSIGN(std::shared_ptr aligned, + NestedProjectionUtils::AlignArrayToReadType(struct_arr, read_type, pool)); + ASSERT_TRUE(aligned->type()->Equals(*read_type)) << aligned->type()->ToString(); + auto out = std::static_pointer_cast(aligned); + ASSERT_EQ(std::static_pointer_cast(out->GetFieldByName("a"))->GetString(0), + "x"); + auto b = out->GetFieldByName("b"); + ASSERT_EQ(b->null_count(), b->length()); +} + +TEST(NestedProjectionUtilsTest, AlignArrayToReadTypeAppliesReadNullability) { + auto* pool = arrow::default_memory_pool(); + arrow::Int32Builder ab(pool); + ASSERT_TRUE(ab.AppendValues({1, 2}).ok()); + std::shared_ptr a_arr; + ASSERT_TRUE(ab.Finish(&a_arr).ok()); + auto data_field = DataField::ConvertDataFieldToArrowField( + DataField(10, arrow::field("a", arrow::int32(), /*nullable=*/false))); + auto struct_arr = arrow::StructArray::Make({a_arr}, {data_field}).ValueOrDie(); + auto read_type = arrow::struct_({MakeField("a", arrow::int32(), 10)}); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr aligned, + NestedProjectionUtils::AlignArrayToReadType(struct_arr, read_type, pool)); + ASSERT_TRUE(aligned->type()->field(0)->nullable()); +} + +TEST(NestedProjectionUtilsTest, AlignArrayToReadTypeFieldIdChangeNullFillsNotLeak) { + // a(id=10) replaced by a(id=11), same name/type: new field must read null, not leak. + auto* pool = arrow::default_memory_pool(); + arrow::Int32Builder ab(pool); + ASSERT_TRUE(ab.AppendValues({42}).ok()); + std::shared_ptr a_arr; + ASSERT_TRUE(ab.Finish(&a_arr).ok()); + auto data_struct = arrow::struct_({MakeField("a", arrow::int32(), 10)}); + auto struct_arr = arrow::StructArray::Make({a_arr}, data_struct->fields()).ValueOrDie(); + auto read_type = arrow::struct_({MakeField("a", arrow::int32(), 11)}); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr aligned, + NestedProjectionUtils::AlignArrayToReadType(struct_arr, read_type, pool)); + auto a_out = std::static_pointer_cast(aligned)->GetFieldByName("a"); + ASSERT_NE(a_out, nullptr); + ASSERT_EQ(a_out->null_count(), a_out->length()); +} + +TEST(NestedProjectionUtilsTest, AlignArrayToReadTypeRejectsLeafTypeChange) { + auto* pool = arrow::default_memory_pool(); + arrow::Int32Builder ab(pool); + ASSERT_TRUE(ab.AppendValues({1}).ok()); + std::shared_ptr a_arr; + ASSERT_TRUE(ab.Finish(&a_arr).ok()); + auto data_struct = arrow::struct_({MakeField("a", arrow::int32(), 10)}); + auto struct_arr = arrow::StructArray::Make({a_arr}, data_struct->fields()).ValueOrDie(); + auto read_type = arrow::struct_({MakeField("a", arrow::int64(), 10)}); + + ASSERT_NOK_WITH_MSG(NestedProjectionUtils::AlignArrayToReadType(struct_arr, read_type, pool), + "AlignArrayToReadType unsupported leaf type change"); +} + +TEST(NestedProjectionUtilsTest, AlignArrayToReadTypeKeepsNestedLargeBinaryBlob) { + auto* pool = arrow::default_memory_pool(); + arrow::LargeBinaryBuilder blobb(pool); + ASSERT_TRUE(blobb.AppendValues({"a", "b"}).ok()); + std::shared_ptr blob; + ASSERT_TRUE(blobb.Finish(&blob).ok()); + auto data_struct = arrow::struct_({MakeField("blob", arrow::large_binary(), 10)}); + auto struct_arr = arrow::StructArray::Make({blob}, data_struct->fields()).ValueOrDie(); + auto read_type = arrow::struct_( + {MakeField("blob", arrow::large_binary(), 10), MakeField("c", arrow::int32(), 11)}); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr aligned, + NestedProjectionUtils::AlignArrayToReadType(struct_arr, read_type, pool)); + auto blob_out = std::static_pointer_cast(aligned)->GetFieldByName("blob"); + ASSERT_EQ(blob_out->type_id(), arrow::Type::LARGE_BINARY); + ASSERT_TRUE(blob_out->Equals(*blob)); +} + TEST(NestedProjectionUtilsTest, HasNestedSubfieldProjectionNoProjection) { auto file_schema = arrow::schema({ MakeField("f0", arrow::int32(), 1), diff --git a/test/inte/write_and_read_inte_test.cpp b/test/inte/write_and_read_inte_test.cpp index b1dd84219..22dc142f9 100644 --- a/test/inte/write_and_read_inte_test.cpp +++ b/test/inte/write_and_read_inte_test.cpp @@ -420,6 +420,95 @@ TEST_P(WriteAndReadInteTest, TestNestedType) { ASSERT_TRUE(success); } +TEST_P(WriteAndReadInteTest, TestSchemaEvolutionAddFieldInsideListAndMap) { + auto [file_format, file_system] = GetParam(); + if (file_format == "lance" || file_format == "avro") { + return; + } + auto list_struct = + arrow::struct_({arrow::field("a", arrow::int32()), arrow::field("b", arrow::utf8())}); + auto map_value = arrow::struct_({arrow::field("m1", arrow::int32())}); + arrow::FieldVector fields = { + arrow::field("id", arrow::int32()), + arrow::field("items", arrow::list(arrow::field("item", list_struct))), + arrow::field("props", arrow::map(arrow::utf8(), map_value)), + }; + std::map options = { + {Options::MANIFEST_FORMAT, "avro"}, + {Options::FILE_FORMAT, file_format}, + {Options::TARGET_FILE_SIZE, "1024"}, + {Options::BUCKET, "-1"}, + {Options::FILE_SYSTEM, file_system}, + // Exercise ORC lazy decoding, which returns nested strings as dictionaries. + {"orc.read.enable-lazy-decoding", "true"}, + }; + if (file_system == "jindo") { + options = AddOptionsForJindo(options); + } + ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(test_dir_, arrow::schema(fields), + /*partition_keys=*/{}, /*primary_keys=*/{}, + options, /*is_streaming_mode=*/false)); + ASSERT_OK_AND_ASSIGN(auto batch, TestHelper::MakeRecordBatch(arrow::struct_(fields), + R"([ + [1, [[10, "x"], [20, "y"]], [["k1", [100]]]], + [2, [[30, "z"]], [["k2", [200]]]] + ])", + /*partition_map=*/{}, + /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0, + /*expected_commit_messages=*/std::nullopt)); + + std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar"); + SchemaManager schema_manager(dir_->GetFileSystem(), table_path); + ASSERT_OK_AND_ASSIGN(auto schema_v0, schema_manager.ReadSchema(0)); + std::vector fields_v0 = schema_v0->Fields(); + int32_t next_id = schema_v0->HighestFieldId(); + + auto items_field = fields_v0[1].ArrowField(); + auto items_list = arrow::internal::checked_pointer_cast(items_field->type()); + auto items_struct = + arrow::internal::checked_pointer_cast(items_list->value_type()); + auto c_field = DataField::ConvertDataFieldToArrowField( + DataField(++next_id, arrow::field("c", arrow::int32()))); + auto new_items_type = arrow::list(items_list->value_field()->WithType( + arrow::struct_({items_struct->field(0), items_struct->field(1), c_field}))); + + auto props_field = fields_v0[2].ArrowField(); + auto props_map = arrow::internal::checked_pointer_cast(props_field->type()); + auto props_value = + arrow::internal::checked_pointer_cast(props_map->item_type()); + auto m2_field = DataField::ConvertDataFieldToArrowField( + DataField(++next_id, arrow::field("m2", arrow::int32()))); + auto new_props_type = arrow::map( + props_map->key_type(), + props_map->item_field()->WithType(arrow::struct_({props_value->field(0), m2_field}))); + + std::vector fields_v1 = fields_v0; + fields_v1[1] = DataField(fields_v0[1].Id(), items_field->WithType(new_items_type)); + fields_v1[2] = DataField(fields_v0[2].Id(), props_field->WithType(new_props_type)); + ASSERT_OK(WriteNextSchema(fields_v1, next_id, options)); + + auto expected_type = arrow::struct_({ + arrow::field("_VALUE_KIND", arrow::int8()), + arrow::field("id", arrow::int32()), + arrow::field("items", arrow::list(arrow::field( + "item", arrow::struct_({arrow::field("a", arrow::int32()), + arrow::field("b", arrow::utf8()), + arrow::field("c", arrow::int32())})))), + arrow::field("props", arrow::map(arrow::utf8(), + arrow::struct_({arrow::field("m1", arrow::int32()), + arrow::field("m2", arrow::int32())}))), + }); + ASSERT_OK_AND_ASSIGN(auto splits, + helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt)); + ASSERT_OK_AND_ASSIGN(bool success, helper->ReadAndCheckResult(expected_type, splits, + R"([ + [0, 1, [[10, "x", null], [20, "y", null]], [["k1", [100, null]]]], + [0, 2, [[30, "z", null]], [["k2", [200, null]]]] + ])")); + ASSERT_TRUE(success); +} + TEST_P(WriteAndReadInteTest, TestAppendExternalPath) { arrow::FieldVector fields = { arrow::field("f0", arrow::utf8()), arrow::field("f1", arrow::int32()), @@ -2455,10 +2544,9 @@ TEST_P(WriteAndReadInteTest, TestMapSharedShreddingStructValueSchemaEvolutionRea fields_with_changed_tag_value[1] = DataField(fields_v0[1].Id(), tag_field->WithType(changed_tag_value_type)); ASSERT_OK(WriteNextSchema(fields_with_changed_tag_value, schema_v0->HighestFieldId(), options)); - ASSERT_NOK_WITH_MSG(read_fields({"tags"}), - "PruneDataType does not support partial projection inside map: src " - "map> vs target " - "map>"); + ASSERT_NOK_WITH_MSG( + read_fields({"tags"}), + "PruneDataType nested item type mismatch inside map: read string vs data int64"); auto profile_field = fields_v0[2].ArrowField(); auto profile_struct =