Skip to content

Commit 1acca73

Browse files
zhf999zhf999
authored
fix(parquet): fallback nested field reading to RowGroup reading (walkaround) (alibaba#371)
1 parent 1770fee commit 1acca73

2 files changed

Lines changed: 281 additions & 1 deletion

File tree

‎src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp‎

Lines changed: 270 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -873,5 +873,275 @@ TEST_F(PageFilteredRowGroupReaderTest, ComputePageRangesWithDictionaryEncoding)
873873
auto partial_concat = arrow::Concatenate(result_partial->chunks()).ValueOrDie();
874874
ASSERT_TRUE(partial_concat->Equals(expected_struct));
875875
}
876+
/// Helper: build a StructArray with a top-level int32 "id" column and a nested struct column
877+
/// "info" containing two int32 fields: "x" and "y".
878+
/// id[i] = i, info.x[i] = i * 100, info.y[i] = i * 100 + 1, for i in [0, N).
879+
///
880+
/// Arrow schema: { id: int32, info: struct<x: int32, y: int32> }
881+
/// Parquet leaf columns: [id (index 0), info.x (index 1), info.y (index 2)]
882+
static std::shared_ptr<arrow::StructArray> MakeNestedStructData(int32_t num_rows) {
883+
arrow::Int32Builder id_builder, x_builder, y_builder;
884+
EXPECT_TRUE(id_builder.Reserve(num_rows).ok());
885+
EXPECT_TRUE(x_builder.Reserve(num_rows).ok());
886+
EXPECT_TRUE(y_builder.Reserve(num_rows).ok());
887+
for (int32_t i = 0; i < num_rows; ++i) {
888+
id_builder.UnsafeAppend(i);
889+
x_builder.UnsafeAppend(i * 100);
890+
y_builder.UnsafeAppend(i * 100 + 1);
891+
}
892+
auto id_array = id_builder.Finish().ValueOrDie();
893+
auto x_array = x_builder.Finish().ValueOrDie();
894+
auto y_array = y_builder.Finish().ValueOrDie();
895+
896+
auto field_x = arrow::field("x", arrow::int32());
897+
auto field_y = arrow::field("y", arrow::int32());
898+
auto inner_struct =
899+
arrow::StructArray::Make({x_array, y_array}, {field_x, field_y}).ValueOrDie();
900+
901+
auto field_id = arrow::field("id", arrow::int32());
902+
auto field_info = arrow::field("info", arrow::struct_({field_x, field_y}));
903+
return arrow::StructArray::Make({id_array, inner_struct}, {field_id, field_info}).ValueOrDie();
904+
}
905+
906+
/// Test: rowgroup-level filtering on a file with nested struct columns.
907+
///
908+
/// This test exposes the bug where BuildPageFilteredSchema fails to correctly map
909+
/// Parquet leaf column indices to Arrow fields for nested types, and
910+
/// ReadFilteredRowGroup cannot correctly assemble nested column results.
911+
///
912+
/// Schema: { id: int32, info: struct<x: int32, y: int32> }
913+
/// Parquet leaf columns: [id=0, info.x=1, info.y=2]
914+
/// 100 rows, 10 per page, 2 row groups.
915+
/// Predicate: id >= 70 → row groups 0 skipped, row groups 1 read → 50 rows expected.
916+
/// The read schema requests both "id" and "info" columns.
917+
TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnRowGroupFilter) {
918+
std::string file_name = dir_->Str() + "/nested_struct_filter.parquet";
919+
auto data = MakeNestedStructData(100);
920+
WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50);
921+
922+
auto field_x = arrow::field("x", arrow::int32());
923+
auto field_y = arrow::field("y", arrow::int32());
924+
auto read_schema = arrow::schema({arrow::field("id", arrow::int32()),
925+
arrow::field("info", arrow::struct_({field_x, field_y}))});
926+
927+
auto predicate = PredicateBuilder::GreaterOrEqual(
928+
/*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(70));
929+
930+
std::shared_ptr<arrow::ChunkedArray> result;
931+
ReadWithPredicateImpl(file_name, read_schema, predicate, &result);
932+
933+
// Should get rows 50-99 = 50 rows
934+
ASSERT_TRUE(result);
935+
ASSERT_EQ(50, result->length());
936+
937+
// Build expected result: rows 50-99 from the original data
938+
auto expected = data->Slice(50, 50);
939+
ASSERT_TRUE(expected->Equals(result->chunk(0)));
940+
}
941+
942+
/// Test: Page-level filtering reading the nested struct column along with the predicate column.
943+
///
944+
/// This verifies that when reading a subset of columns that includes a nested column
945+
/// and the predicate column, the schema mapping and column assembly work correctly.
946+
///
947+
/// Schema: { id: int32, info: struct<x: int32, y: int32> }
948+
/// Read schema: { id: int32, info: struct<x: int32, y: int32> }
949+
/// Predicate on "id": id >= 70.
950+
TEST_F(PageFilteredRowGroupReaderTest, NestedStructColumnOnlyReadIdField) {
951+
std::string file_name = dir_->Str() + "/nested_struct_only_nested.parquet";
952+
auto data = MakeNestedStructData(100);
953+
WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50);
954+
955+
auto field_id = arrow::field("id", arrow::int32());
956+
auto field_x = arrow::field("x", arrow::int32());
957+
auto field_y = arrow::field("y", arrow::int32());
958+
auto field_info = arrow::field("info", arrow::struct_({field_x, field_y}));
959+
// Read "id" column only
960+
auto read_schema = arrow::schema({field_id});
961+
962+
// Predicate is on "id" (field_index=0 in file schema)
963+
auto predicate = PredicateBuilder::GreaterOrEqual(
964+
/*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(70));
965+
966+
std::shared_ptr<arrow::ChunkedArray> result;
967+
ReadWithPredicateImpl(file_name, read_schema, predicate, &result);
968+
969+
// Should get rows 70-99 = 30 rows
970+
ASSERT_TRUE(result);
971+
ASSERT_EQ(30, result->length());
972+
973+
auto result_struct = std::dynamic_pointer_cast<arrow::StructArray>(result->chunk(0));
974+
ASSERT_TRUE(result_struct);
975+
ASSERT_TRUE(data->field(0)->Slice(70, 30)->Equals(result_struct->field(0)));
976+
}
977+
978+
/// Helper: build a StructArray with an int32 "id" column and a list<int32> "tags" column.
979+
/// id[i] = i, tags[i] = [i*10, i*10+1], for i in [0, N).
980+
///
981+
/// Arrow schema: { id: int32, tags: list<item: int32> }
982+
/// Parquet leaf columns: [id (index 0), tags.item (index 1)]
983+
static std::shared_ptr<arrow::StructArray> MakeListColumnData(int32_t num_rows) {
984+
arrow::Int32Builder id_builder;
985+
EXPECT_TRUE(id_builder.Reserve(num_rows).ok());
986+
for (int32_t i = 0; i < num_rows; ++i) {
987+
id_builder.UnsafeAppend(i);
988+
}
989+
auto id_array = id_builder.Finish().ValueOrDie();
990+
991+
auto value_builder = std::make_shared<arrow::Int32Builder>();
992+
arrow::ListBuilder list_builder(arrow::default_memory_pool(), value_builder);
993+
for (int32_t i = 0; i < num_rows; ++i) {
994+
EXPECT_TRUE(list_builder.Append().ok());
995+
EXPECT_TRUE(value_builder->Append(i * 10).ok());
996+
EXPECT_TRUE(value_builder->Append(i * 10 + 1).ok());
997+
}
998+
auto list_array = list_builder.Finish().ValueOrDie();
999+
1000+
auto field_id = arrow::field("id", arrow::int32());
1001+
auto field_tags = arrow::field("tags", arrow::list(arrow::field("item", arrow::int32())));
1002+
return arrow::StructArray::Make({id_array, list_array}, {field_id, field_tags}).ValueOrDie();
1003+
}
1004+
1005+
/// Helper: build a StructArray with an int32 "id" column and a map<utf8, int32> "props" column.
1006+
/// id[i] = i, props[i] = {"k_i": i * 100}, for i in [0, N).
1007+
///
1008+
/// Arrow schema: { id: int32, props: map<utf8, int32> }
1009+
/// Parquet leaf columns: [id (index 0), props.key (index 1), props.value (index 2)]
1010+
static std::shared_ptr<arrow::StructArray> MakeMapColumnData(int32_t num_rows) {
1011+
arrow::Int32Builder id_builder;
1012+
EXPECT_TRUE(id_builder.Reserve(num_rows).ok());
1013+
for (int32_t i = 0; i < num_rows; ++i) {
1014+
id_builder.UnsafeAppend(i);
1015+
}
1016+
auto id_array = id_builder.Finish().ValueOrDie();
1017+
1018+
auto key_builder = std::make_shared<arrow::StringBuilder>();
1019+
auto value_builder = std::make_shared<arrow::Int32Builder>();
1020+
arrow::MapBuilder map_builder(arrow::default_memory_pool(), key_builder, value_builder);
1021+
for (int32_t i = 0; i < num_rows; ++i) {
1022+
EXPECT_TRUE(map_builder.Append().ok());
1023+
std::string key = "k_" + std::to_string(i);
1024+
EXPECT_TRUE(key_builder->Append(key).ok());
1025+
EXPECT_TRUE(value_builder->Append(i * 100).ok());
1026+
}
1027+
auto map_array = map_builder.Finish().ValueOrDie();
1028+
1029+
auto field_id = arrow::field("id", arrow::int32());
1030+
auto field_props = arrow::field("props", arrow::map(arrow::utf8(), arrow::int32()));
1031+
return arrow::StructArray::Make({id_array, map_array}, {field_id, field_props}).ValueOrDie();
1032+
}
1033+
1034+
/// Test: rowgroup-level filtering on a file with a list column.
1035+
///
1036+
/// Schema: { id: int32, tags: list<item: int32> }
1037+
/// 100 rows, 10 per page, 2 row groups.
1038+
/// Predicate: id >= 70 → row groups 0 skipped, row groups 1 read → 50 rows expected.
1039+
TEST_F(PageFilteredRowGroupReaderTest, NestedListColumnRowGroupFilter) {
1040+
std::string file_name = dir_->Str() + "/nested_list_filter.parquet";
1041+
auto data = MakeListColumnData(100);
1042+
WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50);
1043+
1044+
auto read_schema =
1045+
arrow::schema({arrow::field("id", arrow::int32()),
1046+
arrow::field("tags", arrow::list(arrow::field("item", arrow::int32())))});
1047+
1048+
auto predicate = PredicateBuilder::GreaterOrEqual(
1049+
/*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(70));
1050+
1051+
std::shared_ptr<arrow::ChunkedArray> result;
1052+
ReadWithPredicateImpl(file_name, read_schema, predicate, &result);
1053+
1054+
ASSERT_TRUE(result);
1055+
ASSERT_EQ(50, result->length());
1056+
1057+
// Build expected result: rows 50-99 from the original data
1058+
auto expected = data->Slice(50, 50);
1059+
ASSERT_TRUE(expected->Equals(result->chunk(0)));
1060+
}
1061+
1062+
/// Test: rowgroup filtering on a file with a map column.
1063+
///
1064+
/// Schema: { id: int32, props: map<utf8, int32> }
1065+
/// 100 rows, 10 per page, 2 row groups.
1066+
/// Predicate: id >= 70 → row groups 0 skipped, row groups 1 read → 50 rows expected.
1067+
TEST_F(PageFilteredRowGroupReaderTest, NestedMapColumnRowGroupFilter) {
1068+
std::string file_name = dir_->Str() + "/nested_map_filter.parquet";
1069+
auto data = MakeMapColumnData(100);
1070+
WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50);
1071+
1072+
auto read_schema =
1073+
arrow::schema({arrow::field("id", arrow::int32()),
1074+
arrow::field("props", arrow::map(arrow::utf8(), arrow::int32()))});
1075+
1076+
auto predicate = PredicateBuilder::GreaterOrEqual(
1077+
/*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(70));
1078+
1079+
std::shared_ptr<arrow::ChunkedArray> result;
1080+
ReadWithPredicateImpl(file_name, read_schema, predicate, &result);
1081+
1082+
ASSERT_TRUE(result);
1083+
ASSERT_EQ(50, result->length());
1084+
1085+
// Build expected result: rows 50-99 from the original data
1086+
auto expected = data->Slice(50, 50);
1087+
ASSERT_TRUE(expected->Equals(result->chunk(0)));
1088+
}
1089+
1090+
/// Test: rowgroup-level filtering with multiple adjacent nested columns (struct + list).
1091+
///
1092+
/// Schema: { id: int32, info: struct<x: int32, y: int32>, tags: list<item: int32> }
1093+
/// This tests the boundary handling when two nested fields are adjacent in the schema.
1094+
/// Predicate: id >= 70 → row groups 0 skipped, row groups 1 read → 50 rows expected.
1095+
TEST_F(PageFilteredRowGroupReaderTest, MultipleAdjacentNestedColumns) {
1096+
std::string file_name = dir_->Str() + "/multi_nested.parquet";
1097+
1098+
// Build data with id, info (struct), tags (list)
1099+
arrow::Int32Builder id_builder, x_builder, y_builder;
1100+
ASSERT_TRUE(id_builder.Reserve(100).ok());
1101+
ASSERT_TRUE(x_builder.Reserve(100).ok());
1102+
ASSERT_TRUE(y_builder.Reserve(100).ok());
1103+
auto value_builder = std::make_shared<arrow::Int32Builder>();
1104+
arrow::ListBuilder list_builder(arrow::default_memory_pool(), value_builder);
1105+
1106+
for (int32_t i = 0; i < 100; ++i) {
1107+
id_builder.UnsafeAppend(i);
1108+
x_builder.UnsafeAppend(i * 100);
1109+
y_builder.UnsafeAppend(i * 100 + 1);
1110+
ASSERT_TRUE(list_builder.Append().ok());
1111+
ASSERT_TRUE(value_builder->Append(i * 10).ok());
1112+
}
1113+
auto id_array = id_builder.Finish().ValueOrDie();
1114+
auto x_array = x_builder.Finish().ValueOrDie();
1115+
auto y_array = y_builder.Finish().ValueOrDie();
1116+
auto list_array = list_builder.Finish().ValueOrDie();
1117+
1118+
auto field_x = arrow::field("x", arrow::int32());
1119+
auto field_y = arrow::field("y", arrow::int32());
1120+
auto inner_struct =
1121+
arrow::StructArray::Make({x_array, y_array}, {field_x, field_y}).ValueOrDie();
1122+
1123+
auto field_id = arrow::field("id", arrow::int32());
1124+
auto field_info = arrow::field("info", arrow::struct_({field_x, field_y}));
1125+
auto field_tags = arrow::field("tags", arrow::list(arrow::field("item", arrow::int32())));
1126+
auto data = arrow::StructArray::Make({id_array, inner_struct, list_array},
1127+
{field_id, field_info, field_tags})
1128+
.ValueOrDie();
1129+
1130+
WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/50);
1131+
1132+
auto read_schema = arrow::schema({field_id, field_info, field_tags});
1133+
auto predicate = PredicateBuilder::GreaterOrEqual(
1134+
/*field_index=*/0, /*field_name=*/"id", FieldType::INT, Literal(70));
1135+
1136+
std::shared_ptr<arrow::ChunkedArray> result;
1137+
ReadWithPredicateImpl(file_name, read_schema, predicate, &result);
1138+
1139+
ASSERT_TRUE(result);
1140+
ASSERT_EQ(50, result->length());
1141+
1142+
// Build expected result: rows 50-99 from the original data
1143+
auto expected = data->Slice(50, 50);
1144+
ASSERT_TRUE(expected->Equals(result->chunk(0)));
1145+
}
8761146

8771147
} // namespace paimon::parquet::test

‎src/paimon/format/parquet/parquet_file_batch_reader.cpp‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
#include "paimon/common/metrics/metrics_impl.h"
4040
#include "paimon/common/utils/arrow/status_utils.h"
4141
#include "paimon/common/utils/options_utils.h"
42+
#include "paimon/core/schema/arrow_schema_validator.h"
4243
#include "paimon/format/parquet/parquet_field_id_converter.h"
4344
#include "paimon/format/parquet/parquet_format_defs.h"
4445
#include "paimon/format/parquet/parquet_timestamp_converter.h"
@@ -129,6 +130,13 @@ Status ParquetFileBatchReader::SetReadSchema(
129130

130131
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Schema> file_schema, reader_->GetSchema());
131132
std::unordered_map<std::string, std::vector<int32_t>> field_index_map;
133+
bool has_nested_field = false;
134+
for (const auto& field : read_schema->fields()) {
135+
if (ArrowSchemaValidator::IsNestedType(field->type())) {
136+
has_nested_field = true;
137+
break;
138+
}
139+
}
132140
int32_t i = 0;
133141
for (const auto& field : file_schema->fields()) {
134142
std::vector<int32_t> v;
@@ -166,7 +174,9 @@ Status ParquetFileBatchReader::SetReadSchema(
166174
bool enable_page_index_filter,
167175
OptionsUtils::GetValueFromMap<bool>(options_, PARQUET_READ_ENABLE_PAGE_INDEX_FILTER,
168176
DEFAULT_PARQUET_READ_ENABLE_PAGE_INDEX_FILTER));
169-
if (enable_page_index_filter) {
177+
// walkaround: page index filter does not support nested fields for now, skip page index
178+
// filter if there is any nested field in the schema
179+
if (enable_page_index_filter && !has_nested_field) {
170180
// Build column name to index map for page-level filtering.
171181
// For leaf columns, indices[0] is the correct leaf column index in Parquet.
172182
// For nested types (struct/list/map), FlattenSchema produces multiple leaf indices,

0 commit comments

Comments
 (0)