Arrow-Datasets C++: Scanner::Scan visitor is executing serially despite use_threads=true

[moba15]

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

[KOKOSde]

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.

[MukundaKatta]

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.

[moba15]

Thank you for the explanation; it helped a lot.
I was just confused by the documentation.

[raulcd]

@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!

[moba15]

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."