fix: nested list schema evolution (alibaba#446)

Author: XiaoHongbo-Hope

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<std::unique_ptr<DataEvolutionFileReader>> DataEvolutionFileReader::Create(
     std::vector<std::unique_ptr<BatchReader>>&& readers,
     const std::shared_ptr<arrow::Schema>& read_schema, int32_t read_batch_size,
@@ -82,6 +85,7 @@ Result<BatchReader::ReadBatchWithBitmap> 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(

src/paimon/core/io/field_mapping_reader.cpp

@@ -149,6 +149,11 @@ Result<std::unique_ptr<FieldMappingReader>> 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<std::shared_ptr<arrow::Array>> 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<arrow::Array> 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<std::shared_ptr<arrow::Array>> FieldMappingReader::CastNonPartitionArrayI
                                        arrow::compute::CastOptions::Safe(), arrow_pool_.get()));
             }
             PAIMON_ASSIGN_OR_RAISE(
-                std::shared_ptr<arrow::Array> 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_array,
                                       arrow::StructArray::Make(casted_array, casted_field_names));

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<arrow::DataType>& 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<DataField> data_fields = {
+        DataField(100, arrow::field("items", arrow::list(arrow::field("item", data_struct))))};
+    std::vector<DataField> 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::StructArray>(
+        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<MockFileBatchReader>(
+        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::ChunkedArray>(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<arrow::DataType>& 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<DataField> data_fields = {
+        DataField(100, arrow::field("items", arrow::list(arrow::field("item", data_struct))))};
+    std::vector<DataField> 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::StructArray>(
+        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<DataField> data_fields = {DataField(0, arrow::field("f0", arrow::utf8())),
                                           DataField(1, arrow::field("f1", arrow::float32())),

src/paimon/core/utils/field_mapping.cpp

@@ -172,10 +172,11 @@ Result<std::vector<std::shared_ptr<CastExecutor>>> 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;
             }