From f490d80fe976e94f686bd6ad191c9a8ebcfd81ca Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 12 Jul 2026 13:51:08 +0800 Subject: [PATCH] [fix](be) Count complex Parquet structs safely ### What problem does this PR solve? Issue Number: N/A Related PR: N/A Problem Summary: Parquet COUNT pushdown used a STRUCT root repetition level to identify top-level rows when a representative leaf was repeated. Because the STRUCT repetition level is zero, all level entries were skipped and a correctness check aborted the BE. The non-null threshold also used the repeated leaf definition level and could count NULL structs. Identify top-level rows with repetition level zero and use the root nullable definition level so STRUCT columns containing ARRAY or MAP children are counted correctly. ### Release note Fix BE aborts and incorrect COUNT results for Parquet STRUCT columns containing repeated children. ### Check List (For Author) - Test: Unit Test - NewParquetReaderTest targeted complex STRUCT and existing complex COUNT tests - Behavior changed: Yes, COUNT pushdown handles repeated descendants using top-level row semantics. - Does this need documentation: No --- be/src/format_v2/parquet/parquet_reader.cpp | 12 +-- .../format_v2/parquet/parquet_reader_test.cpp | 84 +++++++++++++++++++ 2 files changed, 90 insertions(+), 6 deletions(-) diff --git a/be/src/format_v2/parquet/parquet_reader.cpp b/be/src/format_v2/parquet/parquet_reader.cpp index 82ad3f78891c80..a5ab42ecfcfa7d 100644 --- a/be/src/format_v2/parquet/parquet_reader.cpp +++ b/be/src/format_v2/parquet/parquet_reader.cpp @@ -171,16 +171,16 @@ int64_t count_loaded_non_null_values(const ParquetColumnSchema& root_schema, return count; } - // For repeated encodings, one top-level row starts when the leaf repetition level moves above - // no higher than the top-level container's repeated boundary. Empty MAP/LIST rows have no - // entries but still carry a level slot; they are non-NULL and must be counted by count(col). - const int16_t non_null_definition_level = - static_cast(root_schema.definition_level - 1); + // For repeated encodings, repetition level zero starts a top-level row. Empty MAP/LIST rows + // have no entries but still carry a level slot; they are non-NULL and must be counted by + // count(col). The root nullable level distinguishes a NULL top-level value from a non-NULL + // value regardless of which repeated leaf represents its shape. + const int16_t non_null_definition_level = root_schema.nullable_definition_level; int64_t counted_rows = 0; int64_t non_null_rows = 0; for (int64_t level_idx = 0; level_idx < levels_written && counted_rows < expected_rows; ++level_idx) { - if (rep_levels[level_idx] >= root_schema.repetition_level) { + if (rep_levels[level_idx] != 0) { continue; } ++counted_rows; diff --git a/be/test/format_v2/parquet/parquet_reader_test.cpp b/be/test/format_v2/parquet/parquet_reader_test.cpp index 6269a9fc2a7245..71d1cc291754ae 100644 --- a/be/test/format_v2/parquet/parquet_reader_test.cpp +++ b/be/test/format_v2/parquet/parquet_reader_test.cpp @@ -593,6 +593,48 @@ std::shared_ptr build_nullable_string_struct_array() { return finish_array(&builder); } +std::shared_ptr build_nullable_struct_with_list_array(bool list_first) { + auto list_type = arrow::list(arrow::field("element", arrow::int32(), false)); + auto scalar_field = arrow::field("scalar", arrow::int32(), false); + auto list_field = arrow::field("items", list_type, true); + auto struct_type = arrow::struct_(list_first ? arrow::FieldVector {list_field, scalar_field} + : arrow::FieldVector {scalar_field, list_field}); + + auto scalar_builder = std::make_shared(); + auto list_value_builder = std::make_shared(); + auto list_builder = std::make_shared(arrow::default_memory_pool(), + list_value_builder, list_type); + std::vector> field_builders = + list_first ? std::vector> {list_builder, + scalar_builder} + : std::vector> {scalar_builder, + list_builder}; + arrow::StructBuilder builder(struct_type, arrow::default_memory_pool(), + std::move(field_builders)); + + EXPECT_TRUE(builder.Append().ok()); + EXPECT_TRUE(scalar_builder->Append(1).ok()); + EXPECT_TRUE(list_builder->Append().ok()); + EXPECT_TRUE(list_value_builder->Append(10).ok()); + EXPECT_TRUE(list_value_builder->Append(11).ok()); + + EXPECT_TRUE(builder.AppendNull().ok()); + + EXPECT_TRUE(builder.Append().ok()); + EXPECT_TRUE(scalar_builder->Append(2).ok()); + EXPECT_TRUE(list_builder->AppendEmptyValue().ok()); + + EXPECT_TRUE(builder.Append().ok()); + EXPECT_TRUE(scalar_builder->Append(3).ok()); + EXPECT_TRUE(list_builder->AppendNull().ok()); + + EXPECT_TRUE(builder.Append().ok()); + EXPECT_TRUE(scalar_builder->Append(4).ok()); + EXPECT_TRUE(list_builder->Append().ok()); + EXPECT_TRUE(list_value_builder->Append(20).ok()); + return finish_array(&builder); +} + void write_nullable_map_parquet_file(const std::string& file_path) { auto array = build_nullable_int_string_map_array(); auto field = arrow::field("arr", array->type(), true); @@ -644,6 +686,26 @@ void write_nullable_string_struct_parquet_file(const std::string& file_path) { ROW_COUNT, builder.build())); } +void write_nullable_struct_with_list_parquet_file(const std::string& file_path) { + auto scalar_first = build_nullable_struct_with_list_array(false); + auto list_first = build_nullable_struct_with_list_array(true); + auto table = arrow::Table::Make( + arrow::schema({arrow::field("scalar_first", scalar_first->type(), true), + arrow::field("list_first", list_first->type(), true)}), + {scalar_first, list_first}); + + auto file_result = arrow::io::FileOutputStream::Open(file_path); + ASSERT_TRUE(file_result.ok()) << file_result.status(); + std::shared_ptr out = *file_result; + + ::parquet::WriterProperties::Builder builder; + builder.version(::parquet::ParquetVersion::PARQUET_2_6); + builder.data_page_version(::parquet::ParquetDataPageVersion::V2); + builder.compression(::parquet::Compression::UNCOMPRESSED); + PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), out, + ROW_COUNT, builder.build())); +} + void write_int96_timestamp_parquet_file(const std::string& file_path) { auto field = arrow::field("ts_tz", arrow::timestamp(arrow::TimeUnit::MICRO), true); auto array = @@ -1185,6 +1247,28 @@ TEST_F(NewParquetReaderTest, CountStructColumnUsesLevelsOnlyPath) { EXPECT_EQ(profile.get_counter("MaterializationTime")->value(), 0); } +TEST_F(NewParquetReaderTest, CountStructWithRepeatedChildUsesTopLevelRowBoundaries) { + write_nullable_struct_with_list_parquet_file(_file_path); + + for (int32_t column_id = 0; column_id < 2; ++column_id) { + auto reader = create_reader(); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + ASSERT_TRUE(reader->open(std::make_shared()).ok()); + + format::FileAggregateRequest request; + request.agg_type = TPushAggOp::type::COUNT; + request.columns.push_back({.projection = format::LocalColumnIndex::top_level( + format::LocalColumnId(column_id))}); + format::FileAggregateResult result; + ASSERT_TRUE(reader->get_aggregate_result(request, &result).ok()); + + // Rows are: non-empty ARRAY, NULL STRUCT, empty ARRAY, NULL ARRAY, non-empty ARRAY. + // COUNT(struct) excludes only the NULL STRUCT regardless of child field order. + EXPECT_EQ(result.count, 4); + } +} + TEST_F(NewParquetReaderTest, GetSchemaReturnsNullableNestedChildren) { write_struct_filter_parquet_file(_file_path); auto reader = create_reader();