Complete mid-level `FlightClient` #3402

alamb · 2022-12-27T21:06:57Z

~~Draft as it builds on #3391~~

Which issue does this PR close?

Rationale for this change

We are building a mid level flight client that takes care of encoding/decoding details (that we can build FlightSQL client on top of)

What changes are included in this PR?

Implement remaining FlightClient APIs: do_put, do_exchange, get_schema, do_action, list_action, list_flights
Documentation
Tests

Are there any user-facing changes?

New APis

…hema}

alamb · 2022-12-27T21:12:41Z

arrow-flight/src/client.rs

+    ///   .expect("error calling do_put");
+    /// # }
+    /// ```
+    pub async fn do_put<S: Stream<Item = FlightData> + Send + 'static>(


The real value add of this API / client, in my opinion, is this docstring / example and the tests

alamb · 2022-12-31T13:00:32Z

arrow-flight/tests/client.rs

@@ -76,26 +79,6 @@ async fn test_handshake_error() {
    .await;
 }

-#[tokio::test]
-async fn test_handshake_metadata() {


I consolidated metadata testing into all the existing tests, so there is no need for a standalone metadata test

tustvold

Mostly just some minor nits, I would change do_put and do_exchange to accept fallible streams and just blindly propagate through the error

tustvold · 2023-01-04T10:37:01Z

arrow-flight/src/client.rs

+    /// // encode the batch as a stream of `FlightData`
+    /// let flight_data_stream = FlightDataEncoderBuilder::new()
+    ///   .build(futures::stream::iter(vec![Ok(batch)]))
+    ///   // data encoder return Results, but do_put requires FlightData


Perhaps this should accept a Stream<Item=Result<FlightData>>, otherwise I'm not sure how one could use this to stream a fallible computation without panicking?

Hmm, I guess that the input is not fallible? I'm not sure what the semantics is on put if it encounters non-complete data.

Perhaps this should accept a Stream<Item=Result>, otherwise I'm not sure how one could use this to stream a fallible computation without panicking?

I agree -- I was thinking about this too -- I'll see what I can do (though maybe I will do as as a follow on PR)

Tracking in #3462

Proposal in #3464

tustvold · 2023-01-04T10:39:05Z

arrow-flight/src/client.rs

+    /// let mut client = FlightClient::new(channel);
+    ///
+    /// // encode the batch as a stream of `FlightData`
+    /// let flight_data_stream = FlightDataEncoderBuilder::new()


It's probably a bit late now, but looking at this API in use, it is kind of a little strange Stream isn't mentioned at all in the name. i.e. s/Encoder/Stream or something...

This code hasn't been released -- I think we can change the name

Also Codec seems to be a popular choice FlightDataCodec 🤔

arrow-flight/src/client.rs

tustvold · 2023-01-04T10:45:40Z

arrow-flight/tests/client.rs

+            .await
+            .expect("error making request");
+
+        let response: Result<Vec<_>, _> = response_stream.try_collect().await;


unwrap_err? Or is this some lack of Debug issue?

It is lack of debug in Error::External --maybe I can add a Debug impl 🤔

I tried putzing with this -- the issue is not that FlightError is not Debug it is that BoxStream is not.

Here is the error when I used unwrap_err():

279 | let response = client | ________________________^ 280 | | .do_put(futures::stream::iter(input_flight_data.clone())) 281 | | .await | |__________________^ `dyn Stream<Item = std::result::Result<PutResult, FlightError>> + std::marker::Send` cannot be formatted using `{:?}` because it doesn't implement `Debug` 282 | .unwrap_err(); | ---------- required by a bound introduced by this call | = help: the trait `Debug` is not implemented for `dyn Stream<Item = std::result::Result<PutResult, FlightError>> + std::marker::Send` = help: the following other types implement trait `Debug`: (dyn Any + 'static) (dyn Any + Sync + std::marker::Send + 'static) (dyn Any + std::marker::Send + 'static) (dyn tracing_core::field::Value + 'static) = note: required for `Box<dyn Stream<Item = std::result::Result<PutResult, FlightError>> + std::marker::Send>` to implement `Debug` = note: 1 redundant requirement hidden = note: required for `Pin<Box<dyn Stream<Item = std::result::Result<PutResult, FlightError>> + std::marker::Send>>` to implement `Debug`

I tried putzing with the types some but I could not get the traits to be happy

tustvold · 2023-01-04T10:45:53Z

arrow-flight/tests/client.rs

+        let response = client
+            .do_exchange(futures::stream::iter(input_flight_data.clone()))
+            .await;
+        let response = match response {


tustvold · 2023-01-04T10:48:38Z

arrow-flight/tests/common/server.rs

+            .take()
+            .ok_or_else(|| Status::internal("No do_put response configured"))?;
+
+        let stream = futures::stream::iter(response).map_err(|e| e.into());


Suggested change

let stream = futures::stream::iter(response).map_err(|e| e.into());

let stream = futures::stream::iter(response).map_err(Into::into);

tustvold · 2023-01-04T10:49:33Z

arrow-flight/tests/common/server.rs

+
+        let stream = futures::stream::iter(response).map_err(|e| e.into());
+
+        Ok(Response::new(Box::pin(stream) as _))


Suggested change

Ok(Response::new(Box::pin(stream) as _))

Ok(Response::new(stream.boxed()))

I applied these changes to the rest of this file

tustvold · 2023-01-04T10:50:04Z

arrow-flight/tests/client.rs

+            .expect("error making request");
+
+        let response: Result<Vec<_>, _> = response_stream.try_collect().await;
+        let response = match response {


same issue as above

tustvold · 2023-01-04T10:50:27Z

arrow-flight/tests/common/server.rs

+        let stream = futures::stream::iter(response).map_err(|e| e.into());
+
+        Ok(Response::new(Box::pin(stream) as _))


Suggested change

let stream = futures::stream::iter(response).map_err(|e| e.into());

Ok(Response::new(Box::pin(stream) as _))

let stream = futures::stream::iter(response).map_err(Into::into);

Ok(Response::new(stream.boxed()))

tustvold · 2023-01-04T10:52:46Z

arrow-flight/src/client.rs

+    /// ```
+    pub async fn list_flights(
+        &mut self,
+        expression: impl Into<Bytes>,


Is the idea a higher-level FlightSQL client will handle the Any encoding nonsense?

that was the idea -- flight doesn't really say anything about what "expression" should be (aka it doesn't say it will be an encoded protobuf message) FlightSQL does use protobuf encoded messages but I think that should be handled in higher level

arrow-flight/src/client.rs

viirya · 2023-01-04T19:25:28Z

arrow-flight/src/client.rs

    // TODO other methods
-    // list_flights
-    // get_schema
-    // do_put
    // do_action


I think do_action is implemented now?

indeed -- good catch

will remove

…ctions

Co-authored-by: Liang-Chi Hsieh <[email protected]> Co-authored-by: Raphael Taylor-Davies <[email protected]>

… into alamb/more_flight_actions

…ctions

ursabot · 2023-01-05T14:41:32Z

Benchmark runs are scheduled for baseline = 81abc1a and contender = 42ffc3f. 42ffc3f is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped ⚠️ Benchmarking of arrow-rs-commits is not supported on ec2-t3-xlarge-us-east-2] ec2-t3-xlarge-us-east-2
[Skipped ⚠️ Benchmarking of arrow-rs-commits is not supported on test-mac-arm] test-mac-arm
[Skipped ⚠️ Benchmarking of arrow-rs-commits is not supported on ursa-i9-9960x] ursa-i9-9960x
[Skipped ⚠️ Benchmarking of arrow-rs-commits is not supported on ursa-thinkcentre-m75q] ursa-thinkcentre-m75q
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java

alamb · 2023-01-05T15:02:44Z

It appears I forgot some tests here -- added in #3463

github-actions bot added arrow Changes to the arrow crate arrow-flight Changes to the arrow-flight crate labels Dec 27, 2022

xudong963 self-requested a review December 29, 2022 14:09

alamb mentioned this pull request Dec 29, 2022

Implement RecordBatch <--> FlightData encode/decode + tests #3391

Merged

1 task

alamb force-pushed the alamb/more_flight_actions branch from 3d28ca5 to 133d5f0 Compare December 29, 2022 16:13

alamb changed the title ~~Implement FlightClient::do_put and FlightClient::do_exchange~~ Complete FlightClient: do_put, do_exchange, get_schema, do_action, list_action, list_flights Dec 31, 2022

alamb changed the title ~~Complete FlightClient: do_put, do_exchange, get_schema, do_action, list_action, list_flights~~ Complete mid-level FlightClient Dec 31, 2022

alamb force-pushed the alamb/more_flight_actions branch from 133d5f0 to 45701c3 Compare December 31, 2022 12:26

alamb added 2 commits December 31, 2022 07:58

Implement FlightClient::do_put and FlightClient::do_exchange

31e39c3

Implement ArrowClient::{list_flights, list_actions, do_action, get_sc…

352029a

…hema}

alamb force-pushed the alamb/more_flight_actions branch from 45701c3 to 352029a Compare December 31, 2022 12:58

github-actions bot removed the arrow Changes to the arrow crate label Dec 31, 2022

alamb commented Dec 31, 2022

View reviewed changes

alamb marked this pull request as ready for review December 31, 2022 13:00

alamb requested a review from tustvold December 31, 2022 13:19

tustvold approved these changes Jan 4, 2023

View reviewed changes