Update rust vesion to 1.57

ballista-examples/Cargo.toml

@@ -26,7 +26,7 @@ license = "Apache-2.0"
 keywords = [ "arrow", "distributed", "query", "sql" ]
 edition = "2021"
 publish = false
-rust-version = "1.56"
+rust-version = "1.57"
 
 [dependencies]
 datafusion = { path = "../datafusion" }

ballista/rust/client/Cargo.toml

@@ -24,7 +24,7 @@ homepage = "https://github.com/apache/arrow-datafusion"
 repository = "https://github.com/apache/arrow-datafusion"
 authors = ["Apache Arrow <[email protected]>"]
 edition = "2021"
-rust-version = "1.56"
+rust-version = "1.57"
 
 [dependencies]
 ballista-core = { path = "../core", version = "0.6.0" }

ballista/rust/scheduler/src/state/mod.rs

@@ -567,7 +567,7 @@ fn find_unresolved_shuffles(
         Ok(plan
             .children()
             .iter()
-            .map(|child| find_unresolved_shuffles(child))
+            .map(find_unresolved_shuffles)
             .collect::<Result<Vec<_>>>()?
             .into_iter()
             .flatten()

benchmarks/Cargo.toml

@@ -25,7 +25,7 @@ homepage = "https://github.com/apache/arrow-datafusion"
 repository = "https://github.com/apache/arrow-datafusion"
 license = "Apache-2.0"
 publish = false
-rust-version = "1.56"
+rust-version = "1.57"
 
 [features]
 simd = ["datafusion/simd"]

datafusion-cli/Cargo.toml

@@ -24,7 +24,7 @@ keywords = [ "arrow", "datafusion", "ballista", "query", "sql" ]
 license = "Apache-2.0"
 homepage = "https://github.com/apache/arrow-datafusion"
 repository = "https://github.com/apache/arrow-datafusion"
-rust-version = "1.56"
+rust-version = "1.57"
 
 [dependencies]
 clap = "2.33"

datafusion-cli/Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-FROM rust:1.56 as builder
+FROM rust:1.57 as builder
 
 COPY ./datafusion /usr/src/datafusion
 

datafusion-examples/Cargo.toml

@@ -26,7 +26,7 @@ license = "Apache-2.0"
 keywords = [ "arrow", "query", "sql" ]
 edition = "2021"
 publish = false
-rust-version = "1.56"
+rust-version = "1.57"
 
 [[example]]
 name = "avro_sql"

datafusion/Cargo.toml

@@ -31,7 +31,7 @@ include = [
     "Cargo.toml",
 ]
 edition = "2021"
-rust-version = "1.56"
+rust-version = "1.57"
 
 [lib]
 name = "datafusion"

datafusion/src/physical_plan/datetime_expressions.rs

@@ -42,6 +42,7 @@ use arrow::{
 };
 use chrono::prelude::*;
 use chrono::Duration;
+use std::borrow::Borrow;
 
 /// given a function `op` that maps a `&str` to a Result of an arrow native type,
 /// returns a `PrimitiveArray` after the application
@@ -77,7 +78,10 @@ where
         })?;
 
     // first map is the iterator, second is for the `Option<_>`
-    array.iter().map(|x| x.map(|x| op(x)).transpose()).collect()
+    array
+        .iter()
+        .map(|x| x.map(op.borrow()).transpose())
+        .collect()
 }
 
 // given an function that maps a `&str` to a arrow native type,

datafusion/src/physical_plan/expressions/average.rs

@@ -37,8 +37,8 @@ use super::{format_state_name, sum};
 #[derive(Debug)]
 pub struct Avg {
     name: String,
+    #[allow(dead_code)]
     data_type: DataType,
-    nullable: bool,
     expr: Arc<dyn PhysicalExpr>,
 }
 
@@ -73,7 +73,6 @@ impl Avg {
             name: name.into(),
             expr,
             data_type,
-            nullable: true,
         }
     }
 }

datafusion/src/physical_plan/sort_preserving_merge.rs

@@ -346,7 +346,7 @@ struct SortPreservingMergeStream {
     receivers: Vec<mpsc::Receiver<ArrowResult<RecordBatch>>>,
 
     /// Drop helper for tasks feeding the [`receivers`](Self::receivers)
-    drop_helper: AbortOnDropMany<()>,
+    _drop_helper: AbortOnDropMany<()>,
 
     /// For each input stream maintain a dequeue of SortKeyCursor
     ///
@@ -379,7 +379,7 @@ struct SortPreservingMergeStream {
 impl SortPreservingMergeStream {
     fn new(
         receivers: Vec<mpsc::Receiver<ArrowResult<RecordBatch>>>,
-        drop_helper: AbortOnDropMany<()>,
+        _drop_helper: AbortOnDropMany<()>,
         schema: SchemaRef,
         expressions: &[PhysicalSortExpr],
         target_batch_size: usize,
@@ -394,7 +394,7 @@ impl SortPreservingMergeStream {
             schema,
             cursors,
             receivers,
-            drop_helper,
+            _drop_helper,
             column_expressions: expressions.iter().map(|x| x.expr.clone()).collect(),
             sort_options: expressions.iter().map(|x| x.options).collect(),
             target_batch_size,

datafusion/src/physical_plan/string_expressions.rs

@@ -117,10 +117,7 @@ where
     let string_array = downcast_string_arg!(args[0], "string", T);
 
     // first map is the iterator, second is for the `Option<_>`
-    Ok(string_array
-        .iter()
-        .map(|string| string.map(|s| op(s)))
-        .collect())
+    Ok(string_array.iter().map(|string| string.map(&op)).collect())
 }
 
 fn handle<'a, F, R>(args: &'a [ColumnarValue], op: F, name: &str) -> Result<ColumnarValue>

datafusion/src/scalar.rs

@@ -68,7 +68,7 @@ pub enum ScalarValue {
     /// large binary
     LargeBinary(Option<Vec<u8>>),
     /// list of nested ScalarValue (boxed to reduce size_of(ScalarValue))
-    #[allow(clippy::box_vec)]
+    #[allow(clippy::box_collection)]
     List(Option<Box<Vec<ScalarValue>>>, Box<DataType>),
     /// Date stored as a signed 32bit int
     Date32(Option<i32>),
@@ -87,7 +87,7 @@ pub enum ScalarValue {
     /// Interval with DayTime unit
     IntervalDayTime(Option<i64>),
     /// struct of nested ScalarValue (boxed to reduce size_of(ScalarValue))
-    #[allow(clippy::box_vec)]
+    #[allow(clippy::box_collection)]
     Struct(Option<Box<Vec<ScalarValue>>>, Box<Vec<Field>>),
 }
 

datafusion/src/sql/parser.rs

@@ -85,7 +85,7 @@ pub struct CreateExternalTable {
 #[derive(Debug, Clone, PartialEq)]
 pub enum Statement {
     /// ANSI SQL AST node
-    Statement(SQLStatement),
+    Statement(Box<SQLStatement>),
     /// Extension: `CREATE EXTERNAL TABLE`
     CreateExternalTable(CreateExternalTable),
 }
@@ -167,13 +167,17 @@ impl<'a> DFParser<'a> {
                     }
                     _ => {
                         // use the native parser
-                        Ok(Statement::Statement(self.parser.parse_statement()?))
+                        Ok(Statement::Statement(Box::from(
+                            self.parser.parse_statement()?,
+                        )))
                     }
                 }
             }
             _ => {
                 // use the native parser
-                Ok(Statement::Statement(self.parser.parse_statement()?))
+                Ok(Statement::Statement(Box::from(
+                    self.parser.parse_statement()?,
+                )))
             }
         }
     }
@@ -183,7 +187,7 @@ impl<'a> DFParser<'a> {
         if self.parser.parse_keyword(Keyword::EXTERNAL) {
             self.parse_create_external_table()
         } else {
-            Ok(Statement::Statement(self.parser.parse_create()?))
+            Ok(Statement::Statement(Box::from(self.parser.parse_create()?)))
         }
     }
 

datafusion/src/test/exec.rs

@@ -549,7 +549,7 @@ impl ExecutionPlan for BlockingExec {
     async fn execute(&self, _partition: usize) -> Result<SendableRecordBatchStream> {
         Ok(Box::pin(BlockingStream {
             schema: Arc::clone(&self.schema),
-            refs: Arc::clone(&self.refs),
+            _refs: Arc::clone(&self.refs),
         }))
     }
 
@@ -577,7 +577,7 @@ pub struct BlockingStream {
     schema: SchemaRef,
 
     /// Ref-counting helper to check if the stream are still in memory.
-    refs: Arc<()>,
+    _refs: Arc<()>,
 }
 
 impl Stream for BlockingStream {

dev/docker/ballista-base.dockerfile

@@ -23,7 +23,7 @@
 
 
 # Base image extends debian:buster-slim
-FROM rust:1.56.0-buster AS builder
+FROM rust:1.57.0-buster AS builder
 
 RUN apt update && apt -y install musl musl-dev musl-tools libssl-dev openssl
 

python/Cargo.toml

@@ -25,7 +25,7 @@ description = "Build and run queries against data"
 readme = "README.md"
 license = "Apache-2.0"
 edition = "2021"
-rust-version = "1.56"
+rust-version = "1.57"
 
 [dependencies]
 tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread", "sync"] }