clippy

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

src/interchange/src/envelopes.rs

@@ -52,14 +52,14 @@ where
                         let mut afters = vec![];
 
                         let mut cursor = batch.cursor();
-                        while cursor.key_valid(&batch) {
-                            let k = cursor.key(&batch);
+                        while cursor.key_valid(batch) {
+                            let k = cursor.key(batch);
 
                             // Partition updates into retractions (befores)
                             // and insertions (afters).
-                            while cursor.val_valid(&batch) {
-                                let v = cursor.val(&batch);
-                                cursor.map_times(&batch, |t, diff| {
+                            while cursor.val_valid(batch) {
+                                let v = cursor.val(batch);
+                                cursor.map_times(batch, |t, diff| {
                                     let diff = Tr::owned_diff(diff);
                                     let update = (
                                         Tr::owned_time(t),
@@ -72,7 +72,7 @@ where
                                         afters.push(update);
                                     }
                                 });
-                                cursor.step_val(&batch);
+                                cursor.step_val(batch);
                             }
 
                             // Sort by timestamp.
@@ -112,7 +112,7 @@ where
                                 session.give(((Tr::owned_key(k), group), t, Diff::ONE));
                             }
 
-                            cursor.step_key(&batch);
+                            cursor.step_key(batch);
                         }
                     }
                 });

src/repr/src/row.rs

@@ -496,7 +496,7 @@ mod columnar {
         fn extend_from_self(&mut self, other: Self::Borrowed<'_>, range: Range<usize>) {
             if !range.is_empty() {
                 // Imported bounds will be relative to this starting offset.
-                let values_len = self.values.len() as u64;
+                let values_len: u64 = self.values.len().try_into().expect("must fit");
 
                 // Push all bytes that we can, all at once.
                 let other_lower = if range.start == 0 {
@@ -505,8 +505,11 @@ mod columnar {
                     other.bounds.index_as(range.start - 1)
                 };
                 let other_upper = other.bounds.index_as(range.end - 1);
-                self.values
-                    .extend_from_self(other.values, other_lower as usize..other_upper as usize);
+                self.values.extend_from_self(
+                    other.values,
+                    usize::try_from(other_lower).expect("must fit")
+                        ..usize::try_from(other_upper).expect("must fit"),
+                );
 
                 // Each bound needs to be shifted by `values_len - other_lower`.
                 if values_len == other_lower {

src/repr/src/timestamp.rs

@@ -149,7 +149,7 @@ mod columnar_timestamp {
     impl columnar::Container for Timestamps<Vec<Timestamp>> {
         #[inline(always)]
         fn extend_from_self(&mut self, other: Self::Borrowed<'_>, range: Range<usize>) {
-            self.0.extend_from_self(&other.0, range)
+            self.0.extend_from_self(other.0, range)
         }
         #[inline(always)]
         fn reserve_for<'a, I>(&mut self, selves: I)

src/storage/examples/upsert_open_loop.rs

@@ -595,7 +595,7 @@ async fn run_benchmark(
                 upsert_stream.sink(
                     Exchange::new(move |_| u64::cast_from(chosen_worker)),
                     &format!("source-{source_id}-counter"),
-                    move |input| {
+                    move |(input, input_frontier)| {
                         if !active_worker {
                             return;
                         }
@@ -611,7 +611,7 @@ async fn run_benchmark(
                             }
                         });
 
-                        if input.frontier().is_empty() {
+                        if input_frontier.is_empty() {
                             assert_eq!(num_records_total, num_additions);
                             info!(
                                 "Processing source {source_id} finished \
@@ -621,9 +621,9 @@ async fn run_benchmark(
                             );
                         } else if PartialOrder::less_than(
                             &frontier.borrow(),
-                            &input.frontier().frontier(),
+                            &input_frontier.frontier(),
                         ) {
-                            frontier = input.frontier().frontier().to_owned();
+                            frontier = input_frontier.frontier().to_owned();
                             let data_timestamp = frontier.clone().into_option().unwrap();
                             let elapsed = start.elapsed();
 
@@ -916,7 +916,8 @@ where
     let mut upsert_op =
         AsyncOperatorBuilder::new(format!("source-{source_id}-upsert"), scope.clone());
 
-    let (output, output_stream): (_, Stream<_, (Vec<u8>, Vec<u8>, i32)>) = upsert_op.new_output();
+    let (output, output_stream): (_, Stream<_, (Vec<u8>, Vec<u8>, i32)>) =
+        upsert_op.new_output::<CapacityContainerBuilder<_>>();
     let mut input = upsert_op.new_input_for(
         source_stream,
         Exchange::new(|d: &(Vec<u8>, Vec<u8>)| d.0.hashed()),
@@ -1014,7 +1015,8 @@ where
     let mut upsert_op =
         AsyncOperatorBuilder::new(format!("source-{source_id}-upsert"), scope.clone());
 
-    let (output, output_stream): (_, Stream<_, (Vec<u8>, Vec<u8>, i32)>) = upsert_op.new_output();
+    let (output, output_stream): (_, Stream<_, (Vec<u8>, Vec<u8>, i32)>) =
+        upsert_op.new_output::<CapacityContainerBuilder<_>>();
     let mut input = upsert_op.new_input_for(
         source_stream,
         Exchange::new(|d: &(Vec<u8>, Vec<u8>)| d.0.hashed()),

src/timely-util/benches/reclock.rs

@@ -105,7 +105,7 @@ fn bench_reclock_simple(c: &mut Criterion) {
                                     data_cap.downgrade(&Partitioned::new_singleton(0, ts - 1));
                                 }
                                 if ts.is_multiple_of(data_interval) {
-                                    data.session(data_cap.clone()).give((
+                                    data.activate().session(&data_cap).give((
                                         ts,
                                         Partitioned::new_singleton(0, ts),
                                         Diff::ONE,