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

ci/Dockerfile

@@ -6,7 +6,8 @@ ARG RUST_TOOLCHAIN
 
 RUN apt-get update -yy && \
     DEBIAN_FRONTEND=noninteractive apt-get -y install make build-essential cmake protobuf-compiler curl \
-    openssl libssl-dev libcurl4-openssl-dev pkg-config bash openjdk-11-jdk wget unzip git tmux lld -yy
+    openssl libssl-dev libcurl4-openssl-dev pkg-config bash openjdk-11-jdk wget unzip git tmux lld -yy \
+    && rm -rf /var/lib/{apt,dpkg,cache,log}/
 
 SHELL ["/bin/bash", "-c"]
 
@@ -21,14 +22,13 @@ WORKDIR /risingwave
 
 ENV PATH /root/.cargo/bin/:$PATH
 
-# install build tools
-RUN cargo install cargo-llvm-cov cargo-nextest cargo-udeps cargo-hakari cargo-sort cargo-make
-
-# pre-cache required crates
-RUN git clone https://github.com/singularity-data/risingwave && cd risingwave && cargo fetch && cd .. && rm -rf risingwave
-
 # add required rustup components
 RUN rustup component add rustfmt llvm-tools-preview clippy
 
-# install sqllogictest
-RUN cargo install --git https://github.com/risinglightdb/sqllogictest-rs --features bin
+# install build tools
+RUN cargo install cargo-llvm-cov cargo-nextest cargo-udeps cargo-hakari cargo-sort cargo-make cargo-cache \
+    && cargo install --git https://github.com/risinglightdb/sqllogictest-rs --features bin \
+    && cargo cache -a
+
+# pre-cache required crates
+RUN git clone https://github.com/singularity-data/risingwave && cd risingwave && cargo fetch && cd .. && rm -rf risingwave && cargo cache -g

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/docker-compose.yml

@@ -1,7 +1,7 @@
 version: "3.9"
 services:
   rw-build-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v202206091256
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20220610
 #    build:
 #      context: ..
 #      dockerfile: ./ci/Dockerfile

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

e2e_test/batch/functions/concat.slt

@@ -9,12 +9,12 @@ ab
 query T
 select concat(NULL);
 ----
-
+(empty)
 
 query T
 select concat(NULL, NULL);
 ----
-
+(empty)
 
 query T
 select concat(1, 1.01);

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/prost/helpers/src/lib.rs

@@ -25,7 +25,6 @@ use proc_macro2::TokenStream as TokenStream2;
 use proc_macro_error::{proc_macro_error, ResultExt};
 use syn::{DataStruct, DeriveInput};
 
-#[cfg_attr(coverage, no_coverage)]
 mod generate;
 
 // This attribute will be placed before any pb types, including messages and enums.

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]