GH-48636: [C++][Parquet] Improve parquet reading using multi threads

[OmBiradar]

Rationale for this change

Currently the parquet file structs are read sequentially, even when the use_threads option is used by the user. This PR aims to bridge this gap by making the struct reading truly parallel.

What changes are included in this PR?

Changes to the LoadBatch and BuildArray functions in the StructReader in the Parquet reader to enable multi threaded reads of structs in parquet files.

Are these changes tested?

• Yes I have tested them. Details can be found in this testing and benchmarking repo
• Weather benchmarks should be added for parallel reads is a question I would ask the reviewer.

Are there any user-facing changes?

This is purely a performance related PR. No changes to the user or API is needed.

• GitHub Issue: [Python] Enable multi-threaded reads of struct-list stored data in parquet files #48636

[github-actions]

Thanks for opening a pull request!

If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose

Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project.

Then could you also rename the pull request title in the following format?

GH-${GITHUB_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}

or

MINOR: [${COMPONENT}] ${SUMMARY}

See also:

• Other pull requests
• Contribution Guidelines - Contributing Overview

[Copilot]

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

Updates Parquet testing submodule pins and introduces optional parallelism when loading/building nested struct children readers to improve throughput when use_threads() is enabled.

Changes:

• Bump testing and cpp/submodules/parquet-testing submodule commits.
• Use ::arrow::internal::OptionalParallelFor to parallelize StructReader::LoadBatch.
• Parallelize StructReader::BuildArray child array construction before converting chunks to single arrays.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
testing	Updates submodule commit SHA.
cpp/submodules/parquet-testing	Updates submodule commit SHA for parquet-testing.
cpp/src/parquet/arrow/reader.cc	Adds optional parallel execution for struct child LoadBatch and BuildArray.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[OmBiradar]

template <class FUNCTION>
Status OptionalParallelFor(bool use_threads, int num_tasks, FUNCTION&& func,
                           Executor* executor = internal::GetCpuThreadPool()) {
  if (use_threads) {
    return ParallelFor(num_tasks, std::forward<FUNCTION>(func), executor);
  } else {
    for (int i = 0; i < num_tasks; ++i) {
      RETURN_NOT_OK(func(i));
    }
    return Status::OK();
  }
}

At the end, it's always cast to an int so truncating is unavoidable.

[OmBiradar]

template <class FUNCTION>
Status OptionalParallelFor(bool use_threads, int num_tasks, FUNCTION&& func,
                           Executor* executor = internal::GetCpuThreadPool()) {
  if (use_threads) {
    return ParallelFor(num_tasks, std::forward<FUNCTION>(func), executor);
  } else {
    for (int i = 0; i < num_tasks; ++i) {
      RETURN_NOT_OK(func(i));
    }
    return Status::OK();
  }
}

At the end, it's always cast to an int so truncating is unavoidable.

[OmBiradar]

@wgtmac @pitrou

Should I add benchmarks for the multi threaded reading scenario, I am asking as I have not seen any other multi threaded performance benchmarks.

[wgtmac]

TBH, I don't think it is a good approach as we've tried this in the past. The main gotcha is that reading costs of different columns vary significantly by nature. For example, strings take longer time to decompress and decode but integers are smaller and faster. If the file is on a cloud object store, the majority time is blocked on waiting for I/O which may exhaust the thread pool if it is a wide column file.

[OmBiradar]

Yes, even I faced the same IO limitation while testing the changes.

But as this is only triggered when the user wants multi threaded reading. When the file is close to the processor i.e in memory there is a significant improvement (considering a normal case). I personally tested it by loading the whole file into memory and then benchmarking the reading.

An edge case like a heavy column (I mean w.r.t the computing time) the delay is inevitable unless there is a complex system to manage the building and loading of a single column using multiple threads.

I would say that this change would not be the best case but give workable results?

It would always beat a single thread I think? Only exception I can think is of having many many light weight columns in the struct, then maybe the overhead of syncing the threads might slow down the multi threaded approach over the single threaded one.

Could you suggest any other approach that I am not able to think about?

[hombit]

@wgtmac could you please explain your argument a little bit? I'm not really getting it. From my understanding, what this PR is doing is making the read of a Parquet file with one or a few wide nested columns be parallelized in the same way as reading a Parquet file with exactly the same data but with "flattened" columns.
@OmBiradar is working on this from the perspective of better performance of local reads of astronomical data. See dataset examples here and here.

[wgtmac]

@hombit My explanation comes more from an IO-bound workload. If there is a valid CPU-bound use case, I think it is fine to add this as an improvement. So the implication is that we require thread-safety across different columns readers if this PR gets merged. For example, a custom ArrowInputStream implementation that supports I/O prefetching now needs to consider the new pattern.

[OmBiradar]

Hey @wgtmac

we require thread-safety across different columns readers if this PR gets merged

I will look into this

a custom ArrowInputStream implementation that supports I/O prefetching now needs to consider the new pattern.

Thanks for suggesting a starting position.

[wgtmac]

@OmBiradar I don't mean we need to provide a custom ArrowInputStream implementation in this library because it is really difficult to satisfy different prefetching requirements specific to application use cases. This is the flexibility that downstream users are able to plugin their own ArrowInputStream implementations to do I/O prefetching so we need to know the new constraint that imposed on their custom implementations.

Instead, could you fix CI failures? Some parquet test failures are related.

[OmBiradar]

@wgtmac

we need to know the new constraint that imposed on their custom implementations.

Yes I got you. I understand the point here is that the downstream users don't face any problems related to this, and if there is some constraint, they should know about it.

Instead, could you fix CI failures? Some parquet test failures are related.

Yes I saw them.

Quit a lot of CI are failing, initially I thought I had done something very wrong, but I see CI failing in the main branch of arrow itself.

Will try to solve the failing CI.

[OmBiradar]

hey guys!

I know it's been a long time since I have made the changes. The thing is I am new to apache arrow, thus I am still understanding the codebase while trying to contribute to it.

I just wanted to say that I am actively working on it.

[wgtmac]

I see there are still failing CIs. Could you please fix them?

[OmBiradar]

Yes.
The CI failure it's due to arrow-dataset-file-parquet-test failing due to timeout, I am working on it.

[OmBiradar]

Hey @wgtmac
i looked into the failing test, which is specifically - TestParquetFileFormatScan.ScanRecordBatchReaderProjectedNested/0Threaded16b1024r This test requires reading of a nested parquet file having struct columns.
I used gdb to obtain a backtrace after I was confident the program has hit a "deadlock" of some sort.
Analysing the traceback, I found that the

1. threads which are suppose to read the structs, hand off the reading of its children fields to other threads and go into a "wait" state.
2. The child field reading threads cannot execute because the thread pool is fully saturated with threads which are on "wait"

This creates the deadlock due to the cross dependency between threads and threads spend time waiting on each other.
Thus, I believe this is a sync-async problem, where generally a blocking thread should not spawn other threads and wait on them. Here a async type thread management would be nice.
As there is also a note in the cpp/src/parquet/arrow/reader.cc file in 80fe83a by lidavidm and pitrou where it says

Making the Parquet reader truly asynchronous requires heavy refactoring, so the generator piggybacks on ReadRangeCache.

I believe that it enables the multi-threaded reading of row groups, but it does not consider threads producing new threads to read various fields in a struct.
I really don't have much idea on how to approach this, could you please provide any help @wgtmac

Could making arrow parquet reader truly async be a possible solution?

[wgtmac]

Thanks for the investigation. I agree this is a real issue, and it is not easy to fix correctly. Making the Parquet reader “truly async” would be a much larger design and engineering effort without breaking existing API. I don't think that is something we should try to solve opportunistically in this PR.

For now, I think we should keep the existing async implementation as-is and avoid changing the current scheduling model in StructReader. So my suggestion is to abandon this PR, unfortunately.

[OmBiradar]

Hey @wgtmac could I try to make the parquet reader "truely async" in a seperate PR?

[wgtmac]

Hey @wgtmac could I try to make the parquet reader "truely async" in a seperate PR?

TBH, I don't think this is a good-first-issue due to its complexity. A lot of time will be spent on addressing review comments back and forth. :)

[OmBiradar]

I am actually a GSoC mentee for this summer. I still have 2 months left to complete. I could make really good progress :)

@wgtmac

[wgtmac]

If you have only two months, I'd encourage you to work on something smaller. Don't mean to scare you, but large features like this may take months or even years before merging. You may easily find some clues from merged PRs of Parquet. That said, you're still welcome to try it out provided that your approach does not introduce breaking change.

[OmBiradar]

Thanks for the heads up