[C++][Parquet] Should we prevent from writing empty rowgroup in WriteRecordBatch

[mapleFU]

Describe the enhancement requested

Arrow Parquet Writer uses AppendRowGroup to produce the RowGroup during writing a Parquet file. The code path is listed in FileWriterImpl::WriteRecordBatch as below:

    while (offset < batch.num_rows()) {
      const int64_t batch_size =
          std::min(max_row_group_length - row_group_writer_->num_rows(),
                   batch.num_rows() - offset);
      RETURN_NOT_OK(WriteBatch(offset, batch_size));
      offset += batch_size;

      // Flush current row group if it is full.
      if (row_group_writer_->num_rows() >= max_row_group_length) {
        RETURN_NOT_OK(NewBufferedRowGroup());
      }
    }

Assume the max_row_group_length == k, if input recordBatch.num_rows() % k == 0, we would append an empty row-group.

The behavior of empty row-group is a bit tricky. It looks like below. The empty row-group only have metadata, and data does not exists.

     {
       "Id": "2",  "TotalBytes": "0",  "TotalCompressedBytes": "0",  "Rows": "0",
       "ColumnChunks": [
          {"Id": "0", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "1", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "2", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "3", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "4", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "5", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "6", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "7", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "8", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "9", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "10", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" },
          {"Id": "11", "Values": "0", "StatsSet": "False",
           "Compression": "UNCOMPRESSED", "Encodings": "", "UncompressedSize": "0", "CompressedSize": "0" }
        ]
     }

Although it's hard to prevent all cases from "empty row-group", maybe we can prevent the case for the last write to the File

Component(s)

C++, Parquet

[mapleFU]

cc @pitrou @wgtmac

[mapleFU]

The fixing can be as simple as:

diff --git a/cpp/src/parquet/arrow/writer.cc b/cpp/src/parquet/arrow/writer.cc
index 07c627d5e..dd0f5be07 100644
--- a/cpp/src/parquet/arrow/writer.cc
+++ b/cpp/src/parquet/arrow/writer.cc
@@ -462,7 +462,7 @@ class FileWriterImpl : public FileWriter {
       offset += batch_size;
 
       // Flush current row group if it is full.
-      if (row_group_writer_->num_rows() >= max_row_group_length) {
+      if (row_group_writer_->num_rows() >= max_row_group_length && offset < batch.num_rows()) {
         RETURN_NOT_OK(NewBufferedRowGroup());
       }
     }