build: bump toolchain to 2023-09-09

ci/Dockerfile

@@ -43,7 +43,7 @@ RUN curl -L --proto '=https' --tlsv1.2 -sSf https://raw.githubusercontent.com/ca
 RUN cargo binstall -y --no-symlinks cargo-llvm-cov cargo-nextest cargo-hakari cargo-sort cargo-cache cargo-audit \
     [email protected] \
     [email protected] \
-    && cargo install sccache --locked \
+    sccache@0.5.4 \
     && cargo cache -a \
     && rm -rf "/root/.cargo/registry/index" \
     && rm -rf "/root/.cargo/registry/cache" \

ci/build-ci-image.sh

@@ -13,7 +13,7 @@ cat ../rust-toolchain
 # !!! CHANGE THIS WHEN YOU WANT TO BUMP CI IMAGE !!! #
 #          AND ALSO docker-compose.yml               #
 ######################################################
-export BUILD_ENV_VERSION=v20230821
+export BUILD_ENV_VERSION=v20230829
 
 export BUILD_TAG="public.ecr.aws/x5u3w5h6/rw-build-env:${BUILD_ENV_VERSION}"
 

ci/docker-compose.yml

@@ -71,7 +71,7 @@ services:
       retries: 5
 
   source-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230821
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230829
     depends_on:
       - mysql
       - db
@@ -80,7 +80,7 @@ services:
       - ..:/risingwave
 
   sink-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230821
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230829
     depends_on:
       - mysql
       - db
@@ -90,12 +90,12 @@ services:
       - ..:/risingwave
 
   rw-build-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230821
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230829
     volumes:
       - ..:/risingwave
 
   ci-flamegraph-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230821
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230829
     # NOTE(kwannoel): This is used in order to permit
     # syscalls for `nperf` (perf_event_open),
     # so it can do CPU profiling.
@@ -106,7 +106,7 @@ services:
       - ..:/risingwave
 
   regress-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230821
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230829
     depends_on:
       db:
         condition: service_healthy

ci/rust-toolchain

@@ -1,2 +1,2 @@
 [toolchain]
-channel = "nightly-2023-05-31"
+channel = "nightly-2023-08-21"

ci/scripts/gen-flamegraph.sh

@@ -79,7 +79,7 @@ install_all() {
   git clone https://github.com/gimli-rs/addr2line
   pushd addr2line
   git checkout 0.20.0
-  echo "nightly-2023-04-07" > rust-toolchain
+  echo "nightly-2023-08-21" > rust-toolchain
   cargo b --examples -r
   mv ./target/release/examples/addr2line $(which addr2line)
   popd

rustfmt.toml

@@ -9,4 +9,7 @@ group_imports = "StdExternalCrate"
 reorder_impl_items = true
 reorder_imports = true
 tab_spaces = 4
-wrap_comments = true
+# TODO: Fix it later. It produces too many unwanted changes related with markdown lists now. e.g.,
+# https://github.com/rust-lang/rustfmt/issues/5862
+# https://github.com/rust-lang/rustfmt/issues/5836
+# wrap_comments = true

src/batch/benches/expand.rs

@@ -44,7 +44,7 @@ fn bench_expand(c: &mut Criterion) {
                 let chunk_num = SIZE / chunk_size;
                 b.to_async(&rt).iter_batched(
                     || create_expand_executor(vec![vec![0, 1], vec![2]], chunk_size, chunk_num),
-                    |e| execute_executor(e),
+                    execute_executor,
                     BatchSize::SmallInput,
                 );
             },

src/batch/benches/filter.rs

@@ -46,7 +46,7 @@ fn bench_filter(c: &mut Criterion) {
                 let chunk_num = TOTAL_SIZE / chunk_size;
                 b.to_async(&rt).iter_batched(
                     || create_filter_executor(chunk_size, chunk_num),
-                    |e| execute_executor(e),
+                    execute_executor,
                     BatchSize::SmallInput,
                 );
             },

src/batch/benches/hash_agg.rs

@@ -142,7 +142,7 @@ fn bench_hash_agg(c: &mut Criterion) {
                                 chunk_num,
                             )
                         },
-                        |e| execute_executor(e),
+                        execute_executor,
                         BatchSize::SmallInput,
                     );
                 },

src/batch/benches/limit.rs

@@ -51,7 +51,7 @@ fn bench_limit(c: &mut Criterion) {
                 let chunk_num = SIZE / chunk_size;
                 b.to_async(&rt).iter_batched(
                     || create_limit_executor(chunk_size, chunk_num, 128, 128),
-                    |e| execute_executor(e),
+                    execute_executor,
                     BatchSize::SmallInput,
                 );
             },

src/batch/benches/sort.rs

@@ -80,7 +80,7 @@ fn bench_order_by(c: &mut Criterion) {
                     let chunk_num = SIZE / chunk_size;
                     b.to_async(&rt).iter_batched(
                         || create_order_by_executor(chunk_size, chunk_num, single_column),
-                        |e| execute_executor(e),
+                        execute_executor,
                         BatchSize::SmallInput,
                     );
                 },

src/batch/benches/top_n.rs

@@ -85,7 +85,7 @@ fn bench_top_n(c: &mut Criterion) {
                     let chunk_num = SIZE / chunk_size;
                     b.to_async(&rt).iter_batched(
                         || create_top_n_executor(chunk_size, chunk_num, single_column, 128, 128),
-                        |e| execute_executor(e),
+                        execute_executor,
                         BatchSize::SmallInput,
                     );
                 },

src/batch/benches/utils/mod.rs

@@ -53,7 +53,7 @@ pub fn bench_join(
                                     right_chunk_num,
                                 )
                             },
-                            |e| execute_executor(e),
+                            execute_executor,
                             BatchSize::SmallInput,
                         );
                     },

src/batch/src/executor/group_top_n.rs

@@ -217,7 +217,7 @@ impl<K: HashKey> GroupTopNExecutor<K> {
         }
 
         let mut chunk_builder = DataChunkBuilder::new(self.schema.data_types(), self.chunk_size);
-        for (_, h) in groups.iter_mut() {
+        for (_, h) in &mut groups {
             let mut heap = TopNHeap::empty();
             swap(&mut heap, h);
             for ele in heap.dump() {
@@ -308,7 +308,7 @@ mod tests {
             let mut stream = top_n_executor.execute();
             let res = stream.next().await;
 
-            assert!(matches!(res, Some(_)));
+            assert!(res.is_some());
             if let Some(res) = res {
                 let res = res.unwrap();
                 assert!(
@@ -338,7 +338,7 @@ mod tests {
             }
 
             let res = stream.next().await;
-            assert!(matches!(res, None));
+            assert!(res.is_none());
         }
 
         assert_eq!(0, parent_mem.get_bytes_used());

src/batch/src/executor/merge_sort_exchange.rs

@@ -329,7 +329,7 @@ mod tests {
 
         let mut stream = executor.execute();
         let res = stream.next().await;
-        assert!(matches!(res, Some(_)));
+        assert!(res.is_some());
         if let Some(res) = res {
             let res = res.unwrap();
             assert_eq!(res.capacity(), 3 * num_sources);

src/batch/src/executor/order_by.rs

@@ -196,7 +196,7 @@ mod tests {
 
         let mut stream = order_by_executor.execute();
         let res = stream.next().await;
-        assert!(matches!(res, Some(_)));
+        assert!(res.is_some());
         if let Some(res) = res {
             let res = res.unwrap();
             let col0 = res.column_at(0);
@@ -246,7 +246,7 @@ mod tests {
 
         let mut stream = order_by_executor.execute();
         let res = stream.next().await;
-        assert!(matches!(res, Some(_)));
+        assert!(res.is_some());
         if let Some(res) = res {
             let res = res.unwrap();
             let col0 = res.column_at(0);
@@ -296,7 +296,7 @@ mod tests {
 
         let mut stream = order_by_executor.execute();
         let res = stream.next().await;
-        assert!(matches!(res, Some(_)));
+        assert!(res.is_some());
         if let Some(res) = res {
             let res = res.unwrap();
             let col0 = res.column_at(0);

src/batch/src/executor/sort_over_window.rs

@@ -177,7 +177,7 @@ impl SortOverWindowExecutor {
     ) {
         let mut states =
             WindowStates::new(this.calls.iter().map(create_window_state).try_collect()?);
-        for row in rows.iter() {
+        for row in &*rows {
             for (call, state) in this.calls.iter().zip_eq_fast(states.iter_mut()) {
                 // TODO(rc): batch appending
                 state.append(

src/batch/src/executor/source.rs

@@ -66,7 +66,7 @@ impl BoxedExecutorBuilder for SourceExecutor {
 
         // prepare connector source
         let source_props: HashMap<String, String> =
-            HashMap::from_iter(source_node.properties.clone().into_iter());
+            HashMap::from_iter(source_node.properties.clone());
         let config = ConnectorProperties::extract(source_props)
             .map_err(|e| RwError::from(ConnectorError(e.into())))?;
 

src/batch/src/executor/top_n.rs

@@ -225,7 +225,7 @@ impl Eq for HeapElem {}
 
 impl PartialOrd for HeapElem {
     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
-        self.encoded_row.partial_cmp(&other.encoded_row)
+        Some(self.cmp(other))
     }
 }
 
@@ -350,7 +350,7 @@ mod tests {
         let mut stream = top_n_executor.execute();
         let res = stream.next().await;
 
-        assert!(matches!(res, Some(_)));
+        assert!(res.is_some());
         if let Some(res) = res {
             let res = res.unwrap();
             assert_eq!(res.cardinality(), 3);
@@ -361,7 +361,7 @@ mod tests {
         }
 
         let res = stream.next().await;
-        assert!(matches!(res, None));
+        assert!(res.is_none());
     }
 
     #[tokio::test]
@@ -408,6 +408,6 @@ mod tests {
         let mut stream = top_n_executor.execute();
         let res = stream.next().await;
 
-        assert!(matches!(res, None));
+        assert!(res.is_none());
     }
 }

src/batch/src/monitor/stats.rs

@@ -63,7 +63,7 @@ pub static GLOBAL_BATCH_TASK_METRICS: LazyLock<BatchTaskMetrics> =
 impl BatchTaskMetrics {
     /// The created [`BatchTaskMetrics`] is already registered to the `registry`.
     fn new(registry: &Registry) -> Self {
-        let task_labels = vec!["query_id", "stage_id", "task_id"];
+        let task_labels = ["query_id", "stage_id", "task_id"];
         let mut descs = Vec::with_capacity(8);
 
         let task_first_poll_delay = GaugeVec::new(opts!(