ARROW-11022: [Rust] Upgrade to Tokio 1.0

Upgrades crates to use latest versions of
* tokio (1.0)
* tonic (0.4)
* prost (0.7)

I ran the tests/benchmarks to make sure nothing is broken.

Closes #9230 from Dandandan/tokio_1

Lead-authored-by: Heres, Daniel <danielheres@gmail.com>
Co-authored-by: Jorge C. Leitao <jorgecarleitao@gmail.com>
Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>
Co-authored-by: Dmitry Patsura <zaets28rus@gmail.com>
Co-authored-by: Sutou Kouhei <kou@clear-code.com>
Co-authored-by: Neville Dipale <nevilledips@gmail.com>
Co-authored-by: Yibo Cai <yibo.cai@arm.com>
Co-authored-by: Weston Pace <weston.pace@gmail.com>
Co-authored-by: Maarten A. Breddels <maartenbreddels@gmail.com>
Co-authored-by: mqy <meng.qingyou@gmail.com>
Co-authored-by: Yordan Pavlov <yordan.pavlov@outlook.com>
Co-authored-by: Johannes Müller <JohannesMueller@fico.com>
Co-authored-by: Antoine Pitrou <antoine@python.org>
Co-authored-by: Max Burke <max@urbanlogiq.com>
Co-authored-by: Mahmut Bulut <vertexclique@gmail.com>
Co-authored-by: Joris Van den Bossche <jorisvandenbossche@gmail.com>
Co-authored-by: Ryan Jennings <ryan@ryanj.net>
Co-authored-by: Kenta Murata <mrkn@mrkn.jp>
Co-authored-by: Krisztián Szűcs <szucs.krisztian@gmail.com>
Co-authored-by: Matt Brubeck <mbrubeck@limpet.net>
Co-authored-by: Daniël Heres <danielheres@gmail.com>
Co-authored-by: Jörn Horstmann <joern.horstmann@signavio.com>
Co-authored-by: Benjamin Kietzman <bengilgit@gmail.com>
Signed-off-by: Neville Dipale <nevilledips@gmail.com>

Author: 22 people

rust/arrow-flight/Cargo.toml

@@ -27,15 +27,15 @@ license = "Apache-2.0"
 
 [dependencies]
 arrow = { path = "../arrow", version = "4.0.0-SNAPSHOT" }
-tonic = "0.3"
+tonic = "0.4"
 bytes = "1"
-prost = "0.6"
-prost-derive = "0.6"
-tokio = { version = "0.2", features = ["macros"] }
+prost = "0.7"
+prost-derive = "0.7"
+tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread"] }
 futures = { version = "0.3", default-features = false, features = ["alloc"]}
 
 [build-dependencies]
-tonic-build = "0.3"
+tonic-build = "0.4"
 # Pin specific version of the tonic-build dependencies to avoid auto-generated
 # (and checked in) arrow.flight.protocol.rs from changing
 proc-macro2 = "=1.0.24"

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

@@ -10,8 +10,8 @@ pub struct HandshakeRequest {
     pub protocol_version: u64,
     ///
     /// Arbitrary auth/handshake info.
-    #[prost(bytes, tag = "2")]
-    pub payload: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "2")]
+    pub payload: ::prost::alloc::vec::Vec<u8>,
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct HandshakeResponse {
@@ -21,17 +21,17 @@ pub struct HandshakeResponse {
     pub protocol_version: u64,
     ///
     /// Arbitrary auth/handshake info.
-    #[prost(bytes, tag = "2")]
-    pub payload: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "2")]
+    pub payload: ::prost::alloc::vec::Vec<u8>,
 }
 ///
 /// A message for doing simple auth.
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct BasicAuth {
     #[prost(string, tag = "2")]
-    pub username: std::string::String,
+    pub username: ::prost::alloc::string::String,
     #[prost(string, tag = "3")]
-    pub password: std::string::String,
+    pub password: ::prost::alloc::string::String,
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Empty {}
@@ -41,41 +41,41 @@ pub struct Empty {}
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionType {
     #[prost(string, tag = "1")]
-    pub r#type: std::string::String,
+    pub r#type: ::prost::alloc::string::String,
     #[prost(string, tag = "2")]
-    pub description: std::string::String,
+    pub description: ::prost::alloc::string::String,
 }
 ///
 /// A service specific expression that can be used to return a limited set
 /// of available Arrow Flight streams.
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Criteria {
-    #[prost(bytes, tag = "1")]
-    pub expression: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "1")]
+    pub expression: ::prost::alloc::vec::Vec<u8>,
 }
 ///
 /// An opaque action specific for the service.
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Action {
     #[prost(string, tag = "1")]
-    pub r#type: std::string::String,
-    #[prost(bytes, tag = "2")]
-    pub body: std::vec::Vec<u8>,
+    pub r#type: ::prost::alloc::string::String,
+    #[prost(bytes = "vec", tag = "2")]
+    pub body: ::prost::alloc::vec::Vec<u8>,
 }
 ///
 /// An opaque result returned after executing an action.
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Result {
-    #[prost(bytes, tag = "1")]
-    pub body: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "1")]
+    pub body: ::prost::alloc::vec::Vec<u8>,
 }
 ///
 /// Wrap the result of a getSchema call
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SchemaResult {
     /// schema of the dataset as described in Schema.fbs::Schema.
-    #[prost(bytes, tag = "1")]
-    pub schema: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "1")]
+    pub schema: ::prost::alloc::vec::Vec<u8>,
 }
 ///
 /// The name or tag for a Flight. May be used as a way to retrieve or generate
@@ -87,14 +87,15 @@ pub struct FlightDescriptor {
     ///
     /// Opaque value used to express a command. Should only be defined when
     /// type = CMD.
-    #[prost(bytes, tag = "2")]
-    pub cmd: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "2")]
+    pub cmd: ::prost::alloc::vec::Vec<u8>,
     ///
     /// List of strings identifying a particular dataset. Should only be defined
     /// when type = PATH.
     #[prost(string, repeated, tag = "3")]
-    pub path: ::std::vec::Vec<std::string::String>,
+    pub path: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
 }
+/// Nested message and enum types in `FlightDescriptor`.
 pub mod flight_descriptor {
     ///
     /// Describes what type of descriptor is defined.
@@ -121,17 +122,17 @@ pub mod flight_descriptor {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FlightInfo {
     /// schema of the dataset as described in Schema.fbs::Schema.
-    #[prost(bytes, tag = "1")]
-    pub schema: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "1")]
+    pub schema: ::prost::alloc::vec::Vec<u8>,
     ///
     /// The descriptor associated with this info.
     #[prost(message, optional, tag = "2")]
-    pub flight_descriptor: ::std::option::Option<FlightDescriptor>,
+    pub flight_descriptor: ::core::option::Option<FlightDescriptor>,
     ///
     /// A list of endpoints associated with the flight. To consume the whole
     /// flight, all endpoints must be consumed.
     #[prost(message, repeated, tag = "3")]
-    pub endpoint: ::std::vec::Vec<FlightEndpoint>,
+    pub endpoint: ::prost::alloc::vec::Vec<FlightEndpoint>,
     /// Set these to -1 if unknown.
     #[prost(int64, tag = "4")]
     pub total_records: i64,
@@ -145,29 +146,29 @@ pub struct FlightEndpoint {
     ///
     /// Token used to retrieve this stream.
     #[prost(message, optional, tag = "1")]
-    pub ticket: ::std::option::Option<Ticket>,
+    pub ticket: ::core::option::Option<Ticket>,
     ///
     /// A list of URIs where this ticket can be redeemed. If the list is
     /// empty, the expectation is that the ticket can only be redeemed on the
     /// current service where the ticket was generated.
     #[prost(message, repeated, tag = "2")]
-    pub location: ::std::vec::Vec<Location>,
+    pub location: ::prost::alloc::vec::Vec<Location>,
 }
 ///
 /// A location where a Flight service will accept retrieval of a particular
 /// stream given a ticket.
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Location {
     #[prost(string, tag = "1")]
-    pub uri: std::string::String,
+    pub uri: ::prost::alloc::string::String,
 }
 ///
 /// An opaque identifier that the service can use to retrieve a particular
 /// portion of a stream.
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Ticket {
-    #[prost(bytes, tag = "1")]
-    pub ticket: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "1")]
+    pub ticket: ::prost::alloc::vec::Vec<u8>,
 }
 ///
 /// A batch of Arrow data as part of a stream of batches.
@@ -177,29 +178,29 @@ pub struct FlightData {
     /// The descriptor of the data. This is only relevant when a client is
     /// starting a new DoPut stream.
     #[prost(message, optional, tag = "1")]
-    pub flight_descriptor: ::std::option::Option<FlightDescriptor>,
+    pub flight_descriptor: ::core::option::Option<FlightDescriptor>,
     ///
     /// Header for message data as described in Message.fbs::Message.
-    #[prost(bytes, tag = "2")]
-    pub data_header: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "2")]
+    pub data_header: ::prost::alloc::vec::Vec<u8>,
     ///
     /// Application-defined metadata.
-    #[prost(bytes, tag = "3")]
-    pub app_metadata: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "3")]
+    pub app_metadata: ::prost::alloc::vec::Vec<u8>,
     ///
     /// The actual batch of Arrow data. Preferably handled with minimal-copies
     /// coming last in the definition to help with sidecar patterns (it is
     /// expected that some implementations will fetch this field off the wire
     /// with specialized code to avoid extra memory copies).
-    #[prost(bytes, tag = "1000")]
-    pub data_body: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "1000")]
+    pub data_body: ::prost::alloc::vec::Vec<u8>,
 }
 ///*
 /// The response message associated with the submission of a DoPut.
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PutResult {
-    #[prost(bytes, tag = "1")]
-    pub app_metadata: std::vec::Vec<u8>,
+    #[prost(bytes = "vec", tag = "1")]
+    pub app_metadata: ::prost::alloc::vec::Vec<u8>,
 }
 #[doc = r" Generated client implementations."]
 pub mod flight_service_client {
@@ -1010,6 +1011,7 @@ pub mod flight_service_server {
                     Ok(http::Response::builder()
                         .status(200)
                         .header("grpc-status", "12")
+                        .header("content-type", "application/grpc")
                         .body(tonic::body::BoxBody::empty())
                         .unwrap())
                 }),

rust/benchmarks/Cargo.toml

@@ -35,7 +35,7 @@ arrow = { path = "../arrow" }
 parquet = { path = "../parquet" }
 datafusion = { path = "../datafusion" }
 structopt = { version = "0.3", default-features = false }
-tokio = { version = "0.2", features = ["macros", "rt-core", "rt-threaded"] }
+tokio = { version = "^1.0", features = ["macros", "rt", "rt-multi-thread"] }
 futures = "0.3"
 env_logger = "^0.8"
 snmalloc-rs = {version = "0.2", optional = true, features= ["cache-friendly"] }

rust/datafusion/Cargo.toml

@@ -59,7 +59,7 @@ chrono = "0.4"
 async-trait = "0.1.41"
 futures = "0.3"
 pin-project-lite= "^0.2.0"
-tokio = { version = "0.2", features = ["macros", "blocking", "rt-core", "rt-threaded", "sync"] }
+tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread", "sync"] }
 log = "^0.4"
 md-5 = "^0.9.1"
 sha2 = "^0.9.1"
@@ -69,9 +69,9 @@ ordered-float = "2.0"
 rand = "0.8"
 criterion = "0.3"
 tempfile = "3"
-prost = "0.6"
+prost = "0.7"
 arrow-flight = { path = "../arrow-flight", version = "4.0.0-SNAPSHOT" }
-tonic = "0.3"
+tonic = "0.4"
 
 [[bench]]
 name = "aggregate_query_sql"

rust/datafusion/benches/aggregate_query_sql.rs

@@ -44,7 +44,7 @@ pub fn seedable_rng() -> StdRng {
 }
 
 fn query(ctx: Arc<Mutex<ExecutionContext>>, sql: &str) {
-    let mut rt = Runtime::new().unwrap();
+    let rt = Runtime::new().unwrap();
 
     // execute the query
     let df = ctx.lock().unwrap().sql(&sql).unwrap();

rust/datafusion/benches/math_query_sql.rs

@@ -37,7 +37,7 @@ use datafusion::datasource::MemTable;
 use datafusion::execution::context::ExecutionContext;
 
 fn query(ctx: Arc<Mutex<ExecutionContext>>, sql: &str) {
-    let mut rt = Runtime::new().unwrap();
+    let rt = Runtime::new().unwrap();
 
     // execute the query
     let df = ctx.lock().unwrap().sql(&sql).unwrap();

rust/datafusion/benches/sort_limit_query_sql.rs

@@ -32,7 +32,7 @@ use datafusion::execution::context::ExecutionContext;
 use tokio::runtime::Runtime;
 
 fn query(ctx: Arc<Mutex<ExecutionContext>>, sql: &str) {
-    let mut rt = Runtime::new().unwrap();
+    let rt = Runtime::new().unwrap();
 
     // execute the query
     let df = ctx.lock().unwrap().sql(&sql).unwrap();
@@ -66,7 +66,7 @@ fn create_context() -> Arc<Mutex<ExecutionContext>> {
     )
     .unwrap();
 
-    let mut rt = Runtime::new().unwrap();
+    let rt = Runtime::new().unwrap();
 
     let ctx_holder: Arc<Mutex<Vec<Arc<Mutex<ExecutionContext>>>>> =
         Arc::new(Mutex::new(vec![]));

rust/datafusion/src/physical_plan/coalesce_batches.rs

@@ -235,7 +235,7 @@ mod tests {
     use arrow::array::UInt32Array;
     use arrow::datatypes::{DataType, Field, Schema};
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test(flavor = "multi_thread")]
     async fn test_concat_batches() -> Result<()> {
         let schema = test_schema();
         let partition = create_vec_batches(&schema, 10)?;

rust/datafusion/src/physical_plan/repartition.rs

@@ -243,7 +243,7 @@ mod tests {
     use arrow::datatypes::{DataType, Field, Schema};
     use arrow::record_batch::RecordBatch;
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test(flavor = "multi_thread")]
     async fn one_to_many_round_robin() -> Result<()> {
         // define input partitions
         let schema = test_schema();
@@ -263,7 +263,7 @@ mod tests {
         Ok(())
     }
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test(flavor = "multi_thread")]
     async fn many_to_one_round_robin() -> Result<()> {
         // define input partitions
         let schema = test_schema();
@@ -280,7 +280,7 @@ mod tests {
         Ok(())
     }
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test(flavor = "multi_thread")]
     async fn many_to_many_round_robin() -> Result<()> {
         // define input partitions
         let schema = test_schema();

rust/integration-testing/Cargo.toml

@@ -36,10 +36,10 @@ async-trait = "0.1.41"
 clap = "2.33"
 futures = "0.3"
 hex = "0.4"
-prost = "0.6"
+prost = "0.7"
 serde = { version = "1.0", features = ["rc"] }
 serde_derive = "1.0"
 serde_json = { version = "1.0", features = ["preserve_order"] }
-tokio = { version = "0.2", features = ["macros", "rt-core", "rt-threaded"] }
-tonic = "0.3"
+tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread"] }
+tonic = "0.4"
 tracing-subscriber = { version = "0.2.15", optional = true }