bump tonic to 0.12 and prost to 0.13 for arrow-flight (#6041)

* bump `tonic` to 0.12 and `prost` to 0.13 for `arrow-flight`

Signed-off-by: Bugen Zhao <[email protected]>

* fix example tests

Signed-off-by: Bugen Zhao <[email protected]>

---------

Signed-off-by: Bugen Zhao <[email protected]>

Author: BugenZhao

arrow-flight/Cargo.toml

@@ -44,11 +44,11 @@ bytes = { version = "1", default-features = false }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 once_cell = { version = "1", optional = true }
 paste = { version = "1.0" }
-prost = { version = "0.12.3", default-features = false, features = ["prost-derive"] }
+prost = { version = "0.13.1", default-features = false, features = ["prost-derive"] }
 # For Timestamp type
-prost-types = { version = "0.12.3", default-features = false }
+prost-types = { version = "0.13.1", default-features = false }
 tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "rt-multi-thread"] }
-tonic = { version = "0.11.0", default-features = false, features = ["transport", "codegen", "prost"] }
+tonic = { version = "0.12.0", default-features = false, features = ["transport", "codegen", "prost"] }
 
 # CLI-related dependencies
 anyhow = { version = "1.0", optional = true }
@@ -70,8 +70,9 @@ cli = ["anyhow", "arrow-cast/prettyprint", "clap", "tracing-log", "tracing-subsc
 [dev-dependencies]
 arrow-cast = { workspace = true, features = ["prettyprint"] }
 assert_cmd = "2.0.8"
-http = "0.2.9"
-http-body = "0.4.5"
+http = "1.1.0"
+http-body = "1.0.0"
+hyper-util = "0.1"
 pin-project-lite = "0.2"
 tempfile = "3.3"
 tokio-stream = { version = "0.1", features = ["net"] }

arrow-flight/examples/flight_sql_server.rs

@@ -783,7 +783,8 @@ impl ProstMessageExt for FetchResults {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use futures::TryStreamExt;
+    use futures::{TryFutureExt, TryStreamExt};
+    use hyper_util::rt::TokioIo;
     use std::fs;
     use std::future::Future;
     use std::net::SocketAddr;
@@ -843,7 +844,8 @@ mod tests {
             .serve_with_incoming(stream);
 
         let request_future = async {
-            let connector = service_fn(move |_| UnixStream::connect(path.clone()));
+            let connector =
+                service_fn(move |_| UnixStream::connect(path.clone()).map_ok(TokioIo::new));
             let channel = Endpoint::try_from("http://example.com")
                 .unwrap()
                 .connect_with_connector(connector)

arrow-flight/gen/Cargo.toml

@@ -33,5 +33,5 @@ publish = false
 # Pin specific version of the tonic-build dependencies to avoid auto-generated
 # (and checked in) arrow.flight.protocol.rs from changing
 proc-macro2 = { version = "=1.0.86", default-features = false }
-prost-build = { version = "=0.12.6", default-features = false }
-tonic-build = { version = "=0.11.0", default-features = false, features = ["transport", "prost"] }
+prost-build = { version = "=0.13.1", default-features = false }
+tonic-build = { version = "=0.12.0", default-features = false, features = ["transport", "prost"] }

arrow-flight/src/arrow.flight.protocol.rs

@@ -21,7 +21,7 @@ use std::task::{Context, Poll};
 
 use futures::ready;
 use http::{HeaderValue, Request, Response};
-use http_body::SizeHint;
+use http_body::{Frame, SizeHint};
 use pin_project_lite::pin_project;
 use tower::{Layer, Service};
 
@@ -99,31 +99,19 @@ impl<B: http_body::Body> http_body::Body for WrappedBody<B> {
     type Data = B::Data;
     type Error = B::Error;
 
-    fn poll_data(
-        mut self: Pin<&mut Self>,
+    fn poll_frame(
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
-        self.as_mut().project().inner.poll_data(cx)
-    }
-
-    fn poll_trailers(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<http::header::HeaderMap>, Self::Error>> {
-        let result: Result<Option<http::header::HeaderMap>, Self::Error> =
-            ready!(self.as_mut().project().inner.poll_trailers(cx));
-
-        let mut trailers = http::header::HeaderMap::new();
-        trailers.insert("test-trailer", HeaderValue::from_static("trailer_val"));
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        let mut result = ready!(self.project().inner.poll_frame(cx));
 
-        match result {
-            Ok(Some(mut existing)) => {
-                existing.extend(trailers.iter().map(|(k, v)| (k.clone(), v.clone())));
-                Poll::Ready(Ok(Some(existing)))
+        if let Some(Ok(frame)) = &mut result {
+            if let Some(trailers) = frame.trailers_mut() {
+                trailers.insert("test-trailer", HeaderValue::from_static("trailer_val"));
             }
-            Ok(None) => Poll::Ready(Ok(Some(trailers))),
-            Err(e) => Poll::Ready(Err(e)),
         }
+
+        Poll::Ready(result)
     }
 
     fn is_end_stream(&self) -> bool {

arrow-flight/src/sql/arrow.flight.protocol.sql.rs

@@ -42,11 +42,11 @@ async-trait = { version = "0.1.41", default-features = false }
 clap = { version = "4", default-features = false, features = ["std", "derive", "help", "error-context", "usage"] }
 futures = { version = "0.3", default-features = false }
 hex = { version = "0.4", default-features = false, features = ["std"] }
-prost = { version = "0.12", default-features = false }
+prost = { version = "0.13", default-features = false }
 serde = { version = "1.0", default-features = false, features = ["rc", "derive"] }
 serde_json = { version = "1.0", default-features = false, features = ["std"] }
 tokio = { version = "1.0", default-features = false }
-tonic = { version = "0.11", default-features = false }
+tonic = { version = "0.12", default-features = false }
 tracing-subscriber = { version = "0.3.1", default-features = false, features = ["fmt"], optional = true }
 num = { version = "0.4", default-features = false, features = ["std"] }
 flate2 = { version = "1", default-features = false, features = ["rust_backend"] }