ARROW-4626: [Flight] Add application-defined metadata to DoGet/DoPut #4282

ghost · 2019-05-09T13:40:33Z

Also covers ARROW-4627.

This is quite an enormous change, if preferred, I can do my best to try and separate changes.

pitrou

Thanks for doing this! Here are some comments on the C++/Python side.

cpp/src/arrow/flight/client.h

cpp/src/arrow/flight/client.cc

cpp/src/arrow/flight/client.h

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

python/pyarrow/_flight.pyx

pitrou · 2019-05-15T14:20:27Z

docs/source/format/Flight.rst

+.. specific language governing permissions and limitations
+.. under the License.
+
+Arrow Flight RPC


Thanks for the doc additions!

Of course! We have teams that wanted fuller docs, so hope this is a reasonable starting point.

wesm · 2019-05-15T18:54:22Z

I'll try to leave some comments on this when I can -- I have been heads down on ARROW-3144 and haven't been doing code reviews for the last week+

ghost · 2019-05-15T18:56:51Z

@wesm no worries, this is also going to need rebasing once your PR lands.

ghost · 2019-05-17T17:05:37Z

I am going to close this until #4047 is merged just to keep the PR queue smaller, as this will need a bunch of rebasing.

ghost · 2019-05-22T18:35:30Z

Reopening now that #4047 is merged.

This also does half of https://issues.apache.org/jira/browse/ARROW-5143, enabling integration testing of (non-nested) dictionaries in Flight.

pitrou

I noticed a couple more things. Are you waiting for a Java review as well?

python/pyarrow/tests/test_flight.py

ghost · 2019-05-28T12:38:55Z

Thanks for the feedback! Rebased & made fixes. @pitrou yes, I'm waiting for a Java review too.

ghost · 2019-06-03T20:29:06Z

There looks to be some spurious CI failures, especially on Windows. I'll try one last time to get those to succeed...

pitrou · 2019-06-03T20:30:17Z

MinGW builds are unfortunately broken currently.

ghost · 2019-06-04T22:35:17Z

Flight cancellation tests look flaky, disabling those in CI.

ghost · 2019-06-07T18:45:27Z

Okay, unit tests finally pass 😄 Just waiting on Java review, I believe.

wesm · 2019-06-07T19:32:05Z

Thanks @lidavidm! I reached out to @jacques-n to see when he can spend some time reviewing the Java side of this

wesm · 2019-06-10T20:42:56Z

cpp/src/arrow/flight/client.cc

+                     std::unique_ptr<grpc::ClientReader<pb::FlightData>> stream,
+                     std::unique_ptr<GrpcStreamReader>* out);
+  std::shared_ptr<Schema> schema() const override;
+  Status ReadNext(std::shared_ptr<RecordBatch>* out) override;


Sorry to phone in from the peanut gallery, but what do you think about introducing a result struct like

struct FlightStreamChunk { std::shared_ptr<RecordBatch> data; std::shared_ptr<Buffer> app_metadata; };

Sure, that's a little more convenient for callers. We can't change the signature of ReadNext here though, only ReadWithMetadata.

In #4483 I introduced an Iterator<T> concept -- it's not clear to me how important it is to conform to the RecordBatchReader interface (or we could also use virtual inheritance so that the flight data reader can be dynamically-casted to Iterator<std::shared_ptr<RecordBatch>>) cc @pitrou

Hmm, maybe it's not so important, so long as in Python it still exposes the utility methods people expect (we find it useful to do do_get().read_pandas() as that's by far the common case). If we could expose both iterator-of-record-batch and iterator-of-chunk that would be best (as not everyone will want the metadata).

On the C++ side having a intended API endpoint would probably the easiest thing (since the metadata can just be ignored if the developer-user doesn't care), and we can make the Python API as ergonomic as desired (I presume -- it is our intent in fact -- that most consumers of the Python API are going to be sitting behind a front-end interface to pyarrow anyway)

Alright, I went ahead and changed it to use that interface; when the other PR is merged, we can then make it actually implement Iterator.

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

jacques-n · 2019-06-11T04:22:07Z

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

  private volatile VectorLoader loader;
  private volatile Throwable ex;
  private volatile FlightDescriptor descriptor;
  private volatile Schema schema;
+  private volatile byte[] applicationMetadata = null;


Let's construct as a listener pattern instead. Assuming people only want the last one seems a bit specific to what must be your use case.

Okay. I initially thought it would make correlating data with the associated metadata hard (and in the protocol, they're sent together - the metadata is not out-of-band for the cases that FlightStream represents), but the Java APIs are not really designed around manipulating individual RecordBatches anyways, so a listener would make sense.

@jacques-n on second thought, I don't think a listener is appropriate - it complicates usage quite a bit, especially if you want to correlate a particular metadatum with a particular RecordBatch. The current code does guarantee that applicationMetadata is not updated except by a call to next, which is the same invariant as for the VectorSchemaRoot contained in FlightStream. So if you do want previous metadata messages, you can retain them yourself (as you would with a listener anyways).

Additionally, API design-wise, it's hard to decide how FlightProducer#acceptPut should register a listener with FlightStream.

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

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

.../flight/src/main/java/org/apache/arrow/flight/example/integration/IntegrationTestClient.java

java/flight/src/test/java/org/apache/arrow/flight/TestBasicOperation.java

ghost · 2019-06-11T20:24:12Z

Side note, I will bump the Netty dependency version...I think this is the source of Netty failing to find methods at runtime (since we specified a Netty version that conflicted with gRPC's).

I'm also going to try re-enabling some of the Flight Java tests, after tracking down what I believe are some memory allocation issues. I think we might want to look at the APIs and make it explicit what is owned where and when, though.

ghost · 2019-06-13T17:22:45Z

Integration tests are broken, will investigate...

ghost · 2019-06-13T21:24:35Z

Ok, all that should be left is rebasing on #4553.

wesm · 2019-06-14T12:50:30Z

There's still a flake looks like. Do you want @jacques-n to have another look before we merge?

ghost · 2019-06-14T12:53:05Z

That's what I get for forgetting flake8...

I would appreciate any further thoughts Jacques has on the metadata API in Java. I'm not convinced on the callback-based API, as it makes the control flow complicated, especially when mixed with the blocking API for reading data.

codecov-io · 2019-06-17T16:05:24Z

Codecov Report

Merging #4282 into master will decrease coverage by 13.25%.
The diff coverage is n/a.

@@             Coverage Diff             @@
##           master    #4282       +/-   ##
===========================================
- Coverage   88.57%   75.31%   -13.26%     
===========================================
  Files         860       56      -804     
  Lines      108022     3192   -104830     
  Branches     1253        0     -1253     
===========================================
- Hits        95678     2404    -93274     
+ Misses      12065      788    -11277     
+ Partials      279        0      -279

Impacted Files	Coverage Δ
python/pyarrow/ipc.pxi
cpp/src/arrow/csv/chunker-test.cc
cpp/src/parquet/column_page.h
cpp/src/parquet/bloom_filter-test.cc
cpp/src/arrow/array/builder_decimal.cc
cpp/src/plasma/client.cc
cpp/src/arrow/io/test-common.h
cpp/src/arrow/util/int-util-test.cc
cpp/src/arrow/python/io.cc
python/pyarrow/hdfs.py
... and 778 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 9425831...e2a260b. Read the comment docs.

ghost · 2019-06-17T21:16:05Z

Ok, tried to fix the last Python tests...looks like gRPC error messages change text across versions (of course), so once #4484 lands, let's convert those tests to instead provide/check the gRPC error code in some way.

jacques-n

Couple small comments, fine to be addressed in a follow-up

Overall, +1

jacques-n · 2019-06-21T17:45:45Z

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

+
+    private final CompletableFuture<Void> result;
+    private final BufferAllocator allocator;
+    private final StreamListener<PutResult> listener;


Can we remove result here given the addition of the listener and just make a CompleteFuture based observer that implements listener (composed rather than all together)

My ask was slightly different. I was proposing that it doesn't make sense to have a completeable future and a listener in the same class. I'm struggling with the pattern where you're doing a double call in the on* event handling calls.

Ah, sorry - to be clear, you'd like to essentially merge PutObserver and SetStreamListener? That sounds reasonable.

It's weird to have listener > future > listener. Why not just stack the two listeners and the second one can do the future result (if it wants to go from listener > future)?

I guess the issue stems from trying to expose the result in two places; asynchronously via an optional listener given when starting DoPut, and through a future accessible on the listener used to write data during a DoPut. Maybe instead of the optional user-supplied listener defaulting to a no-op one, we can provide an easy way to supply one that converts the result to a future. That is, the current API looks like

final FlightClient.ClientStreamListener writer = client.startPut(descriptor, root, new StreamListener<>() { // .... @Override public void onCompleted() { } }); // call writer.putNext... writer.getResult(); // implicitly, we have to complete this future, while providing results to the listener above

and there is duplication in receiving the events from the server, instead we could have

final FutureListener listener = new FutureListener(); // { // optionally override onNext to get application metadata // }; final FlightClient.ClientStreamListener writer = client.startPut(descriptor, root, listener); // call writer.putNext()... listener.getResult();

Is that cleaner? There's a bit more boilerplate, but we could get rid of that if we require the listener to be a FutureListener, create a default instance if the client-supplied listener is null, and have writer.getResult() just pass through to listener.getResult().

jacques-n · 2019-06-21T17:49:29Z

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

@@ -99,6 +118,9 @@ public void close() throws Exception {
        .collect(Collectors.toList());

    AutoCloseables.close(Iterables.concat(closeables, ImmutableList.of(root.get())));
+    if (applicationMetadata != null) {
+      applicationMetadata.close();


add this reference to auto closeables above to ensure it is closed even if another closeable throws (also don't need the null check there)

jacques-n · 2019-06-21T19:25:42Z

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

+    applicationMetadata = metadata;
+  }
+
+  /** Create a PutResult with application-specific metadata. */


probably good to state reference counting behavior in this kind of method.

ghost · 2019-06-21T19:54:18Z

Rebased & addressed the last bit of feedback.

jacques-n · 2019-06-24T12:48:28Z

I think that makes sense. Providing a simple listener impl for returning the items via sync methods seems like a cleaner approach.

…

On Sat, Jun 22, 2019, 6:02 AM lihalite ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In java/flight/src/main/java/org/apache/arrow/flight/FlightClient.java <#4282 (comment)>: > return stream; } - private static class SetStreamObserver<T> implements StreamObserver<T> { - private final SettableFuture<T> result = SettableFuture.create(); - private volatile T resultLocal; + private static class SetStreamObserver implements StreamObserver<Flight.PutResult> { + + private final CompletableFuture<Void> result; + private final BufferAllocator allocator; + private final StreamListener<PutResult> listener; I guess the issue stems from trying to expose the result in two places; asynchronously via an optional listener given when starting DoPut, and through a future accessible on the listener used to write data during a DoPut. Maybe instead of the optional user-supplied listener defaulting to a no-op one, we can provide an easy way to supply one that converts the result to a future. That is, the current API looks like final FlightClient.ClientStreamListener writer = client.startPut(descriptor, root, new StreamListener<>() { // .... @OverRide public void onCompleted() { } });// call writer.putNext... writer.getResult(); // implicitly, we have to complete this future, while providing results to the listener above and there is duplication in receiving the events from the server, instead we could have final FutureListener listener = new FutureListener(); // { // optionally override onNext to get application metadata// };final FlightClient.ClientStreamListener writer = client.startPut(descriptor, root, listener);// call writer.putNext()... listener.getResult(); Is that cleaner? There's a bit more boilerplate, but we could get rid of that if we *require* the listener to be a FutureListener, create a default instance if the client-supplied listener is null, and have writer.getResult() just pass through to listener.getResult()`. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#4282?email_source=notifications&email_token=AABMYNVQTHDVFLDBJ24CJ4TP3YPHNA5CNFSM4HL2TJIKYY3PNVWWK3TUL52HS4DFWFIHK3DMKJSXC5LFON2FEZLWNFSXPKTDN5WW2ZLOORPWSZGOB4K6U7Y#discussion_r296445157>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AABMYNQ7GT4NMWNCDWETN73P3YPHNANCNFSM4HL2TJIA> .

ghost · 2019-06-24T15:31:20Z

Rebased and implemented a sync metadata interface as per @jacques-n's suggestions. Example:

arrow/java/flight/src/test/java/org/apache/arrow/flight/TestApplicationMetadata.java

Line 117 in f302c79

public void uploadMetadataSync() {

jacques-n · 2019-06-24T17:29:18Z

Thanks @lihalite! Lgtm. +1 on the java side.

wesm · 2019-06-24T23:01:15Z

Build failing to due to lint issues and failure in the integration tests. I'll be happy to merge once the tests are green

ghost · 2019-06-26T13:27:13Z

No worries, should've rebased originally...

Personal build: https://travis-ci.com/lihalite/arrow/builds/116979112

pitrou · 2019-06-26T15:02:36Z

I wonder why there's no AppVeyor build. Do you have one on your fork?

ghost · 2019-06-26T15:06:45Z

I manually triggered one: https://ci.appveyor.com/project/lihalite/arrow/builds/25560414

appveyor.yml is set not to trigger builds unless the HEAD commit touches certain files, if I'm reading it right...

pitrou · 2019-06-26T15:08:50Z

Ah, indeed. Ideally this clause would have examined all files in the PR, not only in the last commit, though.

wesm · 2019-06-26T15:47:48Z

Appveyor just doesn't trigger about 5-10% of the time in my experience, probably a bug on their end

wesm · 2019-06-26T15:48:40Z

Here's the running Appveyor build: https://ci.appveyor.com/project/lihalite/arrow/builds/25560411. Will probably wait a bit for a couple more of the main builds to run

ghost · 2019-06-26T18:49:30Z

Looks like tests finally pass, except R on AppVeyor (which I just kicked again).

nealrichardson · 2019-06-26T18:55:23Z

The R failure is unrelated: https://downforeveryoneorjustme.com/cloud.r-project.org

wesm · 2019-06-26T19:19:12Z

Merging, thanks @lihalite and code reviewers!

ghost · 2019-06-26T19:20:24Z

Thanks everyone, this has been quite the journey!

wesm · 2019-06-26T19:21:08Z

@lihalite would you like to start a section about Flight-specific improvements in 0.14 in https://docs.google.com/document/d/1ljkW5tBh7cDfPRg_z6YY-1cbXfXxpfpRUlSatbV8128/edit?usp=sharing?

ghost · 2019-06-26T19:22:12Z

Sure, I'll take a look when I get a chance.

pitrou requested changes May 15, 2019

View reviewed changes

ghost closed this May 17, 2019

ghost reopened this May 22, 2019

pitrou requested changes May 27, 2019

View reviewed changes

python/pyarrow/tests/test_flight.py Outdated Show resolved Hide resolved

python/pyarrow/tests/test_flight.py Outdated Show resolved Hide resolved

python/pyarrow/tests/test_flight.py Outdated Show resolved Hide resolved

python/pyarrow/tests/test_flight.py Outdated Show resolved Hide resolved

pitrou approved these changes Jun 3, 2019

View reviewed changes

wesm reviewed Jun 10, 2019

View reviewed changes

jacques-n reviewed Jun 11, 2019

View reviewed changes

jacques-n reviewed Jun 21, 2019

View reviewed changes

lidavidm added 15 commits June 26, 2019 09:25

Enable non-nested dictionary batches in Flight integration tests

b4dbc44

Add client-side cancelation of DoGet operations

fdaa76e

Replace ARROW_EXPORT with ARROW_FLIGHT_EXPORT

b0f71d9

Pass Flight context to ListActions in Python

0484c33

Disable Flight cancellation tests in CI

ccfef2d

Use ArrowBuf instead of byte[] for Flight metadata

1225b67

Make Netty version consistent with gRPC

7ac44df

Make FlightStream cancellable from acceptPut

1718d9b

Try to always close FlightStream after acceptPut

72c2a3f

Use FlightStreamChunk in Flight/C++

eff2239

Fix Flight integration tests using metadata

85e2169

Fix new CheckStyle violations

c551d85

Mark flaky Flight test

4cebc54

Inline CompletableFuture in Flight acceptPut

8fd99cd

Rework interface for accessing server-sent metadata during DoPut

6f1cd8d

wesm closed this in 63971ad Jun 26, 2019

BryanCutler mentioned this pull request Jun 27, 2019

ARROW-5063: [Java] Followup to fix compilation for TestFlightClient #4714

Closed

asfimport mentioned this pull request Jul 1, 2020

[Flight] Add application metadata field to DoGet #21163

Closed

ARROW-4626: [Flight] Add application-defined metadata to DoGet/DoPut #4282

ARROW-4626: [Flight] Add application-defined metadata to DoGet/DoPut #4282

Conversation

ghost commented May 9, 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

wesm commented May 15, 2019

ghost commented May 15, 2019

ghost commented May 17, 2019

ghost commented May 22, 2019

pitrou left a comment

Choose a reason for hiding this comment

ghost commented May 28, 2019

ghost commented Jun 3, 2019

pitrou commented Jun 3, 2019

ghost commented Jun 4, 2019

ghost commented Jun 7, 2019

wesm commented Jun 7, 2019

wesm Jun 10, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wesm Jun 10, 2019 • edited Loading

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

ghost Jun 11, 2019 • edited by ghost Loading

Choose a reason for hiding this comment

ghost commented Jun 11, 2019 • edited by ghost Loading

ghost commented Jun 13, 2019

ghost commented Jun 13, 2019

wesm commented Jun 14, 2019

ghost commented Jun 14, 2019

codecov-io commented Jun 17, 2019

Codecov Report

ghost commented Jun 17, 2019

jacques-n 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

ghost Jun 22, 2019 • edited by ghost Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ghost commented Jun 21, 2019

jacques-n commented Jun 24, 2019 via email

ghost commented Jun 24, 2019 • edited by ghost Loading

jacques-n commented Jun 24, 2019

wesm commented Jun 24, 2019

ghost commented Jun 26, 2019

pitrou commented Jun 26, 2019

ghost commented Jun 26, 2019

pitrou commented Jun 26, 2019

wesm commented Jun 26, 2019

wesm commented Jun 26, 2019

ghost commented Jun 26, 2019

nealrichardson commented Jun 26, 2019

wesm commented Jun 26, 2019

ghost commented Jun 26, 2019

wesm commented Jun 26, 2019

ghost commented Jun 26, 2019

wesm Jun 10, 2019 •

edited

Loading

wesm Jun 10, 2019 •

edited

Loading

ghost Jun 11, 2019 •

edited by ghost

Loading

ghost commented Jun 11, 2019 •

edited by ghost

Loading

ghost Jun 22, 2019 •

edited by ghost

Loading

ghost commented Jun 24, 2019 •

edited by ghost

Loading