Repository: apache/arrow
File: cpp/src/parquet/column_writer.cc (C++)

Code before:
  while (offset < num_levels) {
    int64_t max_batch_size = std::min(batch_size, num_levels - offset);
    int64_t end_offset = num_levels;           // end offset of the current batch
    int64_t check_page_limit_end_offset = -1;  // offset to check page limit (if not -1)

    int64_t page_buffered_rows = curr_page_buffered_rows();
    ARROW_DCHECK_LE(page_buffered_rows, max_rows_per_page);

    // Iterate rep_levels to find the shortest sequence that ends before a record
    // boundary (i.e. rep_levels == 0) with a size no less than max_batch_size
    for (int64_t i = offset; i < num_levels; ++i) {
      if (rep_levels[i] == 0) {
        // Use the beginning of last record to check page limit.
        check_page_limit_end_offset = i;
        if (i - offset >= max_batch_size || page_buffered_rows >= max_rows_per_page) {
          end_offset = i;
          break;
        }
        page_buffered_rows += 1;
      }
    }

    ARROW_DCHECK_LE(offset, end_offset);
    ARROW_DCHECK_LE(check_page_limit_end_offset, end_offset);

    if (end_offset < num_levels) {
      // This is not the last chunk of batch and end_offset is a record boundary.
      // It is a good chance to check the page limit.
      action(offset, end_offset - offset, /*check_page_limit=*/true);
    } else {
      ARROW_DCHECK_EQ(end_offset, num_levels);
      if (offset <= check_page_limit_end_offset) {
        action(offset, check_page_limit_end_offset - offset, /*check_page_limit=*/true);
        offset = check_page_limit_end_offset;
      }
      if (offset < end_offset) {
        // This is the last chunk of batch, and we do not know whether end_offset is a
        // record boundary so we cannot check page limit if pages cannot change on
        // record boundaries.
        action(offset, end_offset - offset,
               /*check_page_limit=*/!pages_change_on_record_boundaries);
      }
    }

    offset = end_offset;
  }
}

template <typename Action, typename GetBufferedRows>
inline void DoInBatches(const int16_t* def_levels, const int16_t* rep_levels,
                        int64_t num_levels, int64_t batch_size, int64_t max_rows_per_page,

Review comment (performance):
I presume this nesting of `if`s could perhaps be rewritten like this. I don't know if it's more readable but it it shorter and simpler.
```c++
    if (check_page_limit_end_offset >= 0) {
      // At least one record boundary is included in this batch.
      // It is a good chance to check the page limit.
      action(offset, check_page_limit_end_offset - offset, /*check_page_limit=*/true);
      offset = check_page_limit_end_offset;
    }
    if (end_offset > offset) {
      // The is the last chunk of batch, and we do not know whether end_offset is a
      // record boundary so we cannot check page limit if pages cannot change on
      // record boundaries.
      ARROW_DCHECK_EQ(end_offset, num_levels);
      action(offset, end_offset - offset,
             /*check_page_limit=*/!pages_change_on_record_boundaries);
    }
```


Code after:
  while (offset < num_levels) {
    int64_t max_batch_size = std::min(batch_size, num_levels - offset);
    int64_t end_offset = num_levels;           // end offset of the current batch
    int64_t check_page_limit_end_offset = -1;  // offset to check page limit (if not -1)

    int64_t page_buffered_rows = curr_page_buffered_rows();
    ARROW_DCHECK_LE(page_buffered_rows, max_rows_per_page);

    // Iterate rep_levels to find the shortest sequence that ends before a record
    // boundary (i.e. rep_levels == 0) with a size no less than max_batch_size
    for (int64_t i = offset; i < num_levels; ++i) {
      if (rep_levels[i] == 0) {
        // Use the beginning of last record to check page limit.
        check_page_limit_end_offset = i;
        if (i - offset >= max_batch_size || page_buffered_rows >= max_rows_per_page) {
          end_offset = i;
          break;
        }
        page_buffered_rows += 1;
      }
    }

    ARROW_DCHECK_LE(offset, end_offset);
    ARROW_DCHECK_LE(check_page_limit_end_offset, end_offset);

    if (check_page_limit_end_offset >= 0) {
      // At least one record boundary is included in this batch.
      // It is a good chance to check the page limit.
      action(offset, check_page_limit_end_offset - offset, /*check_page_limit=*/true);
      offset = check_page_limit_end_offset;
    }
    if (end_offset > offset) {
      // The is the last chunk of batch, and we do not know whether end_offset is a
      // record boundary so we cannot check page limit if pages cannot change on
      // record boundaries.
      ARROW_DCHECK_EQ(end_offset, num_levels);
      action(offset, end_offset - offset,
             /*check_page_limit=*/!pages_change_on_record_boundaries);
    }

    offset = end_offset;
  }
}

template <typename Action, typename GetBufferedRows>
inline void DoInBatches(const int16_t* def_levels, const int16_t* rep_levels,
                        int64_t num_levels, int64_t batch_size, int64_t max_rows_per_page,
                        bool pages_change_on_record_boundaries, Action&& action,
                        GetBufferedRows&& curr_page_buffered_rows) {
  if (!rep_levels) {
    DoInBatchesNonRepeated(num_levels, batch_size, max_rows_per_page,