[C++][Parquet] High memory usage on large Parquet file reading with Project and Filter, with Scanner::Scan()

[alexeyroytman]

Hello.
I'm trying to fulfill a proof-of-concept for Arrow C++ usage for large Parquet file reading with partial-Project and Filter. (I've done the same earlier, with Java code of Apache Parquet project.)
I've seen an example of a code that has ScannerBuilder::Project() and ScannerBuilder::Filter(). In my case, it uses over 300% of CPU, gathers above 30GB of virtual memory, starts thrashing, and at some moment OOM killer kills it.

Some background on the Parquet file contents.

1. For now, I create a Parquet file by myself, and fully control it.
2. Its schema contains of D0..D9 (10 non-nullable strings as BINARY), M0..M4 (5 nullable DOUBLEs).
3. Each of the D(i) is randomly selected from a relatively small subset of strings (up to thousands).
4. M0 has a random value. Other M(j)s are nulls.
5. For D5 and D6 I enable dictionary, statistics and Bloom filter.
6. For M(j)s I disable dictionary, statistics and Bloom filter.
7. The rows are not sorted.
8. The Parquet file is uncompressed.
9. The Parquet file contains 1 billion of rows, its file size is 9.2 GiB. (The file creation time is ~50 minutes.)
10. The original comma-separated file in a tabular form is of 96.5 GiB. (When compressed with "gzip -9", it makes it 8.1 GiB).

The reading context:

1. I project all D(i), M0 and M1 columns. No artificial columns used.
2. I filter on: (D5==s51 || D5==s52) && (D6==s61 || D6==s61). (s51, s52, s61, s62 are all strings matching relevant columns.)
3. The expected number of rows is ~77 million (out of 1 billion).
4. I use ScannerBuilder::Project(), ScannerBuilder::Filter() and then Scanner::Scan().
5. Inside the Scanner::Scan(), sometimes batches have 0 rows, not sure whether this is relevant.

I trace high memory allocations (500 MiB and above), all doing PoolBuffer::Resize() and/or PoolBuffer::Reserve().
I see in the stacktraces:

1. parquet::TypedDecoder<parquet::PhysicalType<(parquet::Type::type)6> >::DecodeArrowNonNull(int, parquet::EncodingTraits<parquet::PhysicalType<(parquet::Type::type)6> >::Accumulator*)
2. TransferColumnData(parquet::internal::RecordReader*, std::unique_ptr<parquet::ColumnChunkMetaData, std::default_delete<parquet::ColumnChunkMetaData> >, std::shared_ptr<arrow::Field> const&, parquet::ColumnDescriptor const*, parquet::arrow::ReaderContext const*, std::shared_ptr<arrow::ChunkedArray>*)

[gistrec]

This looks less like a leak and more like scan concurrency / buffering

Scanner is designed to stream record batches, but with threads and readahead enabled it can still have multiple batches/fragments in flight at the same time. For wide/string-heavy Parquet columns, the decoded in-memory representation can be much larger than the file size

A few things worth trying:

1. Disable threading as a diagnostic check:

ARROW_RETURN_NOT_OK(builder.UseThreads(false));

If RSS drops significantly, then the issue is likely parallel scan buffering rather than the filter itself

2. Iterate batches instead of materializing a large result:

ARROW_ASSIGN_OR_RAISE(auto scanner, builder.Finish());
ARROW_ASSIGN_OR_RAISE(auto it, scanner->ScanBatches());

for (;;) {
  ARROW_ASSIGN_OR_RAISE(auto maybe_batch, it.Next());
  if (!maybe_batch) break;

  const auto& batch = maybe_batch->record_batch;
  // process batch here; don't accumulate all batches unless needed
}

3. Set a smaller batch size:

ARROW_RETURN_NOT_OK(builder.BatchSize(65536));

4. If memory is still high, reduce readahead:

ARROW_RETURN_NOT_OK(builder.FragmentReadahead(1));
// or 0 for the most conservative diagnostic run

This should not prevent normal Parquet predicate/statistics pushdown. The tradeoff is that lower threading/readahead may reduce peak memory at the cost of throughput

[alexeyroytman]

None of the (1), (3), or (4) alone helped. All three ((1), (3), and (4)) also did not help.

When I've implemented (2) with ScanBatches, with or without (1), (3), and (4), the same issue of high memory usage occurred.

[alexeyroytman]

In my case, without ARROW_DEFAULT_MEMORY_POOL=system the too high memory usage is reached before Scan/ScanBatches. Not sure this is relevant.