GH-54533: [C++][Parquet] Avoid emitting empty data pages and advance past empty pages in reader

cpp/src/parquet/arrow/arrow_reader_writer_test.cc

@@ -3144,6 +3144,73 @@ TEST(TestArrowReadWrite, ManySmallLists) {
                        default_arrow_writer_properties());
 }
 
+TEST(TestArrowReadWrite, SingleRowRepeatedSmallPageSize) {
+  // GH-54533: When writing a repeated column with a single row / small page size,
+  // the writer must not emit an empty leading data page, and the reader must
+  // handle data pages without prematurely terminating the column chunk.
+  auto list_type = ::arrow::list(::arrow::boolean());
+  auto array = ArrayFromJSON(list_type, "[[true]]");
+  auto field = ::arrow::field("c0", list_type);
+  auto schema = ::arrow::schema({field});
+  auto table = Table::Make(schema, {std::make_shared<ChunkedArray>(array)});
+
+  auto writer_props = WriterProperties::Builder()
+                          .data_pagesize(64)
+                          ->data_page_version(ParquetDataPageVersion::V2)
+                          ->build();
+  CheckConfiguredRoundtrip(table, table, writer_props);
+}
+
+TEST(TestArrowReadWrite, LastRowGroupSingleRowRepeatedSmallPageSize) {
+  // GH-54533: Last row group holding a single repeated record.
+  auto list_type = ::arrow::list(::arrow::boolean());
+  auto array = ArrayFromJSON(list_type, "[[true], [false], [true]]");
+  auto field = ::arrow::field("c0", list_type);
+  auto schema = ::arrow::schema({field});
+  auto table = Table::Make(schema, {std::make_shared<ChunkedArray>(array)});
+
+  auto writer_props = WriterProperties::Builder()
+                          .data_pagesize(64)
+                          ->data_page_version(ParquetDataPageVersion::V2)
+                          ->build();
+  std::shared_ptr<Table> actual_table;
+  DoRoundtrip(table, /*row_group_size=*/2, &actual_table, writer_props);
+  ::arrow::AssertTablesEqual(*table, *actual_table);
+}
+
+TEST(TestArrowReadWrite, ListDictionarySmallPageSize) {
+  // GH-54533: Dictionary encoding with very small data page size.
+  auto list_type = ::arrow::list(::arrow::int32());
+  auto array = ArrayFromJSON(list_type, "[[1], [2], [3]]");
+  auto field = ::arrow::field("c0", list_type);
+  auto schema = ::arrow::schema({field});
+  auto table = Table::Make(schema, {std::make_shared<ChunkedArray>(array)});
+
+  auto writer_props = WriterProperties::Builder().data_pagesize(4)->build();
+  std::shared_ptr<Table> actual_table;
+  DoRoundtrip(table, /*row_group_size=*/2, &actual_table, writer_props);
+  ::arrow::AssertTablesEqual(*table, *actual_table);
+}
+
+TEST(TestArrowReadWrite, MultiColumnRepeatedSmallPageSize) {
+  // GH-54533: Two columns where one is repeated and page size is small.
+  auto list_type = ::arrow::list(::arrow::boolean());
+  auto f0 = ::arrow::field("id", ::arrow::int64());
+  auto f1 = ::arrow::field("c0", list_type);
+  auto schema = ::arrow::schema({f0, f1});
+
+  auto a0 = ArrayFromJSON(::arrow::int64(), "[1]");
+  auto a1 = ArrayFromJSON(list_type, "[[true]]");
+  auto table = Table::Make(schema, {std::make_shared<ChunkedArray>(a0),
+                                    std::make_shared<ChunkedArray>(a1)});
+
+  auto writer_props = WriterProperties::Builder()
+                          .data_pagesize(64)
+                          ->data_page_version(ParquetDataPageVersion::V2)
+                          ->build();
+  CheckConfiguredRoundtrip(table, table, writer_props);
+}
+
 TEST(TestArrowReadWrite, TableWithDuplicateColumns) {
   // See ARROW-1974
   using ::arrow::ArrayFromVector;

cpp/src/parquet/column_reader.cc

@@ -776,8 +776,8 @@ class ColumnReaderImplBase {
   bool HasNextInternal() {
     // Either there is no data page available yet, or the data page has been
     // exhausted
-    if (num_buffered_values_ == 0 || num_decoded_values_ == num_buffered_values_) {
-      if (!ReadNewPage() || num_buffered_values_ == 0) {
+    while (num_buffered_values_ == 0 || num_decoded_values_ == num_buffered_values_) {
+      if (!ReadNewPage()) {
         return false;
       }
     }

cpp/src/parquet/column_writer.cc

@@ -963,6 +963,9 @@ int64_t ColumnWriterImpl::RleEncodeLevels(const void* src_buffer,
 }
 
 void ColumnWriterImpl::AddDataPage() {
+  if (num_buffered_values_ == 0) {
+    return;
+  }
   int64_t definition_levels_rle_size = 0;
   int64_t repetition_levels_rle_size = 0;
 
@@ -1838,7 +1841,7 @@ class TypedColumnWriterImpl : public ColumnWriterImpl,
     num_buffered_encoded_values_ += num_values;
     num_buffered_nulls_ += num_nulls;
 
-    if (check_page_limit &&
+    if (check_page_limit && num_buffered_values_ > 0 &&
         (current_encoder_->EstimatedDataEncodedSize() >= properties_->data_pagesize() ||
          num_buffered_rows_ >= properties_->max_rows_per_page())) {
       AddDataPage();