GH-39808: [C++][Parquet] Fix cross-row-group test cache limits

The cross-row-group eviction test set range_size_limit == hole_size_limit, which trips the range_size_limit > hole_size_limit check in debug/CI builds (the release build compiles it out). Make range one byte larger so the limits stay valid while still coalescing every chunk into one entry. Also trim the remaining added comments to bare invariants per review feedback.

Author: justinli500

cpp/src/arrow/io/caching.cc

@@ -265,9 +265,8 @@ struct ReadRangeCache::Impl {
     return AllComplete(futures);
   }
 
-  // `entries` is sorted by offset, so we stop once an entry starts at/after
-  // `end_offset`. An entry that starts before but extends past `end_offset`
-  // (coalesced with a range a later consumer still needs) is left in place.
+  // `entries` is sorted by offset; an entry that extends past `end_offset`
+  // (coalesced with a range a later consumer still needs) is kept.
   int64_t EvictEntriesBefore(int64_t end_offset) {
     int64_t n_evicted = 0;
     std::unique_lock<std::mutex> guard(entry_mutex);

cpp/src/arrow/io/caching.h

@@ -142,16 +142,10 @@ class ARROW_EXPORT ReadRangeCache {
   /// \brief Wait until all given ranges have been cached.
   Future<> WaitFor(std::vector<ReadRange> ranges);
 
-  /// \brief Evict cache entries that end at or before `end_offset`.
-  ///
-  /// Releases the memory of entries whose byte range lies entirely before
-  /// `end_offset`. An entry straddling `end_offset` is retained, so this is safe
-  /// even when I/O coalescing merged several requested ranges into one entry.
-  /// Buffers already returned by Read() stay valid through shared ownership.
-  ///
-  /// \param[in] end_offset Exclusive byte bound; entries ending at or before it
-  ///     are evicted.
-  /// \return Number of cache entries evicted.
+  /// \brief Evict cache entries ending at or before `end_offset`, returning the
+  /// count. An entry straddling `end_offset` is retained (safe when coalescing
+  /// merged ranges). Buffers already returned by Read() stay valid through
+  /// shared ownership.
   int64_t EvictEntriesBefore(int64_t end_offset);
 
  protected:

cpp/src/parquet/arrow/arrow_reader_writer_test.cc

@@ -2673,10 +2673,8 @@ TEST(TestArrowReadWrite, GetRecordBatchReaderNoColumns) {
   ASSERT_EQ(actual_batch->num_rows(), num_rows);
 }
 
-// GH-39808: bytes cached by PreBuffer() for an already-decoded row group
-// should be released when the caller explicitly evicts them, otherwise
-// Dataset.to_batches accumulates memory across the lifetime of an open
-// FileReader.
+// GH-39808: bytes cached by PreBuffer() for a decoded row group must be
+// releasable, else Dataset.to_batches accumulates memory over the reader's life.
 TEST(TestArrowReadWrite, EvictPreBufferedDataBefore) {
   ArrowReaderProperties properties = default_arrow_reader_properties();
   properties.set_pre_buffer(true);
@@ -2758,7 +2756,7 @@ TEST(TestArrowReadWrite, EvictPreBufferedDataBeforeReleasesCrossRowGroupEntry) {
   // spanning all row-group boundaries.
   ::arrow::io::CacheOptions options = ::arrow::io::CacheOptions::LazyDefaults();
   options.hole_size_limit = static_cast<int64_t>(buffer->size());
-  options.range_size_limit = static_cast<int64_t>(buffer->size());
+  options.range_size_limit = static_cast<int64_t>(buffer->size()) + 1;
   reader->parquet_reader()->PreBuffer(row_groups, column_indices,
                                       ::arrow::io::IOContext(), options);
   ASSERT_OK(reader->parquet_reader()->WhenBuffered(row_groups, column_indices).status());

cpp/src/parquet/arrow/reader.cc

@@ -1239,10 +1239,9 @@ FileReaderImpl::GetRecordBatchGenerator(std::shared_ptr<FileReader> reader,
                        reader_properties_.cache_options());
     END_PARQUET_CATCH_EXCEPTIONS
   }
-  // GH-39808: when pre-buffering, release each row group's cached bytes once the
-  // contiguous prefix of decoded row groups no longer needs them, keeping the
-  // memory footprint bounded while iterating. Confined to this read-once path,
-  // so PreBuffer()'s contract for other callers is unchanged.
+  // GH-39808: evict each row group's bytes as the decoded prefix advances, so
+  // memory stays bounded. Only this read-once path evicts, so PreBuffer()'s
+  // contract is unchanged for other callers.
   std::shared_ptr<ReadCacheEvictionState> eviction_state;
   if (reader_properties_.pre_buffer() && !column_indices.empty() &&
       !row_group_indices.empty()) {

cpp/src/parquet/arrow/reader_internal.h

@@ -138,13 +138,9 @@ Status TransferColumnData(::parquet::internal::RecordReader* reader,
 // ----------------------------------------------------------------------
 // Pre-buffer eviction
 
-// GH-39808: releases pre-buffered cache entries as row groups are decoded.
-// RowGroupDecoded() is called (possibly out of order, from readahead
-// continuations) once a row group's batches are produced. It advances a
-// watermark over the contiguous prefix of completed row groups and evicts every
-// cache entry ending before the lowest byte any not-yet-completed row group
-// still needs. A coalesced entry spanning a row-group boundary is freed once
-// the watermark passes its end.
+// GH-39808: as row groups finish decoding (possibly out of order under
+// readahead), advance a watermark over the leading run of completed ones and
+// evict cache entries ending before the lowest byte any remaining one needs.
 class ReadCacheEvictionState {
  public:
   // evict_before_offsets[i] = lowest byte offset row groups i..n-1 (in
@@ -159,11 +155,8 @@ class ReadCacheEvictionState {
     }
   }
 
-  // Marks row_group_index decoded and advances the contiguous completed prefix.
-  // Returns the offset to evict before when the prefix advanced, or std::nullopt
-  // when nothing newly became evictable (index out of range, already completed,
-  // or the prefix did not grow). Separated from RowGroupDecoded so the
-  // out-of-order watermark logic is unit-testable without a reader.
+  // Reader-free half of RowGroupDecoded (so the out-of-order advance is testable):
+  // the evict-before offset when the prefix grows, else nullopt.
   std::optional<int64_t> MarkDecodedAndGetEvictOffset(size_t row_group_index) {
     std::lock_guard<std::mutex> lock(mutex_);
     if (row_group_index >= completed_.size() || completed_[row_group_index]) {

cpp/src/parquet/file_reader.h

@@ -201,13 +201,9 @@ class PARQUET_EXPORT ParquetFileReader {
                  const ::arrow::io::IOContext& ctx,
                  const ::arrow::io::CacheOptions& options);
 
-  /// \brief Release cached bytes that end at or before `end_offset`.
-  ///
-  /// Only affects data cached by PreBuffer(). Call once the row groups needing
-  /// those bytes are fully decoded; later reads of evicted ranges may fail.
-  /// Buffers already handed to readers stay valid through shared ownership. A
-  /// no-op (returns 0) if PreBuffer() was not called. Returns the number of
-  /// cache entries evicted.
+  /// \brief Release cached bytes (from PreBuffer()) ending at or before
+  /// `end_offset`. Call once those row groups are decoded; later reads of evicted
+  /// ranges may fail. No-op (returns 0) if PreBuffer() was not called.
   int64_t EvictPreBufferedDataBefore(int64_t end_offset);
 
   /// Retrieve the list of byte ranges that would need to be read to retrieve