Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
de10aa5
Support schema-evolution added fields inside list/map in PruneDataType
XiaoHongbo-Hope Jul 23, 2026
0d00149
Null-fill nested added fields in data-evolution reader assembly
XiaoHongbo-Hope Jul 23, 2026
d0ed347
Reshape exist arrays for nested evolution in single-file reader path
XiaoHongbo-Hope Jul 23, 2026
9d50b65
Add array-level test for AlignArrayToReadType nested null-fill
XiaoHongbo-Hope Jul 23, 2026
bf772a8
Add end-to-end test: read list<struct> file with an evolution-added i…
XiaoHongbo-Hope Jul 23, 2026
83ee0e2
Trim comments
XiaoHongbo-Hope Jul 23, 2026
48c30b9
Address review comments on PR #446
XiaoHongbo-Hope Jul 23, 2026
b565f23
Merge branch 'main' into fix_nested_list_schema_evolution
lxy-9602 Jul 23, 2026
113adb1
Handle dictionary-encoded nested leaves from ORC lazy decoding
XiaoHongbo-Hope Jul 23, 2026
9dc15e3
Fix field-ID-blind equality and nested rename; clang-format
XiaoHongbo-Hope Jul 23, 2026
9492011
Address review: drop redundant DE-reader alignment; JSON dict test
XiaoHongbo-Hope Jul 23, 2026
1740dd5
Add list/map schema-evolution inte test; match nested fields by name
XiaoHongbo-Hope Jul 23, 2026
55b47dc
Cast leaves to the read type and compare nested nullability
XiaoHongbo-Hope Jul 23, 2026
ee14e41
Trim comment
XiaoHongbo-Hope Jul 23, 2026
b8ad679
Update expected message for map value type-change rejection
XiaoHongbo-Hope Jul 23, 2026
00131a8
Restrict AlignArrayToReadType leaf cast to representation-only changes
XiaoHongbo-Hope Jul 23, 2026
0ffc476
Only normalize large_string in leaf representation check, not large_b…
XiaoHongbo-Hope Jul 24, 2026
2b78370
Merge branch 'main' into fix_nested_list_schema_evolution
lxy-9602 Jul 24, 2026
0b85137
Merge branch 'main' into fix_nested_list_schema_evolution
lxy-9602 Jul 24, 2026
2e70b7d
Add nested large_binary blob alignment test
XiaoHongbo-Hope Jul 24, 2026
def0fee
Merge branch 'main' into fix_nested_list_schema_evolution
lxy-9602 Jul 27, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/paimon/common/reader/data_evolution_file_reader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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,
Expand Down Expand Up @@ -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]));
}

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One question I still have: if each file is already aligned via AlignArrayToReadType in field_mapping_reader.cpp, why do we need to do the same thing again in data_evolution_file_reader.cpp?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One question I still have: if each file is already aligned via AlignArrayToReadType in field_mapping_reader.cpp, why do we need to do the same thing again in data_evolution_file_reader.cpp?

Good catch, updated.

PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
Expand Down
25 changes: 20 additions & 5 deletions src/paimon/core/io/field_mapping_reader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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));
Expand Down
85 changes: 85 additions & 0 deletions src/paimon/core/io/field_mapping_reader_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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())),
Expand Down
9 changes: 5 additions & 4 deletions src/paimon/core/utils/field_mapping.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down
Loading
Loading