[C++][FlightRPC][Python] Async DoGet (and DoAction) in the Flight client #55452

Describe the enhancement requested

The Flight client has an async API for one call only: GetFlightInfoAsync in C++
(cpp/src/arrow/flight/client.h), which pyarrow.flight.AsyncioFlightClient wraps through
FlightClient.as_async() (added in #37093).

DoGet, which streams the actual data, has no async version in C++ or Python. Neither does
DoAction.

We request:

Priority 1: async DoGet. An async DoGet in the C++ client that delivers record batches as
they arrive, with a way to slow down the stream when the reader cannot keep up. And a matching
method on AsyncioFlightClient, for example async def do_get(ticket, options) that returns an
async iterator of RecordBatch (or a reader with an awaitable read_next_batch() / read_all()).
DoGet holds a thread for the whole stream. In a load test it used about twice the total thread
time of all our DoAction calls.

Priority 2: async DoAction. The same in C++ and on AsyncioFlightClient. Each action is
short, but in the same load test we made about 8 times more DoAction calls than DoGet calls,
mostly to poll job status. Together they still used a large part of the thread pool. With both calls
async, we could drop the thread pool entirely.

Priority 1 does not need to wait for Priority 2.

Use case

We call a Flight server from an asyncio Python service. Under peak load it runs about 150
concurrent requests against the server. Today each do_get has to run in a thread pool
(loop.run_in_executor) to avoid blocking the event loop, so the pool must be sized for that peak:
one thread for each concurrent request. This means:

• the thread pool size limits how many requests can be in flight at once;
• cancelling the asyncio task does not cancel the RPC — the thread keeps reading until the stream
  ends or the timeout expires;
• one deadline for the whole read has to be split by hand between get_flight_info and do_get.

With as_async() we can already make get_flight_info async, but do_get usually takes most of
the time, so the thread pool is still needed.

Alternatives considered

• Running the blocking client in a thread pool (what we do now).
• Dropping the pyarrow.flight client and talking to the server directly with grpc.aio. This
  means building our own Flight client from scratch: generating stubs from Flight.proto, splitting
  each FlightData message into its IPC header and body and decoding them with pyarrow.ipc,
  tracking the schema across the stream, and re-implementing call headers, auth, TLS, timeouts and
  error mapping. That is a lot of low-level code to write and maintain, and it duplicates what
  the Flight client already does.

Related: #37093 (async GetFlightInfo). A 2024 comment on that issue also asked about async
DoExchange.

Environment

pyarrow 23.0.1, Python 3.14, Linux, gRPC transport.

Component(s)

C++, FlightRPC, Python