Repository navigation
[C++] RecordBatchReader failed when reading parquet file #45116
Description
Describe the bug, including details regarding any error messages, version, and platform.
I tried to use arrow::recoredbatchreader to read multiple rowgroups from a parquet file in parallelism. I use GetRecordBatchReader to acquire recordbatchreader. However, I noticed that when the number of task exceeded the number of cores, the reading would stop at RETURN_NOT_OK(ReadNext(&batch));. The recordbatchreader only works when the number of tasks is less than the number of cores.
And here is my codes:
#include <arrow/api.h>
#include <arrow/io/api.h>
#include <arrow/result.h>
#include <arrow/status.h>
#include <arrow/util/thread_pool.h>
#include <arrow/util/parallel.h>
#include <arrow/util/range.h>
#include <parquet/arrow/reader.h>
#include <parquet/arrow/writer.h>
#include <parquet/column_reader.h>
#include <parquet/column_scanner.h>
#include <parquet/exception.h>
#include <parquet/types.h>
#include <cstring>
#include <chrono>
#include <iostream>
template <class FUNCTION, typename T,
typename R = typename arrow::internal::call_traits::return_type<FUNCTION>::ValueType>
arrow::Future<std::vector<R>> ParallelForAsync_test(
std::vector<T> inputs, FUNCTION&& func,
arrow::internal::Executor* executor = arrow::internal::GetCpuThreadPool()) {
std::vector<arrow::Future<R>> futures(inputs.size());
for (size_t i = 0; i < inputs.size(); ++i) {
ARROW_ASSIGN_OR_RAISE(futures[i], executor->Submit(func, i, std::move(inputs[i])));
}
return All(std::move(futures))
.Then([](const std::vector<arrow::Result<R>>& results) -> arrow::Result<std::vector<R>> {
return arrow::internal::UnwrapOrRaise(results);
});
}
arrow::Result<std::shared_ptr<arrow::Array>> ChunkedArrayToArray(std::shared_ptr<arrow::ChunkedArray> chunked_array) {
auto arrays = chunked_array->chunks();
std::shared_ptr<arrow::Array> result;
ARROW_ASSIGN_OR_RAISE(result, arrow::Concatenate(arrays));
return result;
}
arrow::Status read_whole_file(std::string file, int batch_size, int &size) {
::arrow::MemoryPool* pool = ::arrow::default_memory_pool();
auto reader_properties = parquet::ReaderProperties(pool);
reader_properties.set_buffer_size(4096 * 4);
reader_properties.enable_buffered_stream();
auto arrow_reader_props = parquet::ArrowReaderProperties();
arrow_reader_props.set_batch_size(4 * 1024);
arrow_reader_props.set_use_threads(true);
parquet::arrow::FileReaderBuilder reader_builder;
ARROW_RETURN_NOT_OK(reader_builder.OpenFile(file, false, reader_properties));
reader_builder.memory_pool(pool);
reader_builder.properties(arrow_reader_props);
std::unique_ptr<parquet::arrow::FileReader> arrow_reader;
ARROW_ASSIGN_OR_RAISE(arrow_reader, reader_builder.Build());
auto p_reader = arrow_reader->parquet_reader();
int nrgs = p_reader->metadata()->num_row_groups();
int nrows = p_reader->metadata()->num_rows();
int ncolumns = p_reader->metadata()->num_columns();
auto cpu_executor = ::arrow::internal::GetCpuThreadPool();
int rg_batchsize = nrgs / batch_size;
std::vector<std::shared_ptr<arrow::RecordBatchReader>> vec_reader;
std::cout << "total has " << nrgs << " rowgroups" << std::endl;
for (int j = 0; j <= rg_batchsize; j++) {
std::shared_ptr<arrow::RecordBatchReader> rb_reader;
arrow_reader->GetRecordBatchReader(
arrow::internal::Iota(j * batch_size, std::min((j+1) * batch_size, nrgs)), &rb_reader);
std::cout << "task "<< j << " range [" << j*batch_size << "," << std::min((j+1) * batch_size, nrgs) << "]" << std::endl;
vec_reader.emplace_back(rb_reader);
}
size = vec_reader.size();
std::cout << size << std::endl;
auto thread_start = std::chrono::high_resolution_clock::now();
auto read_recordbatch = [ncolumns, thread_start](size_t i, std::shared_ptr<::arrow::RecordBatchReader> reader)
-> ::arrow::Result<bool>{
auto io_start = std::chrono::high_resolution_clock::now();
auto result = reader->ToTable();
std::vector<std::shared_ptr<::arrow::Array>> vec_array;
if (result.ok()) {
auto table = *result;
for (int i = 0; i < ncolumns; i++) {
auto result = ChunkedArrayToArray(table->column(i));
if (result.ok()) {
auto array = *result;
vec_array.emplace_back(array);
}
}
}
auto io_end = std::chrono::high_resolution_clock::now();
reader->Close();
std::cout << "thread " << i << " " << vec_array[0]->length() << " rows "
<< "start_overhead " << std::chrono::duration<double, std::milli>(io_start-thread_start).count() << " ms "
<< "io_overhead " << std::chrono::duration<double, std::milli>(io_end-io_start).count() << " ms" << std::endl;
return true;
};
auto re = ParallelForAsync_test(std::move(vec_reader), read_recordbatch, cpu_executor)
.MoveResult();
auto re_chka = re.ValueOrDie();
return ::arrow::Status::OK();
}
int main(int argc, char* argv[]) {
std::string file = argv[1];
int batch_size = atoi(argv[2]);
int size = 0;
auto start = std::chrono::steady_clock::now();
read_whole_file(file, batch_size, size);
auto end = std::chrono::steady_clock::now();
printf("%d thread(s) %f ms\n", size, std::chrono::duration<double, std::milli>(end-start).count());
return 0;
}Component(s)
C++
Activity
PPParticle commentedon Dec 28, 2024
The arrow version I use is 14.0.0.
PPParticle commentedon Dec 28, 2024
Furthermore, I tried to increase the capacity of the
thread_pool, and I found that the tasks did not stop but pop asegmentation faultand the tasks did not read any data at all.[-]recordbatchreader failed when reading parquet file[/-][+][C++] recordbatchreader failed when reading parquet file[/+][-][C++] recordbatchreader failed when reading parquet file[/-][+][C++] RecordBatchReader failed when reading parquet file[/+]kou commentedon Dec 29, 2024
Could you try the latest release?
kou commentedon Dec 29, 2024
Could you provide a Parquet file that reproduces this problem?
PPParticle commentedon Dec 29, 2024
I added a predicate statement to make sure the scan ranges valid before
GetRecordBatchReader, which wasif(j * batch_size >= std::min((j + 1) * batch_size, nrgs)) break;. It turned that the scan range that [x,x] was avoided. However, it still did not fix the problem, especially runing the compiled program by./a.out customer.parquet 1.wgtmac commentedon Dec 30, 2024
This line looks weird to me. Please note that GetRecordBatchReader cannot read only partial rows from a single row group. The maximum parallelism is bound by number of row groups in your parquet file.
PPParticle commentedon Dec 30, 2024
I understand what you have said.
The code
int rg_batchsize = nrgs / batch_size;means I mergebatch_sizerowgroups into one recordbatch, and RecordBatchReader would read the rowgroups as one RecordBatch. The reason why I did this was because I could not change the parquet file but I wanted to increase the io granularity each thread.PPParticle commentedon Jan 13, 2025
The problem certainly exits. When the number of potential tasks exceeds the number of cores, the parallelism would stop at somewhere.
stackedsax commentedon Oct 4, 2026
tl;dr: I don't think this is a bug. I reproduced both the hang and the segfault with the sample code on Arrow 24.0.0 (macOS arm64, 14 cores, the attached
customer.parquet), and both come down to how the sample uses the API:RecordBatchReader::ToTable()withuse_threads(true)from inside tasks on the CPU thread pool. The reader fans out onto the same pool and blocks, which deadlocks once the pool is full.<=in the loop produces[62,62]), then indexingvec_array[0]on an empty vector after discarding theConcatenateerror.StatusfromGetRecordBatchReader, which would hide any genuine read error the same way.Details below.
The hang is a nested-parallelism deadlock:
read_recordbatchis submitted toGetCpuThreadPool().ToTable()drives the batch reader. Withset_use_threads(true), each batch callsOptionalParallelForover the columns (cpp/src/parquet/arrow/reader.cc, ~line 1172).ParallelFor(cpp/src/arrow/util/parallel.h) submits one task per column to the same CPU pool, then blocks onfut.status()until they finish.Either of these avoids it, and I verified both:
arrow_reader_props.set_use_threads(false)when you are already parallelising across readers yourself, so the reader doesn't fan out again inside your tasks. All 62 tasks complete on the default pool.GetRecordBatchGeneratoris async and composes with the executor instead of blocking it.The segfault appears as soon as the deadlock is removed (e.g.
SetCapacity(128), which is what the reporter tried), and it's in the sample's lambda:for (j = 0; j <= rg_batchsize; j++)creates one extra reader with the empty range[62,62].ToTable()correctly returns OK with a zero-row table.arrow::Concatenateon its zero-chunk columns returns an error, which the code discards, sovec_arraystays empty.vec_array[0]->length()then dereferences garbage. Under lldb the crash isstd::shared_ptr<arrow::Array>::operator->withthis == nullptr, insideread_whole_file's lambda, not in Arrow.vec_array.empty()(or thebreakthe reporter later added) makes the run complete with exit 0.Two small doc improvements fell out of this, which I'm happy to open a MINOR PR for: the
set_use_threadscomment inproperties.hsays "IO thread pool" but the work goes to the CPU pool, andGetRecordBatchReadercould note that column decoding uses the CPU pool so callers shouldn't block CPU-pool threads on it.@thisisnic I think this can be closed as not an Arrow bug, with the doc fixes as a follow-up.
Note: I used Claude Code to help research and reproduce this, but I read the ticket, the code and the linked commits myself and the conclusion here is mine, not the tool's.
adamreeve commentedon Oct 13, 2026
Thanks for the investigation @stackedsax. I think we should keep this open to cover the documentation updates.