GH-37487: [C++][Parquet] Dataset: Implement sync ParquetFileFormat::GetReader

[mapleFU]

Rationale for this change

As #37487 says. When thread cnt == 1, the thread might blocking in ParquetFileFormat::GetReaderAsync, that's because:

1. ParquetFileFormat::CountRows would call EnsureCompleteMetadata in io_executor
2. EnsureCompleteMetadata call ParquetFileFormat::GetReader, which dispatch real request to async mode
3. async is executed in io_executor.

1/3 in same fix-sized executor, causing deadlock.

What changes are included in this PR?

Implement sync ParquetFileFormat::GetReader.

Are these changes tested?

Currently not

Are there any user-facing changes?

Bugfix

• Closes: [C++] FileFormat::GetReaderAsync hangs when ARROW_IO_THREADS is set to 1 #37487

[github-actions]

⚠️ GitHub issue #37487 has been automatically assigned in GitHub to PR creator.

[mapleFU]

I think we already have the test for GetReader

[mapleFU]

@pitrou @westonpace would you mind take a look?

[mapleFU]

@westonpace By the way, call Blocking Get in same thread pool might cause the deadlock, should we try to find out which would causing this and trying to fix them?

[westonpace]

I'll try and look in more detail later today but here is the general advice.

Async methods which return a future.
Sync methods do not

User calls always start out as synchronous methods (e.g. we don't have adapters for python's asyncio)

We can convert from synchronous to asynchronous using something like...

... Foo(...) {
    Future<...> fut = FooAsync(...);
    return fut.result();
}

This is ok. One conversion from synchronous to asynchronous is ok and inevitable (since user calls start as sync).

However, it is not ok to convert from asynchronous to synchronous. For example, if Foo is defined as above then this would not be ok:

... BarAsync(...) {
   Future<...> fut = SomeAsyncMethod();
   return fut.Then([&] (...) {
       return Foo(...);
   });
}

This is also not ok:

... BarAsync(...) {
    auto x = Foo(...);
    return SomeAsyncMethod(x, ...);
}

Normally the fix is for BarAsync to call FooAsync.

Creating a Bar that is a synchronous method that mirrors BarAsync is ok, but leads to more code.

At a glance, it looks like that is what is happening here, which is maybe ok. However, I have to look in more detail.

[mapleFU]

Thanks for your nice advice!

I think this patch would be OK, I can check more Async usage around dataset.

[mapleFU]

@westonpace can we try to review this patch first? Since I believe at least this patch might solve the problem :-)

[mapleFU]

@pitrou Would you mind help take a look? Since the logic is easy and simple here...

[mapleFU]

ping @pitrou @westonpace @bkietz for help...

[bkietz]

This isn't tested and we don't have follow up from the original issue's poster that this patch fixes the deadlock.

@mapleFU could you add (something like) the original repro as a test? It'd be best if we can confirm that main hangs with a single IO thread and use_threads=False, then confirm that we no longer hang with this fix.

# 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

@yiteng-guo could you try out this patch and confirm it fixes the deadlock for you?

[mapleFU]

Hi @bkietz , I've write a test using a extended file-reader, it can reproduce the case.

class AsyncBufferReader : public ::arrow::io::BufferReader {
 public:
  explicit AsyncBufferReader(std::shared_ptr<Buffer> buffer, const ::arrow::io::IOContext& ctx): ::arrow::io::BufferReader(std::move(buffer)), ctx_(ctx) {}
  /// EXPERIMENTAL: Read data asynchronously.
  Future<std::shared_ptr<Buffer>> ReadAsync(const ::arrow::io::IOContext& ctx, int64_t position,
                                            int64_t nbytes) override {
    auto self = checked_pointer_cast<AsyncBufferReader>(shared_from_this());
    return DeferNotOk(ctx.executor()->Submit([self, position, nbytes]() {
      return self->ReadAt(position, nbytes);
    }));
  }

  const ::arrow::io::IOContext& io_context() const override {
    return ctx_;
  }

  private:
   ::arrow::io::IOContext ctx_;
};

However, default io-pool is widely used in scanner, it's a bit hard to hook all io_context( Might need to change EnsureCompleteMetadata to below) :

Status EnsureCompleteMetadata(parquet::arrow::FileReader* reader = NULLPTR, const std::shared_ptr<ScanOptions>& scan_options=NULLPTR);

So I change the global thread pool during testing, would you mind take a look?

[bkietz]

I think you can use SetIOThreadPoolCapacity and GetIOThreadPoolCapacity here:

Suggested change

	::arrow::io::IOContext default_io_context;
	auto pool_executor =
	dynamic_cast<::arrow::internal::ThreadPool*>(default_io_context.executor());
	if (pool_executor == nullptr) {
	GTEST_SKIP();
	}
	auto origin_capacity = pool_executor->GetCapacity();
	ASSERT_OK(pool_executor->SetCapacity(/*threads=*/1));
	// Reset capacity for pool_executor
	struct PoolResetGuard {
	PoolResetGuard(::arrow::internal::ThreadPool* pool, int origin_capacity)
	: pool(pool), origin_capacity(origin_capacity) {}
	~PoolResetGuard() {
	Status s = pool->SetCapacity(origin_capacity);
	if (!s.ok()) {
	std::cerr << "Failed to reset pool capacity: " << s.ToString() << std::endl;
	}
	}
	::arrow::internal::ThreadPool* pool;
	int origin_capacity;
	};
	PoolResetGuard guard(pool_executor, origin_capacity);
	// Reset capacity for pool_executor
	struct PoolResetGuard {
	int original_capacity = io::GetIOThreadPoolCapacity();
	~PoolResetGuard() {
	DCHECK_OK(io::SetIOThreadPoolCapacity(original_capacity));
	}
	} guard;

Since this directly mutates the IO thread pool (pointed to by the default IO context), you shouldn't need AsyncBufferReader::ReadAsync

I agree that it's odd BufferReader::ReadAsync and MemoryMappedFile::ReadAsync (the only overrides I see) ignore the io context argument. @lidavidm do you think they should transfer to the other executor if it's different?

[mapleFU]

Aha, firstly I'm trying to change ParquetFileFragment uses same io-executor, but I guess it should be put in another patch.

[mapleFU]

I agree that it's odd BufferReader::ReadAsync and MemoryMappedFile::ReadAsync (the only overrides I see) ignore the io context argument

I think BufferReader::ReadAsync is so lightweight that we can ignore ioContext?

[lidavidm]

Given both access memory/"memory" I think that's quite intentional

[mapleFU]

@bkietz I've tried to resolve the comments, would you mind take a look again?

[bkietz]

LGTM, thanks!

[conbench-apache-arrow]

After merging your PR, Conbench analyzed the 6 benchmarking runs that have been run so far on merge-commit 76c4a6e.

There were no benchmark performance regressions. 🎉

The full Conbench report has more details. It also includes information about possible false positives for unstable benchmarks that are known to sometimes produce them.