GH-37937: [C++][FlightRPC] Investigate using gRPC's generic API using gRPC's BidiReactor

[raulcd]

Warning

Do not merge, this is a PoC being discussed at the moment

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

This PR includes breaking changes to public APIs. (If there are any breaking changes to public APIs, please explain which changes are breaking. If not, you can remove this.)

This PR contains a "Critical Fix". (If the changes fix either (a) a security vulnerability, (b) a bug that caused incorrect or invalid data to be produced, or (c) a bug that causes a crash (even when the API contract is upheld), please provide explanation. If not, you can remove this.)

• GitHub Issue: [C++][FlightRPC] Investigate using generic callback stub to bypass Protobuf serialization #37937

[github-actions]

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

[raulcd]

Some initial discussion happened on this same branch on a PR my fork, see more details here:
raulcd#97 (comment)

I am moving the PR to here for visibility.

[raulcd]

@lidavidm @pitrou I've spent some time today learning more about our FlightDataSerialize method and how we could expose an API that leaves gRPC out of the picture using a BufferVector instead. This is only for the write path so far but I wanted to share early to validate this is in-line with what we had been discussing.

[raulcd]

@lidavidm @pitrou I've moved the serialization / Deserialization logic from arrow/flight/transport/grpc/serialization_internal.cc to arrow/flight/serialization_internal.cc.
Those are now serializing a FlightPayload to arrow::BufferVector or deserializing from an arrow::Buffer to FlightData. Those are still supposed to be internal:

/// \brief Serialize a FlightPayload to a vector of buffers.
///
/// The first buffer contains the protobuf wire format header. Subsequent
/// buffers are zero-copy references to the IPC body buffers, with padding
/// buffers inserted as needed for 8-byte alignment.
arrow::Result<arrow::BufferVector> SerializePayloadToBuffers(
    const FlightPayload& payload);

/// \brief Deserialize FlightData from a contiguous buffer.
arrow::Result<FlightData> DeserializeFlightData(
    const std::shared_ptr<arrow::Buffer>& buffer);

The external API to be used is either FlightPayload::SerializeToBuffers() or a new FlightMessageDecoder::Consume(std::shared_ptr<Buffer> buffer).
The gRPC transport side serialization/deserialization is still handled on arrow/flight/transport/grpc/serialization_internal.cc but those are just tiny wrappers for arrow/flight/serialization_internal.cc only managing the grpc::Slices / ByteBuffer on top of it.

I've adapted the gRPC Async PoC to show how this would work with the bidi reactor.

The only point where I am not entirely clear is whether utilities specific to gRPC should be exposed. This matters for users building their own bidi reactors (see the async grpc example on the PR). If our stand-point is that users
should manage gRPC themselves then those users need a way to convert between arrow::Buffer and grpc::ByteBuffer/Slice which isn't trivial.

/// Convert an Arrow Buffer to a gRPC Slice.
arrow::Result<::grpc::Slice> SliceFromBuffer(const std::shared_ptr<arrow::Buffer>& buf);

/// Wrap a gRPC ByteBuffer as a zero-copy Arrow Buffer (and clear the ByteBuffer).
arrow::Status WrapGrpcBuffer(::grpc::ByteBuffer* grpc_buf,
                             std::shared_ptr<arrow::Buffer>* out);

I'd like to move this forward, maybe an initial PR only moving the serialization / deserialization logic to handle arrow::Buffer/BufferVector instead of grpc::ByteBuffers/Slice is the best approach?