Skip to content

GH-47027: [C++][Parquet] Fix repeated column pages not being written when reaching page size limit#47032

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
adamreeve merged 2 commits into from
Jul 14, 2025
Merged

GH-47027: [C++][Parquet] Fix repeated column pages not being written when reaching page size limit #47032

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ab580a0
Fix pages not being split for repeated columns
adamreeve Jul 9, 2025
6f677e5
Fix existing test
adamreeve Jul 9, 2025
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
7 changes: 4 additions & 3 deletions cpp/src/parquet/column_writer.cc
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -1176,7 +1176,7 @@ inline void DoInBatches(const int16_t* def_levels, const int16_t* rep_levels,
while (offset < num_levels) {
int64_t end_offset = std::min(offset + batch_size, num_levels);

// Find next record boundary (i.e. ref_level = 0)
// Find next record boundary (i.e. rep_level = 0)
while (end_offset < num_levels && rep_levels[end_offset] != 0) {
end_offset++;
}
Expand All@@ -1196,13 +1196,14 @@ inline void DoInBatches(const int16_t* def_levels, const int16_t* rep_levels,
last_record_begin_offset--;
}

if (offset < last_record_begin_offset) {
if (offset <= last_record_begin_offset) {
// We have found the beginning of last record and can check page size.
action(offset, last_record_begin_offset - offset, /*check_page_size=*/true);
offset = last_record_begin_offset;
}

// There is no record boundary in this chunk and cannot check page size.
// Write remaining data after the record boundary,
// or all data if no boundary was found.
action(offset, end_offset - offset, /*check_page_size=*/false);
}

Expand Down
59 changes: 57 additions & 2 deletions cpp/src/parquet/column_writer_test.cc
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -108,7 +108,7 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
const ColumnProperties& column_properties = ColumnProperties(),
const ParquetVersion::type version = ParquetVersion::PARQUET_1_0,
const ParquetDataPageVersion data_page_version = ParquetDataPageVersion::V1,
bool enable_checksum = false) {
bool enable_checksum = false, int64_t page_size = kDefaultDataPageSize) {
sink_ = CreateOutputStream();
WriterProperties::Builder wp_builder;
wp_builder.version(version)->data_page_version(data_page_version);
Expand All@@ -124,6 +124,7 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
wp_builder.enable_page_checksum();
}
wp_builder.max_statistics_size(column_properties.max_statistics_size());
wp_builder.data_pagesize(page_size);
writer_properties_ = wp_builder.build();

metadata_ = ColumnChunkMetaDataBuilder::Make(writer_properties_, this->descr_);
Expand DownExpand Up@@ -938,6 +939,60 @@ TEST_F(TestByteArrayValuesWriter, CheckDefaultStats) {
ASSERT_TRUE(this->metadata_is_stats_set());
}

// Test for https://github.com/apache/arrow/issues/47027.
// When writing a repeated column with page indexes enabled
// and batches that are aligned with list boundaries,
// pages should be written after reaching the page limit.
TEST_F(TestValuesWriterInt32Type, PagesSplitWithListAlignedWrites) {

@wgtmacwgtmacJul 9, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I may have forgotten: is there any test that covers non-aligned case? If yes, I'm fine with this special case.

@adamreeveadamreeveJul 9, 2025

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

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

Yes I think this is covered by the TestColumnWriter.WriteDataPagesChangeOnRecordBoundariesWithSmallBatches test. One of the columns in that test has some writes that end up aligned with the start of a list, so I had to adjust the expected results, but I think this extra test where all writes are aligned is worth keeping.

this->SetUpSchema(Repetition::REPEATED);

constexpr int list_length = 10;
constexpr int num_rows = 100;
constexpr int64_t page_size = sizeof(int32_t) * 100;

this->GenerateData(num_rows * list_length);

std::vector<int16_t> repetition_levels(list_length, 1);
repetition_levels[0] = 0;

ColumnProperties column_properties;
column_properties.set_dictionary_enabled(false);
column_properties.set_encoding(Encoding::PLAIN);
column_properties.set_page_index_enabled(true);

auto writer =
this->BuildWriter(list_length, column_properties, ParquetVersion::PARQUET_1_0,
ParquetDataPageVersion::V1, false, page_size);

int64_t pages_written = 0;
int64_t prev_bytes_written = 0;

for (int row_idx = 0; row_idx < num_rows; ++row_idx) {
writer->WriteBatch(list_length, def_levels_.data(), repetition_levels.data(),
values_ptr_ + row_idx * list_length);

int64_t bytes_written = writer->total_bytes_written();
if (bytes_written != prev_bytes_written) {
pages_written++;
prev_bytes_written = bytes_written;
}
// Buffered bytes shouldn't grow larger than the specified page size
ASSERT_LE(writer->estimated_buffered_value_bytes(), page_size);
}

writer->Close();

// pages_written doesn't include the last page written when closing the writer:
ASSERT_EQ(pages_written, 9);

this->SetupValuesOut(num_rows * list_length);
definition_levels_out_.resize(num_rows * list_length);
repetition_levels_out_.resize(num_rows * list_length);
this->ReadColumnFully();

ASSERT_EQ(values_out_, values_);
}

TEST(TestPageWriter, ThrowsOnPagesTooLarge) {
NodePtr item = schema::Int32("item"); // optional item
NodePtr list(GroupNode::Make("b", Repetition::REPEATED, {item}, ConvertedType::LIST));
Expand DownExpand Up@@ -1481,7 +1536,7 @@ TEST(TestColumnWriter, WriteDataPagesChangeOnRecordBoundariesWithSmallBatches) {
auto row_group_reader = file_reader->RowGroup(0);

// Check if pages are changed on record boundaries.
const std::array<int64_t, num_cols> expect_num_pages_by_col = {5, 201, 397, 201};
const std::array<int64_t, num_cols> expect_num_pages_by_col = {5, 201, 397, 400};

@adamreeveadamreeveJul 9, 2025

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

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

This column has 150 levels per row and we write 400 rows, 100 levels at a time. The page size is set to 1 byte, so now we get one page written for every logical row.

const std::array<int64_t, num_cols> expect_num_rows_1st_page_by_col = {99, 1, 1, 1};
const std::array<int64_t, num_cols> expect_num_vals_1st_page_by_col = {99, 50, 99, 150};
for (int32_t i = 0; i < num_cols; ++i) {
Expand Down