ARROW-3289: [C++] Implement Flight DoPut #3524

ghost · 2019-01-29T20:20:51Z

Implements server/client side DoPut in C++ and extends the integration tests to exercise this.

We may want a different API for client-side DoPut that exposes any potential server response; I made it give the client a RecordBatchWriter to be symmetric with DoGet for now though.

wesm · 2019-02-01T05:13:06Z

I am headed out of town tomorrow so I may not be able to get to this until Monday.

ghost · 2019-02-01T13:42:11Z

No worries! I have Python client bindings in the pipeline, might take a look at server as well in the meantime.

wesm

This looks great. I left a number of comments, small things mostly

@jacques-n can you take a glance at the Java? It looked reasonable at first glance but I don't know the async stuff well enough to review

I am wondering out loud how we want to handle asynchronous RPC in C++. I can see an argument for maintaining a set of synchronous APIs, but we might want to implement versions like

AsyncResult<shared_ptr<RecordBatch>> ReadNext();

Then AsyncResult has both a Status (to return failures) and the returned Status (for errors). cc @pitrou in case any thoughts about this. We really haven't done much async in this project yet so it may require some time to think through what is a good approach.

wesm · 2019-02-01T14:30:38Z

cpp/src/arrow/flight/client.cc

@@ -227,7 +77,9 @@ class FlightStreamReader : public RecordBatchReader {
    // For customizing read path for better memory/serialization efficiency
    auto custom_reader = reinterpret_cast<grpc::ClientReader<FlightData>*>(stream_.get());

-    if (custom_reader->Read(&data)) {
+    // Explicitly specify the override to invoke - otherwise compiler
+    // may invoke through vtable (not updated by reinterpret_cast)


That's interesting. Which compiler did you observe this on? I didn't think this was a virtual method

$ clang++ --version Apple LLVM version 9.0.0 (clang-900.0.39.2) Target: x86_64-apple-darwin16.7.0 Thread model: posix InstalledDir: /Library/Developer/CommandLineTools/usr/bin

It's inconsistent - the calls you originally had were done statically, but when I added another instance for DoPut, Clang opted for a virtual call, judging from the disassembly.

Is it really important to avoid a virtual call here? We're probably reading a non-trivial amount of data. If gRPC defined that method as virtual, we should probably not devirtualize it like.

Unfortunately this is a correctness issue; the reader is a generic class instantiated for the Protobuf-generated struct, and uses the Protobuf serializer/deserializer. We'd like to force it to instead user our serializer. The original implementation relied on using reinterpret_cast to change the template parameter, and then hoping the compiler generated the correct static call. I found that the compiler did not always do this, and would fall back to the Protobuf serialization path. This change is intended to force it not to do this.

This seems like undefined behavior to me, and it might be worthwhile to find a better way, but comments by Wes in the code indicate that gRPC doesn't expose a good way to do this.

Hmm... What's the difference between pb::FlightData and FlightData?

Or if that happens, it means the reinterpret_cast is just illegal :-(

Yes, that's why I think it's undefined behavior - it's not a legal reinterpret_cast.

pb::FlightData is the Protobuf-generated struct, FlightData is our own struct (in serialization-internal.h)

It would be nice to add a comment warning about the undefined behavior - so that this doesn't get lost then rediscovered.

I've added a comment before the four uses of reinterpret_cast.

cpp/src/arrow/flight/client.cc

wesm · 2019-02-01T14:36:39Z

cpp/src/arrow/flight/client.cc

+    writer_ = std::move(writer);
+  }
+
+  // TODO: there isn't a way to access this as a user.


As I was reading this I was thinking about whether returning arrow::ipc::RecordBatchWriter is the best thing. If we need other kinds of methods related to acks or RPC-specific details, it might be worth defining an Flight-specific class that is returned from these methods

I was debating this as well. It should at least be easy to wrap a custom interface in RecordBatchWriter, so I guess it's better to leave this open for future extensions.

I turned this into a pImpl, the wrapper doesn't expose this still, but leaves it open for the future. The wrapper is still a RecordBatchWriter; should it be further decoupled?

For now I think we can keep it as a RecordBatchWriter, especially as the base class is just an interface.
Furthermore, we probably shouldn't consider the Flight API stable until it's battle-tested.

I've changed it back to RecordBatchWriter.

wesm · 2019-02-01T14:45:39Z

cpp/src/arrow/flight/server.cc

+      // message sent, and message failed to deserialize. IMO, we
+      // should add logging around the Status returns in
+      // serialization-internal.h to make debugging such cases easier.
+      return grpc::Status::OK;


Yes, though in this particular case the SerializationTraits customization isn't being invoked unless I misread. Does gRPC provide any way to distinguish deserialization failures in its out-of-the-box protobuf API?

Ah, I think I changed it at some point, but I did run into the problem described before. I'll take a closer look at their APIs.

This is where they use SerializationTraits:

https://github.com/grpc/grpc/blob/15bb25ae54c98c1a653281e18883487fd95d2bf0/include/grpcpp/impl/codegen/call_op_set.h#L458-L460

Not getting a message is the same as getting an invalid message, so there's no way for us to tell from the API. I'll add logging statements.

Hmm, can you explain why you return OK here? If the client doesn't send a first message, then probably it should be considered an IPC failure?
(for example, the client might have called WritesDone but still be available for receiving a response... am I misunderstanding this?)

Hmm, that's fair. I'll change it to gRPC INVALID_ARGUMENT.

cpp/src/arrow/flight/test-integration-client.cc

cpp/src/arrow/flight/test-integration-server.cc

wesm · 2019-02-04T19:19:19Z

Any more work you want to do here or should I review / merge this?

wesm · 2019-02-04T19:19:57Z

cpp/cmake_modules/ThirdpartyToolchain.cmake

+      '-DgRPC_ZLIB_PROVIDER=package'
+      '-DCMAKE_CXX_FLAGS=${EP_CXX_FLAGS}'
+      '-DCMAKE_C_FLAGS=${EP_C_FLAGS}'
+      '-DCMAKE_INSTALL_PREFIX=${GRPC_PREFIX}'


Hm I wonder if this works on Windows.

Ah that might not. On the other hand, on MacOS the double quotes got included in gRPC's path (so it tried to find /path/to/grpc").

These single quotes break the EP build for me on Ubuntu 14.04. I'll try removing the quotes

Removing them seems OK. I'll test on Windows soon

ghost · 2019-02-04T19:20:00Z

I think this is ready, if you're happy with FlightPutWriter to implement DoPut.

pitrou · 2019-02-04T19:25:11Z

@wesm I'll review this as well.

wesm · 2019-02-04T19:27:27Z

OK, I will defer to you since you're getting up to speed on Flight and I already reviewed this

pitrou

I have some comments and questions which might mostly be due to me not fully understanding this yet :-)

pitrou · 2019-02-04T17:29:31Z

cpp/src/arrow/flight/client.cc

@@ -227,7 +77,9 @@ class FlightStreamReader : public RecordBatchReader {
    // For customizing read path for better memory/serialization efficiency
    auto custom_reader = reinterpret_cast<grpc::ClientReader<FlightData>*>(stream_.get());

-    if (custom_reader->Read(&data)) {
+    // Explicitly specify the override to invoke - otherwise compiler
+    // may invoke through vtable (not updated by reinterpret_cast)


Is it really important to avoid a virtual call here? We're probably reading a non-trivial amount of data. If gRPC defined that method as virtual, we should probably not devirtualize it like.

java/flight/src/main/java/org/apache/arrow/flight/FlightClient.java

pitrou · 2019-02-04T21:08:07Z

java/flight/src/main/java/org/apache/arrow/flight/FlightClient.java

@@ -211,7 +211,8 @@ public PutObserver(VectorUnloader unloader, ClientCallStreamObserver<ArrowMessag
    @Override
    public void putNext() {
      ArrowRecordBatch batch = unloader.getRecordBatch();
-      while (!observer.isReady()) {
+      // Check the futureResult in case server sent an exception
+      while (!observer.isReady() && !futureResult.isDone()) {


I'm curious: is it customary to do such busy wait in Java?

There is a callback available in this case, though it would complicate the code quite a bit. Let me take a stab at it...

cpp/src/arrow/ipc/json.h

pitrou · 2019-02-04T21:19:01Z

cpp/src/arrow/flight/client.cc

+    writer_ = std::move(writer);
+  }
+
+  // TODO: there isn't a way to access this as a user.


For now I think we can keep it as a RecordBatchWriter, especially as the base class is just an interface.
Furthermore, we probably shouldn't consider the Flight API stable until it's battle-tested.

cpp/src/arrow/flight/client.cc

cpp/src/arrow/flight/serialization-internal.h

pitrou · 2019-02-04T21:42:55Z

cpp/src/arrow/flight/server.cc

+      // message sent, and message failed to deserialize. IMO, we
+      // should add logging around the Status returns in
+      // serialization-internal.h to make debugging such cases easier.
+      return grpc::Status::OK;


Hmm, can you explain why you return OK here? If the client doesn't send a first message, then probably it should be considered an IPC failure?
(for example, the client might have called WritesDone but still be available for receiving a response... am I misunderstanding this?)

pitrou · 2019-02-04T21:51:18Z

@wesm

I am wondering out loud how we want to handle asynchronous RPC in C++. I can see an argument for maintaining a set of synchronous APIs, but we might want to implement versions like

AsyncResult<shared_ptr<RecordBatch>> ReadNext();

Then AsyncResult has both a Status (to return failures) and the returned Status (for errors).

I'm not sure why the two status codes. Also, instead of having our own AsyncResult, we could perhaps just use std::future.

wesm · 2019-02-04T23:44:45Z

we could perhaps just use std::future

It's outside of the scope of this discussion but you need a mechanism to encapsulate both multiple return values as well as the error (if any). So you could do something like

std::future<Result<T>> where Result<T, ...> is an object that can capture multiple output types and a Status. You can also use std::pair<Status, T> if you want of course. We'll want to do async before too long to make sure we are keeping the network as busy as possible, so we can resolve it then

ghost · 2019-02-05T15:43:25Z

I believe I've addressed everything now, except for the Java busy-wait and the CMake windows build. Let's see how AppVeyor goes.

pitrou · 2019-02-05T15:50:18Z

Thanks!

pitrou · 2019-02-05T15:55:52Z

I've started an AppVeyor build on my account here: https://ci.appveyor.com/project/pitrou/arrow/builds/22141649

codecov-io · 2019-02-05T16:11:29Z

Codecov Report

Merging #3524 into master will increase coverage by 0.87%.
The diff coverage is 0%.

@@            Coverage Diff             @@
##           master    #3524      +/-   ##
==========================================
+ Coverage    87.8%   88.68%   +0.87%     
==========================================
  Files         666      537     -129     
  Lines       82365    73404    -8961     
  Branches     1069        0    -1069     
==========================================
- Hits        72323    65098    -7225     
+ Misses       9927     8306    -1621     
+ Partials      115        0     -115

Impacted Files	Coverage Δ
cpp/src/arrow/ipc/json.cc	`89.85% <0%> (-8.56%)`	⬇️
go/arrow/math/uint64_amd64.go
go/arrow/memory/memory_avx2_amd64.go
js/src/enum.ts
go/arrow/array/builder.go
js/src/Arrow.node.ts
js/src/schema.ts
go/arrow/type_traits_boolean.go
js/src/ipc/node/writer.ts
js/src/visitor/vectorloader.ts
... and 122 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update e533a9e...13fb29a. Read the comment docs.

ghost · 2019-02-05T16:46:34Z

For some reason, I can't see the AppVeyor build anymore, though I was able to a little while ago.

For the Java blocking, do you mind if I investigate that in a follow up? I think it's a non-trivial change, especially as we might want to consider making putNext itself non-blocking (but still exposing some sort of flow control so that a server does not experience unbounded memory usage with a slow client).

pitrou · 2019-02-05T16:47:31Z

I had to restart it because of a startup failure, so it got a new number:
https://ci.appveyor.com/project/pitrou/arrow/builds/22142596

pitrou · 2019-02-05T16:47:44Z

For the Java blocking, do you mind if I investigate that in a follow up?

No, that's entirely fine.

ghost · 2019-02-05T16:48:41Z

Awesome, thank you. I'll file a JIRA for it so I don't lose track.

pitrou · 2019-02-05T16:48:55Z

but still exposing some sort of flow control so that a server does not experience unbounded memory usage with a slow client

I would hope grpc helps to deal with that? The IO stack is where the send buffer size is known.

ghost · 2019-02-05T16:50:18Z

Heh...unfortunately gRPC-java will buffer outgoing messages unboundedly. You have to check isReady/wrangle callbacks yourself if you want to limit yourself.

ghost · 2019-02-05T16:50:52Z

If you're curious, I did dig it up in the source once, I could find it again...

pitrou · 2019-02-05T16:52:54Z

A possible improvement would be to have a semi-busy wait, with progressively growing sleep times, to balance responsivity and CPU consumption.

ghost · 2019-02-05T16:56:11Z

I filed ARROW-4484. Backing off on the wait is probably a good compromise.

pitrou · 2019-02-05T17:48:07Z

Build passed. I'm gonna merge. Thank you @lihalite !

ghost · 2019-02-05T17:49:14Z

Awesome, thank you and @wesm for the review! I'll get the Python stuff up soon.

Implements server/client side DoPut in C++ and extends the integration tests to exercise this. We may want a different API for client-side DoPut that exposes any potential server response; I made it give the client a RecordBatchWriter to be symmetric with DoGet for now though. Author: David Li <[email protected]> Closes #3524 from lihalite/arrow-3289 and squashes the following commits: 13fb29a <David Li> Document why VectorUnloader must align batches in Flight f32c0b2 <David Li> Indicate error to client in DoPut if no message sent cd56782 <David Li> Warn about undefined behavior in Flight source 1f816e8 <David Li> Move serialization helpers out of gRPC namespace 21b315a <David Li> Hide FlightPutWriter from public interface for now 6edf2e2 <David Li> Introduce FlightPutWriter 58d6936 <David Li> Enable building with non-CMake c-ares cfa4ca5 <David Li> Properly quote arguments to gRPC CMake build 302dd33 <David Li> Explicitly link Protobuf for Flight 419ad68 <David Li> Log (de)serialization failures in Flight fast-path 562b861 <David Li> Factor out FlightData->Message conversion 65d6ba2 <David Li> Clean up C++ Flight integration client 3e185cb <David Li> Add convenience to parse JSON from file 111b3e6 <David Li> Fix style/lint issues 3cb51ba <David Li> Test all returned locations in Flight integration tests 905ef38 <David Li> Implement C++ Flight DoPut 138141f <David Li> Fix FromProto for FlightDescriptor a11a5ac <David Li> Don't hang in Flight DoPut if server sends exception b3ac01a <David Li> Align RecordBatch on client side in Flight DoPut 846df73 <David Li> Implement put in Java Flight integration server

wesm reviewed Feb 1, 2019

View reviewed changes

wesm reviewed Feb 4, 2019

View reviewed changes

pitrou reviewed Feb 4, 2019

View reviewed changes

David Li added 19 commits February 5, 2019 10:26

Implement put in Java Flight integration server

846df73

Align RecordBatch on client side in Flight DoPut

b3ac01a

Don't hang in Flight DoPut if server sends exception

a11a5ac

Fix FromProto for FlightDescriptor

138141f

Implement C++ Flight DoPut

905ef38

Test all returned locations in Flight integration tests

3cb51ba

Fix style/lint issues

111b3e6

[ARROW-4409] Add convenience to parse JSON from file

3e185cb

Clean up C++ Flight integration client

65d6ba2

Factor out FlightData->Message conversion

562b861

Log (de)serialization failures in Flight fast-path

419ad68

Explicitly link Protobuf for Flight

302dd33

Properly quote arguments to gRPC CMake build

cfa4ca5

Enable building with non-CMake c-ares

58d6936

Introduce FlightPutWriter

6edf2e2

Hide FlightPutWriter from public interface for now

21b315a

Move serialization helpers out of gRPC namespace

1f816e8

Warn about undefined behavior in Flight source

cd56782

Indicate error to client in DoPut if no message sent

f32c0b2

Document why VectorUnloader must align batches in Flight

13fb29a

pitrou closed this in 4004b72 Feb 5, 2019

asfimport mentioned this pull request Feb 6, 2019

[C++] Implement DoPut command for Flight on client and server side #19627

Closed

ARROW-3289: [C++] Implement Flight DoPut #3524

ARROW-3289: [C++] Implement Flight DoPut #3524

Conversation

ghost commented Jan 29, 2019

wesm commented Feb 1, 2019

ghost commented Feb 1, 2019

wesm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wesm commented Feb 4, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ghost commented Feb 4, 2019

pitrou commented Feb 4, 2019

wesm commented Feb 4, 2019

pitrou left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pitrou commented Feb 4, 2019

wesm commented Feb 4, 2019

ghost commented Feb 5, 2019

pitrou commented Feb 5, 2019

pitrou commented Feb 5, 2019

codecov-io commented Feb 5, 2019

Codecov Report

ghost commented Feb 5, 2019

pitrou commented Feb 5, 2019

pitrou commented Feb 5, 2019

ghost commented Feb 5, 2019

pitrou commented Feb 5, 2019

ghost commented Feb 5, 2019

ghost commented Feb 5, 2019

pitrou commented Feb 5, 2019

ghost commented Feb 5, 2019

pitrou commented Feb 5, 2019

ghost commented Feb 5, 2019