-
Notifications
You must be signed in to change notification settings - Fork 4.3k
Arrow-Datasets C++: Scanner::Scan visitor is executing serially despite use_threads=true #49568
Hey together,
I am working with the Apache Arrow C++ Dataset API to scan multiple Parquet files. My goal is to process RecordBatches in parallel using a callback function without materializing the entire table at once.
Following the Dataset Tutorial, I am using the Scan method. According to the documentation:
If multiple threads are used (via use_threads), the visitor will be invoked from those threads and is responsible for any synchronization.
However, in my implementation, the visitor function is called strictly in order—one call only begins after the previous one finishes—even though use_threads is set to true. I have also tried ScanBatchesUnordered, but I am seeing similar serial behavior.
Minimal Working Example:
#include <iostream> #include <memory> #include <arrow/api.h> #include <arrow/compute/api.h> #include <arrow/dataset/api.h> #include <thread> arrow::Status ProcessBatch(const arrow::dataset::TaggedRecordBatch &tagged_batch) { std::cerr << "ThreadId " << std::this_thread::get_id() << " got batch with " << tagged_batch.record_batch->num_rows() << " rows at " << std::chrono::system_clock::now() << "\n"; //Wait: simulate processing time std::this_thread::sleep_for(std::chrono::seconds(5)); return arrow::Status::OK(); } arrow::Status ScanWholeDataset( const std::shared_ptr<arrow::fs::FileSystem> &filesystem, const std::shared_ptr<arrow::dataset::FileFormat> &format, const std::string &base_dir) { // Create custom scan options auto customOption = std::make_shared<arrow::dataset::ScanOptions>(); customOption->use_threads = true; customOption->fragment_readahead = 20; arrow::fs::FileSelector selector; selector.base_dir = base_dir; selector.recursive = true; ARROW_ASSIGN_OR_RAISE( auto factory, arrow::dataset::FileSystemDatasetFactory::Make(filesystem, selector, format, arrow::dataset:: FileSystemFactoryOptions())); ARROW_ASSIGN_OR_RAISE(auto dataset, factory->Finish()); arrow::dataset::ScannerBuilder scan_builder(dataset, customOption); ARROW_ASSIGN_OR_RAISE(auto scanner, scan_builder.Finish()); //Call Scan method with callback function scanner->Scan(ProcessBatch); return arrow::Status::OK(); } arrow::Status Test() { ARROW_RETURN_NOT_OK(arrow::compute::Initialize()); std::string base_path = "xxx"; std::string root_path; std::string uri = "file://xxx"; ARROW_ASSIGN_OR_RAISE(auto fs, arrow::fs::FileSystemFromUri(uri, &root_path)); auto format = std::make_shared<arrow::dataset::ParquetFileFormat>(); ARROW_RETURN_NOT_OK(ScanWholeDataset(fs, format, base_path)); return arrow::Status::OK(); } int main() { auto status = Test(); if (!status.ok()) { std::cerr << "Error: " << status.message() << std::endl; return 1; } return 0; }
Observed Behavior
When running this against 6 Parquet files (approx. 5 GiB total), the timestamps in the output show a perfect 5-second gap between batches.
ThreadId 133083219080896 got batch with 122880 rows at 2026年03月20日 09:25:04.164492317
ThreadId 133083219080896 got batch with 122880 rows at 2026年03月20日 09:25:09.165123649
ThreadId 133083219080896 got batch with 122880 rows at 2026年03月20日 09:25:14.167036139
Apache arrow 22
Am I missing a configuration step in ScanOptions or ScannerBuilder to actually trigger parallel execution of the visitor? Is there a preferred way to handle parallel callbacks in the Dataset API?
Thanks for your help
All reactions
Expanding on @KOKOSde's answer with the specific shape of the API:
Scanner::Scan(visitor) is single-consumer by design — the visitor is serialised across batches regardless of use_threads. use_threads parallelizes everything upstream of your visitor (file open, decompression, decode), and the threads fan back into a single-consumer pipeline where your visitor sees one batch at a time. Your 5-second sleep back-pressures that pipeline, so the decode threads finish early and then idle waiting for your visitor to drain. That's why you're seeing "ThreadId X, ThreadId X, ThreadId X" — there's exactly one consumer thread calling into your visitor.
ScanBatchesUnordered is the same contract with o...
Replies: 2 comments 3 replies
use_threads=true does not mean your visitor body will run in parallel the way a custom thread pool would. Arrow can parallelize file reading and decode, but if the visitor does blocking work, the scan pipeline will still look serial because each callback has to finish before more work can flow through. The practical fix is to keep the visitor tiny, push each RecordBatch into your own queue, and do the heavy processing in a separate pool. If you want direct control, ScanBatchesAsync is a better fit than doing real work inside the visitor.
All reactions
Expanding on @KOKOSde's answer with the specific shape of the API:
Scanner::Scan(visitor) is single-consumer by design — the visitor is serialised across batches regardless of use_threads. use_threads parallelizes everything upstream of your visitor (file open, decompression, decode), and the threads fan back into a single-consumer pipeline where your visitor sees one batch at a time. Your 5-second sleep back-pressures that pipeline, so the decode threads finish early and then idle waiting for your visitor to drain. That's why you're seeing "ThreadId X, ThreadId X, ThreadId X" — there's exactly one consumer thread calling into your visitor.
ScanBatchesUnordered is the same contract with order relaxed. Still single-consumer.
To actually parallelize your processing, decouple
Two patterns work, pick per taste:
1. Consume into a producer/consumer queue
#include <arrow/util/thread_pool.h> auto pool = arrow::internal::GetCpuThreadPool(); auto scanner_reader = ARROW_RESULT(scanner->ToRecordBatchReader()); arrow::Status batch_st; while (true) { std::shared_ptr<arrow::RecordBatch> batch; batch_st = scanner_reader->ReadNext(&batch); if (!batch_st.ok() || batch == nullptr) break; // Fan out the actual work onto the CPU pool. ARROW_RETURN_NOT_OK(pool->Spawn([batch] { // Your ProcessBatch body — runs on whichever pool thread is free. std::cerr << "ThreadId " << std::this_thread::get_id() << " processing " << batch->num_rows() << " rows\n"; std::this_thread::sleep_for(std::chrono::seconds(5)); })); } // Wait for all spawned tasks to complete before returning. pool->WaitForIdle(); return batch_st;
Now the reader thread is the only one calling ReadNext, and the CPU pool runs up to NumCores visitor bodies in parallel. You can cap in-flight batches with a Semaphore if your processing allocates lots of memory.
2. Use ScanBatchesAsync + ApplyCPU
If you want everything inside the Arrow execution model:
auto gen = ARROW_RESULT(scanner->ScanBatchesAsync(pool)); // VisitAsyncGenerator with a parallel transform: arrow::Future<> done = arrow::VisitAsyncGenerator( std::move(gen), [pool](arrow::dataset::TaggedRecordBatch tb) -> arrow::Future<> { return arrow::DeferNotOk(pool->Submit([batch = std::move(tb.record_batch)] { // Your per-batch work. return arrow::Status::OK(); })); }); done.Wait(); return done.status();
This keeps the back-pressure inside Arrow's future machinery; the async generator won't pull a new batch until a previous Future completes, so memory stays bounded even if processing is slow.
The io_context.use_threads gotcha
One more knob that often gets missed: ScanOptions::use_threads controls batch-level parallelism, but I/O parallelism is governed by a separate I/O thread pool that defaults to 8 threads. If your Parquet files are on a slow or remote filesystem, even the decode path can bottleneck there. Check with:
arrow::io::SetIOThreadPoolCapacity(16); // or tune up per your NVMe/S3 setup
Which of these you should pick
sleep_for(5s)in your repro is standing in for CPU work → pattern 1. Most code ends up here.- Your processing is itself async (network calls, another CPU pool) → pattern 2; the async future chain composes cleanly with your existing futures.
- Processing is fast and per-batch, you're just eating I/O latency → the current code is fine; speeding up
use_threads+ I/O pool (above) is all you need.
The key mental model: Arrow's Scanner gives you a stream, not a parallel apply. Parallel apply is your responsibility using one of the two patterns above.
All reactions
Thank you for the explanation; it helped a lot.
I was just confused by the documentation.
All reactions
@moba15 do you think it could be worth to expand the documentation to add what wasn't clear? Your feedback on this would be really valuable to expand what you think was confusing or missing. Thanks!
All reactions
Sorry for the late response:
For me I found it quite confusing that the documentation states
If multiple threads are used (via use_threads), the visitor will be invoked from those threads and is responsible for any synchronization.
for Sanner::Scan.
To me, the phrase "responsible for any synchronization" heavily implied that if use_threads=true, multiple visitor callbacks could execute concurrently at the same time. However, thanks to the explanations here, I now understand that the calls remain strictly sequential (single-consumer), and the warning just means the callback might be invoked from different worker threads across different batches.
It might clarify things for future users if the docs explicitly stated that the visitor is still executed serially. For example, it could be updated to something like:
"If multiple threads are used (via use_threads), the visitor may be invoked from different worker threads. However, the visitor is guaranteed to be called sequentially (one batch at a time). The user is responsible for thread-safety if the visitor shares state across these different thread contexts."