You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "raulcd (via GitHub)" <gi...@apache.org> on 2025/08/22 08:35:32 UTC
[PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayoadWriter for TransportMessageWriter (DoExchange) [arrow]
raulcd opened a new pull request, #47410:
URL: https://github.com/apache/arrow/pull/47410
### 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 can use an `ipc::RecordBatchWriter` with a custom `IpcPayloadWriter` to do not reimplement Dictionary Replacement / Deltas logic.
### What changes are included in this PR?
Remove manually generating ipc Payloads by calling the individual payload calls to use 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:
```python
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298495172
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -188,71 +226,54 @@ class TransportMessageWriter final : public FlightMessageWriter {
Status WriteMetadata(std::shared_ptr<Buffer> app_metadata) override {
FlightPayload payload{};
payload.app_metadata = app_metadata;
- return WritePayload(payload);
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
Review Comment:
In `WritePayload` we used to have the following error path, why are we suddenly silencing the write error here?
```c++
if (!success) {
return MakeFlightError(FlightStatusCode::Internal,
"Could not write metadata to stream (client disconnect?)");
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298507312
##########
python/pyarrow/tests/test_flight.py:
##########
@@ -2536,3 +2536,37 @@ def received_headers(self, headers):
assert ("x-header-bin", b"header\x01value") in factory.headers
assert ("x-trailer", "trailer-value") in factory.headers
assert ("x-trailer-bin", b"trailer\x01value") in factory.headers
+
+
+def test_flight_dictionary_deltas_do_exchange():
+ class DeltaFlightServer(ConstantFlightServer):
+ def do_exchange(self, context, descriptor, reader, writer):
+ if descriptor.command == b'dict_deltas':
+ expected_table = simple_dicts_table()
+ received_table = reader.read_all()
+ assert received_table.equals(expected_table)
Review Comment:
Perhaps add a TODO for inspecting the reader stats?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayoadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #47410:
URL: https://github.com/apache/arrow/pull/47410#issuecomment-3213556873
:warning: GitHub issue #26727 **has been automatically assigned in GitHub** to PR creator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "conbench-apache-arrow[bot] (via GitHub)" <gi...@apache.org>.
conbench-apache-arrow[bot] commented on PR #47410:
URL: https://github.com/apache/arrow/pull/47410#issuecomment-3226036953
After merging your PR, Conbench analyzed the 4 benchmarking runs that have been run so far on merge-commit de520483af199caf9dc101a4fac00d1f6dd72dc5.
There weren't enough matching historic benchmark results to make a call on whether there were regressions.
The [full Conbench report](https://github.com/apache/arrow/runs/48958638361) has more details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2297644611
##########
python/pyarrow/tests/test_flight.py:
##########
@@ -2536,3 +2536,28 @@ def received_headers(self, headers):
assert ("x-header-bin", b"header\x01value") in factory.headers
assert ("x-trailer", "trailer-value") in factory.headers
assert ("x-trailer-bin", b"trailer\x01value") in factory.headers
+
+
+def test_flight_dictionary_deltas_do_exchange():
+ class DeltaFlightServer(ConstantFlightServer):
+ def do_exchange(self, context, descriptor, reader, writer):
+ if descriptor.command == b'dict_deltas':
+ table = simple_dicts_table()
+ options = pa.ipc.IpcWriteOptions(emit_dictionary_deltas=True)
+ writer.begin(table.schema, options=options)
+ writer.write_table(table)
+
+ with DeltaFlightServer() as server, \
+ FlightClient(('localhost', server.port)) as client:
+
+ descriptor = flight.FlightDescriptor.for_command(b"dict_deltas")
+ writer, reader = client.do_exchange(descriptor,
Review Comment:
This doesn't check that the client-side `writer` can also emit dictionary deltas, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayoadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "raulcd (via GitHub)" <gi...@apache.org>.
raulcd commented on PR #47410:
URL: https://github.com/apache/arrow/pull/47410#issuecomment-3213705933
Current CI failures are unrelated and tracked here:
- https://github.com/apache/arrow/issues/47399
- https://github.com/apache/arrow/issues/46690
I can rebase once the ORC issue is fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2296845703
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -188,71 +227,55 @@ class TransportMessageWriter final : public FlightMessageWriter {
Status WriteMetadata(std::shared_ptr<Buffer> app_metadata) override {
FlightPayload payload{};
payload.app_metadata = app_metadata;
- return WritePayload(payload);
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
+ return Close();
+ }
+ // Those messages are not written through the batch writer,
+ // count them separately for stats.
+ extra_metadata_messages_++;
Review Comment:
That works for me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayoadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "raulcd (via GitHub)" <gi...@apache.org>.
raulcd commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2296731940
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -188,71 +227,55 @@ class TransportMessageWriter final : public FlightMessageWriter {
Status WriteMetadata(std::shared_ptr<Buffer> app_metadata) override {
FlightPayload payload{};
payload.app_metadata = app_metadata;
- return WritePayload(payload);
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
+ return Close();
+ }
+ // Those messages are not written through the batch writer,
+ // count them separately for stats.
+ extra_metadata_messages_++;
Review Comment:
I am adding them to the total number of messages written : `write_stats.num_messages`
They were previously added due to `WritePayload` counting them.
I think they should be added, maybe the naming of the variable is confusing and should just be `extra_messages_`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "raulcd (via GitHub)" <gi...@apache.org>.
raulcd commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298528727
##########
python/pyarrow/tests/test_flight.py:
##########
@@ -2536,3 +2536,37 @@ def received_headers(self, headers):
assert ("x-header-bin", b"header\x01value") in factory.headers
assert ("x-trailer", "trailer-value") in factory.headers
assert ("x-trailer-bin", b"trailer\x01value") in factory.headers
+
+
+def test_flight_dictionary_deltas_do_exchange():
+ class DeltaFlightServer(ConstantFlightServer):
+ def do_exchange(self, context, descriptor, reader, writer):
+ if descriptor.command == b'dict_deltas':
+ expected_table = simple_dicts_table()
+ received_table = reader.read_all()
+ assert received_table.equals(expected_table)
Review Comment:
good idea I have it in mind but better to add a comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2297807174
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -160,25 +161,63 @@ class TransportMessageReader final : public FlightMessageReader {
std::shared_ptr<Buffer> app_metadata_;
};
-// TODO(ARROW-10787): this should use the same writer/ipc trick as client
+/// \brief An IpcPayloadWriter for ServerDataStream.
+///
+/// To support app_metadata and reuse the existing IPC infrastructure,
+/// this takes a pointer to a buffer to be combined with the IPC
+/// payload when writing a Flight payload.
+class TransportMessagePayloadWriter : public ipc::internal::IpcPayloadWriter {
+ public:
+ TransportMessagePayloadWriter(ServerDataStream* stream,
+ std::shared_ptr<Buffer>* app_metadata)
+ : stream_(stream), app_metadata_(app_metadata) {}
+
+ Status Start() override { return Status::OK(); }
+ Status WritePayload(const ipc::IpcPayload& ipc_payload) override {
+ FlightPayload payload;
+ payload.ipc_message = ipc_payload;
+
+ if (ipc_payload.type == ipc::MessageType::RECORD_BATCH && *app_metadata_) {
+ payload.app_metadata = std::move(*app_metadata_);
+ }
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
+ return arrow::Status(arrow::StatusCode::IOError,
+ "Could not write record batch to stream");
+ }
+ return arrow::Status::OK();
+ }
+ Status Close() override {
+ // Closing is handled one layer up in TransportMessageWriter::Close
Review Comment:
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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298507778
##########
python/pyarrow/tests/test_flight.py:
##########
@@ -2536,3 +2536,37 @@ def received_headers(self, headers):
assert ("x-header-bin", b"header\x01value") in factory.headers
assert ("x-trailer", "trailer-value") in factory.headers
assert ("x-trailer-bin", b"trailer\x01value") in factory.headers
+
+
+def test_flight_dictionary_deltas_do_exchange():
+ class DeltaFlightServer(ConstantFlightServer):
+ def do_exchange(self, context, descriptor, reader, writer):
+ if descriptor.command == b'dict_deltas':
+ expected_table = simple_dicts_table()
+ received_table = reader.read_all()
+ assert received_table.equals(expected_table)
+
+ options = pa.ipc.IpcWriteOptions(emit_dictionary_deltas=True)
+ writer.begin(expected_table.schema, options=options)
+ writer.write_table(expected_table)
+
+ with DeltaFlightServer() as server, \
+ FlightClient(('localhost', server.port)) as client:
+ expected_table = simple_dicts_table()
+
+ descriptor = flight.FlightDescriptor.for_command(b"dict_deltas")
+ writer, reader = client.do_exchange(descriptor,
+ options=flight.FlightCallOptions(
+ write_options=pa.ipc.IpcWriteOptions(
+ emit_dictionary_deltas=True)
+ )
+ )
+ # Send client table with dictionary updates (deltas should be sent)
+ with writer:
+ writer.begin(expected_table.schema, options=pa.ipc.IpcWriteOptions(
+ emit_dictionary_deltas=True))
+ writer.write_table(expected_table)
+ writer.done_writing()
+ received_table = reader.read_all()
Review Comment:
Same here: add a TODO?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "raulcd (via GitHub)" <gi...@apache.org>.
raulcd commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298518507
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -188,71 +226,54 @@ class TransportMessageWriter final : public FlightMessageWriter {
Status WriteMetadata(std::shared_ptr<Buffer> app_metadata) override {
FlightPayload payload{};
payload.app_metadata = app_metadata;
- return WritePayload(payload);
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
Review Comment:
I am using the same logic that we are using on the client:
https://github.com/apache/arrow/blob/f8b20f131a072ef423e81b8a676f42a82255f4ec/cpp/src/arrow/flight/client.cc#L456-L464
We should return the built error after attempting to close but we should be consistent on the client side too. Should I add the error to the client too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2299795862
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -188,71 +226,54 @@ class TransportMessageWriter final : public FlightMessageWriter {
Status WriteMetadata(std::shared_ptr<Buffer> app_metadata) override {
FlightPayload payload{};
payload.app_metadata = app_metadata;
- return WritePayload(payload);
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
Review Comment:
The update works for me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298505046
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -160,25 +161,62 @@ class TransportMessageReader final : public FlightMessageReader {
std::shared_ptr<Buffer> app_metadata_;
};
-// TODO(ARROW-10787): this should use the same writer/ipc trick as client
+/// \brief An IpcPayloadWriter for ServerDataStream.
+///
+/// To support app_metadata and reuse the existing IPC infrastructure,
+/// this takes a pointer to a buffer to be combined with the IPC
+/// payload when writing a Flight payload.
+class TransportMessagePayloadWriter : public ipc::internal::IpcPayloadWriter {
+ public:
+ TransportMessagePayloadWriter(ServerDataStream* stream,
+ std::shared_ptr<Buffer>* app_metadata)
+ : stream_(stream), app_metadata_(app_metadata) {}
+
+ Status Start() override { return Status::OK(); }
+ Status WritePayload(const ipc::IpcPayload& ipc_payload) override {
+ FlightPayload payload;
+ payload.ipc_message = ipc_payload;
+
+ if (ipc_payload.type == ipc::MessageType::RECORD_BATCH && *app_metadata_) {
+ payload.app_metadata = std::move(*app_metadata_);
+ }
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
+ return arrow::Status(arrow::StatusCode::IOError,
+ "Could not write record batch to stream");
Review Comment:
Why don't we use the same error construction as we used to in `WritePayload`? Something like:
```c++
return MakeFlightError(FlightStatusCode::Internal,
"Could not write record batch to stream (client disconnect?)");
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "raulcd (via GitHub)" <gi...@apache.org>.
raulcd commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298187601
##########
python/pyarrow/tests/test_flight.py:
##########
@@ -2536,3 +2536,28 @@ def received_headers(self, headers):
assert ("x-header-bin", b"header\x01value") in factory.headers
assert ("x-trailer", "trailer-value") in factory.headers
assert ("x-trailer-bin", b"trailer\x01value") in factory.headers
+
+
+def test_flight_dictionary_deltas_do_exchange():
+ class DeltaFlightServer(ConstantFlightServer):
+ def do_exchange(self, context, descriptor, reader, writer):
+ if descriptor.command == b'dict_deltas':
+ table = simple_dicts_table()
+ options = pa.ipc.IpcWriteOptions(emit_dictionary_deltas=True)
+ writer.begin(table.schema, options=options)
+ writer.write_table(table)
+
+ with DeltaFlightServer() as server, \
+ FlightClient(('localhost', server.port)) as client:
+
+ descriptor = flight.FlightDescriptor.for_command(b"dict_deltas")
+ writer, reader = client.do_exchange(descriptor,
Review Comment:
I've added an update to the test. I was planning to add it on the `ReadStats` PR but it's worth validating here. Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayoadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "raulcd (via GitHub)" <gi...@apache.org>.
raulcd commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2293092861
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -188,71 +227,55 @@ class TransportMessageWriter final : public FlightMessageWriter {
Status WriteMetadata(std::shared_ptr<Buffer> app_metadata) override {
FlightPayload payload{};
payload.app_metadata = app_metadata;
- return WritePayload(payload);
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
+ return Close();
+ }
+ // Those messages are not written through the batch writer,
+ // count them separately for stats.
+ extra_metadata_messages_++;
Review Comment:
The client is not counting those messages at the moment, should we open a bug for it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298527136
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -188,71 +226,54 @@ class TransportMessageWriter final : public FlightMessageWriter {
Status WriteMetadata(std::shared_ptr<Buffer> app_metadata) override {
FlightPayload payload{};
payload.app_metadata = app_metadata;
- return WritePayload(payload);
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
Review Comment:
Hmm, my vote would be to remain consistent with previous server code. Inconsistency between server and client isn't necessarily a problem. What do you think @lidavidm ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "raulcd (via GitHub)" <gi...@apache.org>.
raulcd commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298602159
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -188,71 +226,54 @@ class TransportMessageWriter final : public FlightMessageWriter {
Status WriteMetadata(std::shared_ptr<Buffer> app_metadata) override {
FlightPayload payload{};
payload.app_metadata = app_metadata;
- return WritePayload(payload);
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
Review Comment:
I've pushed:
https://github.com/apache/arrow/pull/47410/commits/189968b7b97236d8782bfd4fc8a668bb5b7d9bd2
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayoadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2296515008
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -188,71 +227,55 @@ class TransportMessageWriter final : public FlightMessageWriter {
Status WriteMetadata(std::shared_ptr<Buffer> app_metadata) override {
FlightPayload payload{};
payload.app_metadata = app_metadata;
- return WritePayload(payload);
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
+ return Close();
+ }
+ // Those messages are not written through the batch writer,
+ // count them separately for stats.
+ extra_metadata_messages_++;
Review Comment:
I'm not sure they should count as "metadata" in the first place (aren't the stats for IPC metadata?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2297674439
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -160,25 +161,63 @@ class TransportMessageReader final : public FlightMessageReader {
std::shared_ptr<Buffer> app_metadata_;
};
-// TODO(ARROW-10787): this should use the same writer/ipc trick as client
+/// \brief An IpcPayloadWriter for ServerDataStream.
+///
+/// To support app_metadata and reuse the existing IPC infrastructure,
+/// this takes a pointer to a buffer to be combined with the IPC
+/// payload when writing a Flight payload.
+class TransportMessagePayloadWriter : public ipc::internal::IpcPayloadWriter {
+ public:
+ TransportMessagePayloadWriter(ServerDataStream* stream,
+ std::shared_ptr<Buffer>* app_metadata)
+ : stream_(stream), app_metadata_(app_metadata) {}
+
+ Status Start() override { return Status::OK(); }
+ Status WritePayload(const ipc::IpcPayload& ipc_payload) override {
+ FlightPayload payload;
+ payload.ipc_message = ipc_payload;
+
+ if (ipc_payload.type == ipc::MessageType::RECORD_BATCH && *app_metadata_) {
+ payload.app_metadata = std::move(*app_metadata_);
+ }
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
+ return arrow::Status(arrow::StatusCode::IOError,
+ "Could not write record batch to stream");
+ }
+ return arrow::Status::OK();
+ }
+ Status Close() override {
+ // Closing is handled one layer up in TransportMessageWriter::Close
Review Comment:
Unrelated to this PR, but I'm curious, shouldn't `ServerDataStream::WritesDone` be called here instead of at the end of `ServerTransport::DoExchange`? @lidavidm
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2297659276
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -160,25 +161,63 @@ class TransportMessageReader final : public FlightMessageReader {
std::shared_ptr<Buffer> app_metadata_;
};
-// TODO(ARROW-10787): this should use the same writer/ipc trick as client
+/// \brief An IpcPayloadWriter for ServerDataStream.
+///
+/// To support app_metadata and reuse the existing IPC infrastructure,
+/// this takes a pointer to a buffer to be combined with the IPC
+/// payload when writing a Flight payload.
+class TransportMessagePayloadWriter : public ipc::internal::IpcPayloadWriter {
+ public:
+ TransportMessagePayloadWriter(ServerDataStream* stream,
+ std::shared_ptr<Buffer>* app_metadata)
+ : stream_(stream), app_metadata_(app_metadata) {}
+
+ Status Start() override { return Status::OK(); }
+ Status WritePayload(const ipc::IpcPayload& ipc_payload) override {
+ FlightPayload payload;
+ payload.ipc_message = ipc_payload;
+
+ if (ipc_payload.type == ipc::MessageType::RECORD_BATCH && *app_metadata_) {
+ payload.app_metadata = std::move(*app_metadata_);
+ }
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
+ return arrow::Status(arrow::StatusCode::IOError,
+ "Could not write record batch to stream");
+ }
+ return arrow::Status::OK();
+ }
+ Status Close() override {
+ // Closing is handled one layer up in TransportMessageWriter::Close
+ return Status::OK();
+ }
+
+ private:
+ ServerDataStream* stream_;
+ std::shared_ptr<Buffer>* app_metadata_;
+};
+
class TransportMessageWriter final : public FlightMessageWriter {
public:
explicit TransportMessageWriter(ServerDataStream* stream)
- : stream_(stream), ipc_options_(::arrow::ipc::IpcWriteOptions::Defaults()) {}
+ : stream_(stream),
+ app_metadata_(nullptr),
+ ipc_options_(::arrow::ipc::IpcWriteOptions::Defaults()) {}
Status Begin(const std::shared_ptr<Schema>& schema,
const ipc::IpcWriteOptions& options) override {
- if (started_) {
+ ipc_options_ = options;
+ if (batch_writer_) {
Review Comment:
Why not check `started_`? This should probably be consistent with what `CheckStarted` does.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "raulcd (via GitHub)" <gi...@apache.org>.
raulcd commented on code in PR #47410:
URL: https://github.com/apache/arrow/pull/47410#discussion_r2298191512
##########
cpp/src/arrow/flight/transport_server.cc:
##########
@@ -160,25 +161,63 @@ class TransportMessageReader final : public FlightMessageReader {
std::shared_ptr<Buffer> app_metadata_;
};
-// TODO(ARROW-10787): this should use the same writer/ipc trick as client
+/// \brief An IpcPayloadWriter for ServerDataStream.
+///
+/// To support app_metadata and reuse the existing IPC infrastructure,
+/// this takes a pointer to a buffer to be combined with the IPC
+/// payload when writing a Flight payload.
+class TransportMessagePayloadWriter : public ipc::internal::IpcPayloadWriter {
+ public:
+ TransportMessagePayloadWriter(ServerDataStream* stream,
+ std::shared_ptr<Buffer>* app_metadata)
+ : stream_(stream), app_metadata_(app_metadata) {}
+
+ Status Start() override { return Status::OK(); }
+ Status WritePayload(const ipc::IpcPayload& ipc_payload) override {
+ FlightPayload payload;
+ payload.ipc_message = ipc_payload;
+
+ if (ipc_payload.type == ipc::MessageType::RECORD_BATCH && *app_metadata_) {
+ payload.app_metadata = std::move(*app_metadata_);
+ }
+ ARROW_ASSIGN_OR_RAISE(auto success, stream_->WriteData(payload));
+ if (!success) {
+ return arrow::Status(arrow::StatusCode::IOError,
+ "Could not write record batch to stream");
+ }
+ return arrow::Status::OK();
+ }
+ Status Close() override {
+ // Closing is handled one layer up in TransportMessageWriter::Close
+ return Status::OK();
+ }
+
+ private:
+ ServerDataStream* stream_;
+ std::shared_ptr<Buffer>* app_metadata_;
+};
+
class TransportMessageWriter final : public FlightMessageWriter {
public:
explicit TransportMessageWriter(ServerDataStream* stream)
- : stream_(stream), ipc_options_(::arrow::ipc::IpcWriteOptions::Defaults()) {}
+ : stream_(stream),
+ app_metadata_(nullptr),
+ ipc_options_(::arrow::ipc::IpcWriteOptions::Defaults()) {}
Status Begin(const std::shared_ptr<Schema>& schema,
const ipc::IpcWriteOptions& options) override {
- if (started_) {
+ ipc_options_ = options;
+ if (batch_writer_) {
Review Comment:
I should have removed `started_` from `CheckStarted` and just use the `batch_writer_` as we do on the client. I've updated `CheckStarted` accordingly and remove the now unnecessary `started_` flag.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-26727: [C++][Flight] Use ipc::RecordBatchWriter with custom IpcPayloadWriter for TransportMessageWriter (DoExchange) [arrow]
Posted by "raulcd (via GitHub)" <gi...@apache.org>.
raulcd merged PR #47410:
URL: https://github.com/apache/arrow/pull/47410
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org