make nightly clippy happy (#2186)

apache · Apr 11, 2022 · 7558a55 · 7558a55
1 parent 73ed545
commit 7558a55
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 22 deletions.
diff --git a/ballista/rust/scheduler/src/state/persistent_state.rs b/ballista/rust/scheduler/src/state/persistent_state.rs
@@ -124,28 +124,33 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan>
             .get_from_prefix(&get_stage_prefix(&self.namespace))
             .await?;
 
+        let mut tmp_stages: HashMap<StageKey, Arc<dyn ExecutionPlan>> = HashMap::new();
+        {
+            for (key, entry) in entries {
+                let (job_id, stage_id) = extract_stage_id_from_stage_key(&key).unwrap();
+                let session_id = self
+                    .get_session_from_job(&job_id)
+                    .expect("session id does not exist for job");
+                let session_ctx = self
+                    .session_context_registry
+                    .lookup_session(&session_id)
+                    .await
+                    .expect("SessionContext does not exist in SessionContextRegistry.");
+                let value = U::try_decode(&entry)?;
+                let runtime = session_ctx.runtime_env();
+                let plan = value.try_into_physical_plan(
+                    session_ctx.deref(),
+                    runtime.deref(),
+                    self.codec.physical_extension_codec(),
+                )?;
+
+                tmp_stages.insert((job_id, stage_id), plan);
+            }
+        }
         let mut stages = self.stages.write();
-        for (key, entry) in entries {
-            let (job_id, stage_id) = extract_stage_id_from_stage_key(&key).unwrap();
-            let session_id = self
-                .get_session_from_job(&job_id)
-                .expect("session id does not exist for job");
-            let session_ctx = self
-                .session_context_registry
-                .lookup_session(&session_id)
-                .await
-                .expect("SessionContext does not exist in SessionContextRegistry.");
-            let value = U::try_decode(&entry)?;
-            let runtime = session_ctx.runtime_env();
-            let plan = value.try_into_physical_plan(
-                session_ctx.deref(),
-                runtime.deref(),
-                self.codec.physical_extension_codec(),
-            )?;
-
-            stages.insert((job_id, stage_id), plan);
+        for tmp_stage in tmp_stages {
+            stages.insert(tmp_stage.0, tmp_stage.1);
         }
-
         Ok(())
     }
 

diff --git a/datafusion/physical-expr/src/expressions/lead_lag.rs b/datafusion/physical-expr/src/expressions/lead_lag.rs
@@ -151,11 +151,11 @@ fn shift_with_default_value(
         create_empty_array(value, array.data_type(), array.len())
     } else {
         let slice_offset = (-offset).clamp(0, value_len) as usize;
-        let length = array.len() - offset.abs() as usize;
+        let length = array.len() - offset.unsigned_abs() as usize;
         let slice = array.slice(slice_offset, length);
 
         // Generate array with remaining `null` items
-        let nulls = offset.abs() as usize;
+        let nulls = offset.unsigned_abs() as usize;
         let default_values = create_empty_array(value, slice.data_type(), nulls)?;
         // Concatenate both arrays, add nulls after if shift > 0 else before
         if offset > 0 {