chore(build): bump toolchain to 2022-06-09

ci/build-ci-image.sh

@@ -9,8 +9,9 @@ cd "$DIR"
 cat ../rust-toolchain
 # shellcheck disable=SC2155
 export RUST_TOOLCHAIN=$(cat ../rust-toolchain)
-export BUILD_ENV_VERSION=v20220609
+export BUILD_ENV_VERSION=v20220610
 export BUILD_TAG="public.ecr.aws/x5u3w5h6/rw-build-env:${BUILD_ENV_VERSION}"
 
 docker build -t ${BUILD_TAG} --build-arg "RUST_TOOLCHAIN=${RUST_TOOLCHAIN}" .
+aws ecr-public get-login-password --region us-east-1 | docker login --username AWS --password-stdin public.ecr.aws/x5u3w5h6
 docker push ${BUILD_TAG}

ci/scripts/unit-test.sh

@@ -13,7 +13,7 @@ cargo doc --document-private-items --no-deps
 
 echo "--- Run unit tests with coverage"
 # use tee to disable progress bar
-cargo llvm-cov nextest --lcov --output-path lcov.info --features failpoints -- --no-fail-fast | tee
+cargo llvm-cov nextest --lcov --output-path lcov.info --features failpoints -- --no-fail-fast 2> >(tee)
 
 echo "--- Run doctest"
 cargo test --doc

rust-toolchain

@@ -1 +1 @@
-nightly-2022-05-24
+nightly-2022-06-09

src/connector/src/kinesis/source/reader.rs

@@ -150,13 +150,11 @@ impl KinesisSplitReader {
     async fn get_records(
         &mut self,
     ) -> core::result::Result<GetRecordsOutput, SdkError<GetRecordsError>> {
-        let resp = self
-            .client
+        self.client
             .get_records()
             .set_shard_iterator(self.shard_iter.take())
             .send()
-            .await;
-        resp
+            .await
     }
 }
 

src/expr/src/vector_op/agg/general_sorted_grouper.rs

@@ -111,7 +111,7 @@ impl EqGroups {
         use std::collections::BinaryHeap;
         let mut heap = BinaryHeap::new();
         for (ci, column) in columns.iter().enumerate() {
-            if let Some(ri) = column.starting_indices().get(0) {
+            if let Some(ri) = column.starting_indices().first() {
                 heap.push(Reverse((ri, ci, 0)));
             }
         }

src/expr/src/vector_op/ascii.rs

@@ -16,7 +16,7 @@ use crate::Result;
 
 #[inline(always)]
 pub fn ascii(s: &str) -> Result<i32> {
-    Ok(s.as_bytes().get(0).map(|x| *x as i32).unwrap_or(0))
+    Ok(s.as_bytes().first().map(|x| *x as i32).unwrap_or(0))
 }
 
 #[cfg(test)]

src/frontend/src/binder/expr/mod.rs

@@ -204,7 +204,7 @@ impl Binder {
         if return_type.is_numeric() {
             return Ok(expr);
         }
-        return Err(ErrorCode::InvalidInputSyntax(format!("+ {:?}", return_type)).into());
+        Err(ErrorCode::InvalidInputSyntax(format!("+ {:?}", return_type)).into())
     }
 
     pub(super) fn bind_trim(

src/frontend/src/expr/function_call.rs

@@ -57,7 +57,7 @@ impl std::fmt::Debug for FunctionCall {
                 ExprType::Cast => {
                     assert_eq!(self.inputs.len(), 1);
                     self.inputs[0].fmt(f)?;
-                    return write!(f, "::{:?}", self.return_type);
+                    write!(f, "::{:?}", self.return_type)
                 }
                 ExprType::Add => debug_binary_op(f, "+", &self.inputs),
                 ExprType::Subtract => debug_binary_op(f, "-", &self.inputs),

src/frontend/src/scheduler/distributed/stage.rs

@@ -393,7 +393,7 @@ impl StageRunner {
                 let children = execution_plan_node
                     .children
                     .iter()
-                    .map(|e| self.convert_plan_node(&*e, task_id))
+                    .map(|e| self.convert_plan_node(e, task_id))
                     .collect();
 
                 PlanNodeProst {

src/frontend/src/scheduler/local.rs

@@ -114,7 +114,7 @@ impl LocalQueryExecution {
                 let children = execution_plan_node
                     .children
                     .iter()
-                    .map(|e| self.convert_plan_node(&*e))
+                    .map(|e| self.convert_plan_node(e))
                     .collect::<Result<Vec<PlanNodeProst>>>()?;
 
                 Ok(PlanNodeProst {

src/frontend/test_runner/src/lib.rs

@@ -490,12 +490,12 @@ fn check_err(ctx: &str, expected_err: &Option<String>, actual_err: &Option<Strin
             if expected_err == actual_err {
                 Ok(())
             } else {
-                return Err(anyhow!(
+                Err(anyhow!(
                     "Expected {context} error: {}\n  Actual {context} error: {}",
                     expected_err,
                     actual_err,
                     context = ctx
-                ));
+                ))
             }
         }
     }

src/meta/src/barrier/command.rs

@@ -219,7 +219,7 @@ where
                 table_fragments,
                 dispatches,
                 table_sink_map,
-                source_state: _source_state,
+                source_state: _,
             } => {
                 let mut dependent_table_actors = Vec::with_capacity(table_sink_map.len());
                 for (table_id, actors) in table_sink_map {

src/meta/src/barrier/recovery.rs

@@ -116,14 +116,14 @@ where
         .expect("Retry until recovery success.");
         debug!("recovery success");
 
-        return (
+        (
             new_epoch,
             self.fragment_manager.all_chain_actor_ids().await,
             responses
                 .into_iter()
                 .flat_map(|r| r.create_mview_progress)
                 .collect(),
-        );
+        )
     }
 
     /// Sync all sources in compute nodes, the local source manager in compute nodes may be dirty

src/prost/helpers/src/generate.rs

@@ -124,10 +124,10 @@ pub fn implement(field: &Field) -> TokenStream2 {
         }
     }
 
-    return quote! {
+    quote! {
         #[inline(always)]
         pub fn #getter_fn_name(&self) -> &#ty {
             &self.#field_name
         }
-    };
+    }
 }

src/sqlparser/src/parser.rs

@@ -1356,7 +1356,7 @@ impl Parser {
         let all = self.parse_keyword(Keyword::ALL);
         let distinct = self.parse_keyword(Keyword::DISTINCT);
         if all && distinct {
-            return parser_err!("Cannot specify both ALL and DISTINCT".to_string());
+            parser_err!("Cannot specify both ALL and DISTINCT".to_string())
         } else {
             Ok(distinct)
         }

src/storage/src/hummock/compactor_tests.rs

@@ -48,15 +48,15 @@ mod tests {
             ..Default::default()
         });
         let sstable_store = mock_sstable_store();
-        let storage = HummockStorage::with_default_stats(
+
+        HummockStorage::with_default_stats(
             options.clone(),
             sstable_store,
             hummock_meta_client.clone(),
             Arc::new(StateStoreMetrics::unused()),
         )
         .await
-        .unwrap();
-        storage
+        .unwrap()
     }
 
     #[tokio::test]

src/storage/src/hummock/iterator/forward_user.rs

@@ -249,8 +249,8 @@ impl UserIterator {
         // Handle multi-version
         self.last_key.clear();
         // Handle range scan when key > end_key
-        let res = self.next().await;
-        res
+
+        self.next().await
     }
 
     /// Indicates whether the iterator can be used.

src/storage/src/table/cell_based_table.rs

@@ -64,6 +64,7 @@ pub struct CellBasedTable<S: StateStore> {
     column_ids: Vec<ColumnId>,
 
     /// Statistics.
+    #[allow(dead_code)]
     stats: Arc<StateStoreMetrics>,
 
     /// Indices of distribution keys in pk for computing value meta. None if value meta is not
@@ -337,13 +338,7 @@ impl<S: StateStore> CellBasedTable<S> {
 
     // The returned iterator will iterate data from a snapshot corresponding to the given `epoch`
     pub async fn iter(&self, epoch: u64) -> StorageResult<CellBasedTableRowIter<S>> {
-        CellBasedTableRowIter::new(
-            &self.keyspace,
-            self.column_descs.clone(),
-            epoch,
-            self.stats.clone(),
-        )
-        .await
+        CellBasedTableRowIter::new(&self.keyspace, self.column_descs.clone(), epoch).await
     }
 
     pub async fn iter_with_pk(
@@ -355,7 +350,6 @@ impl<S: StateStore> CellBasedTable<S> {
             self.keyspace.clone(),
             self.column_descs.clone(),
             epoch,
-            self.stats.clone(),
             pk_descs,
         )
         .await
@@ -432,7 +426,6 @@ impl<S: StateStore> CellBasedTable<S> {
             self.column_descs.clone(),
             (start_key, end_key),
             epoch,
-            self.stats.clone(),
         )
         .await
     }
@@ -458,7 +451,6 @@ impl<S: StateStore> CellBasedTable<S> {
             self.column_descs.clone(),
             (start_key, next_key),
             epoch,
-            self.stats.clone(),
         )
         .await
     }
@@ -488,16 +480,13 @@ pub struct CellBasedTableRowIter<S: StateStore> {
     iter: StripPrefixIterator<S::Iter>,
     /// Cell-based row deserializer
     cell_based_row_deserializer: CellBasedRowDeserializer,
-    /// Statistics
-    _stats: Arc<StateStoreMetrics>,
 }
 
 impl<S: StateStore> CellBasedTableRowIter<S> {
     pub(super) async fn new(
         keyspace: &Keyspace<S>,
         table_descs: Vec<ColumnDesc>,
         epoch: u64,
-        _stats: Arc<StateStoreMetrics>,
     ) -> StorageResult<Self> {
         keyspace.state_store().wait_epoch(epoch).await?;
 
@@ -508,7 +497,6 @@ impl<S: StateStore> CellBasedTableRowIter<S> {
         let iter = Self {
             iter,
             cell_based_row_deserializer,
-            _stats,
         };
         Ok(iter)
     }
@@ -518,7 +506,6 @@ impl<S: StateStore> CellBasedTableRowIter<S> {
         table_descs: Vec<ColumnDesc>,
         serialized_pk_bounds: R,
         epoch: u64,
-        _stats: Arc<StateStoreMetrics>,
     ) -> StorageResult<Self>
     where
         R: RangeBounds<B> + Send,
@@ -534,7 +521,6 @@ impl<S: StateStore> CellBasedTableRowIter<S> {
         let iter = Self {
             iter,
             cell_based_row_deserializer,
-            _stats,
         };
         Ok(iter)
     }
@@ -599,11 +585,9 @@ impl<S: StateStore> DedupPkCellBasedTableRowIter<S> {
         keyspace: Keyspace<S>,
         table_descs: Vec<ColumnDesc>,
         epoch: u64,
-        _stats: Arc<StateStoreMetrics>,
         pk_descs: &[OrderedColumnDesc],
     ) -> StorageResult<Self> {
-        let inner =
-            CellBasedTableRowIter::new(&keyspace, table_descs.clone(), epoch, _stats).await?;
+        let inner = CellBasedTableRowIter::new(&keyspace, table_descs.clone(), epoch).await?;
 
         let (data_types, order_types) = pk_descs
             .iter()

src/stream/src/executor/managed_state/aggregation/string_agg.rs

@@ -291,7 +291,8 @@ mod tests {
             .map(|(ord, idx)| OrderPair::new(idx, ord))
             .collect::<Vec<_>>();
         let sort_key_serializer = OrderedArraysSerializer::new(order_pairs);
-        let managed_state = ManagedStringAggState::new(
+
+        ManagedStringAggState::new(
             keyspace.clone(),
             row_count,
             sort_key_indices,
@@ -300,8 +301,7 @@ mod tests {
             sort_key_serializer,
         )
         .await
-        .unwrap();
-        managed_state
+        .unwrap()
     }
 
     #[tokio::test]