[C++] FileFormat::GetReaderAsync hangs when ARROW_IO_THREADS is set to 1

[yiteng-guo]

Describe the bug, including details regarding any error messages, version, and platform.

# test.py
import pyarrow.dataset as pads

ds = pads.dataset("/tmp/test.parq", format="parquet") # can be any parquet
ds.count_rows(use_threads=False)

ARROW_IO_THREADS=1 python test.py hangs forever. I traced this and it's stuck in GetReader.

arrow::FutureImpl::Wait
> arrow::dataset::ParquetFileFormat::GetReader
> arrow::dataset::ParquetFileFragment::EnsureCompleteMetadata

It works normally if I use any ARROW_IO_THREADS larger than 1. I reproduced this on mac with version 13.0.0 and linux with version 12.0.0.

Component(s)

Python

[yiteng-guo]

It's worth noting that this only happens on count_rows but not on to_table and other pyarrow functions.

[mapleFU]

Hmm let me take a look. Perhaps something is already use ioExecutor and re-allocating sub-task in ioExecutor, which might causing deadlock and hang forever

[mapleFU]

I've found out the reason. But I'm too tired today, I'd like to fix it tomorrow. Or you can try to fix this and at me for review.

Change BufferReader::ReadAsync, and set ARROW_IO_THREADS == 1 would reproduce the problem in dataset C++ unittest.

Future<std::shared_ptr<Buffer>> BufferReader::ReadAsync(const IOContext& ctx,
                                                        int64_t position,
                                                        int64_t nbytes) {
  return DeferNotOk(ctx.executor()->Submit([this, position, nbytes]() {
    return DoReadAt(position, nbytes);
  }));
}

(The origin impl uses a blocking read so test cannot find problem here, it's just a hacking to reproduce the problem)

The IO Thread would wait for it:

Future<std::optional<int64_t>> ParquetFileFormat::CountRows(
    const std::shared_ptr<FileFragment>& file, compute::Expression predicate,
    const std::shared_ptr<ScanOptions>& options) {
  auto parquet_file = checked_pointer_cast<ParquetFileFragment>(file);
  if (parquet_file->metadata()) {
    ARROW_ASSIGN_OR_RAISE(auto maybe_count,
                          parquet_file->TryCountRows(std::move(predicate)));
    return Future<std::optional<int64_t>>::MakeFinished(maybe_count);
  } else {
    return DeferNotOk(options->io_context.executor()->Submit(
        [parquet_file, predicate]() -> Result<std::optional<int64_t>> {
          RETURN_NOT_OK(parquet_file->EnsureCompleteMetadata()); // <- here, it submit a task, and will wait for it in parquet_file->EnsureCompleteMetadata, causing deadlock
          return parquet_file->TryCountRows(predicate);
        }));
  }
}

[mapleFU]

Also cc @westonpace @pitrou , I'd like to implement a blocking read in EnsureCompleteMetadata or async thread, would this be ok? Or do we have better solutions here? (Or IO_THREADS = 1 just a bad idea?)

[mapleFU]

I've implement a draft fixing here: #37514

I'm not sure it can solve all deadlock problem, and I'm a bit tried these days so test might be added later, you can try it yourself.