GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange)

[raulcd]

Rationale for this change

The Flight Server DoExchange method currently does not support Dictionary replacement or Dictionary Deltas, similar to how the client currently behaves or how we do for DoGet we should use an ipc::RecordBatchWriter with a custom IpcPayloadWriter instead of reimplementing Dictionary Replacement / Deltas logic.

What changes are included in this PR?

Removes manually generation of individual ipc Payloads and uses an ipc::RecordBatchWriter and a custom TransportMessagePayloadWriter to modify the IpcPayloads into FlightPayloads.

Are these changes tested?

Yes, existing tests cover the DoExchange functionality and new test for Python has been added where Dictionary deltas are being send via DoExchange. The test was failing before this change because the dictionary wasn't updated:

            received_table = reader.read_all()
            expected_table = simple_dicts_table()
>           assert received_table.equals(expected_table)
E           assert False
E            +  where False = equals(pyarrow.Table\nsome_dicts: dictionary<values=string, indices=int64, ordered=0>\n----\nsome_dicts: [  -- dictionary:\n["foo... -- dictionary:\n["foo","baz","quux"]  -- indices:\n[2,1],  -- dictionary:\n["foo","baz","quux","new"]  -- indices:\n[0,3]])
E            +    where equals = pyarrow.Table\nsome_dicts: dictionary<values=string, indices=int64, ordered=0>\n----\nsome_dicts: [  -- dictionary:\n["foo...ull],  -- dictionary:\n["foo","baz","quux"]  -- indices:\n[2,1],  -- dictionary:\n["foo","baz","quux"]  -- indices:\n[0,3]].equals

pyarrow/tests/test_flight.py:2596: AssertionError
================================================================================== short test summary info ===================================================================================
FAILED pyarrow/tests/test_flight.py::test_flight_dictionary_deltas_do_exchange - assert False

Are there any user-facing changes?

No, only that the expected dictionary replacement/deltas will work for DoExchange.

• GitHub Issue: [C++][Flight] DoExchange doesn't support dictionary replacement #26727

[github-actions]

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

[raulcd]

Current CI failures are unrelated and tracked here:

• [C++] AMD64 macOS 13 C++ and ARM64 macOS 14 C++ are failing due to no type named 'int64' in namespace 'google::protobuf' #47399
• [GLib] _LIBCPP_ENABLE_ASSERTIONS is deprecated, please use _LIBCPP_HARDENING_MODE instead on macOS 15 #46690

I can rebase once the ORC issue is fixed

[pitrou]

Unrelated to this PR, but I'm curious, shouldn't ServerDataStream::WritesDone be called here instead of at the end of ServerTransport::DoExchange? @lidavidm

[lidavidm]

Well ideally, we would just expose gRPC and then the server could decide when it wants to end its output (it can continue reading even after doing so)

[pitrou]

Thanks a lot @raulcd !

[conbench-apache-arrow]

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

There weren't enough matching historic benchmark results to make a call on whether there were regressions.

The full Conbench report has more details.