fmt

Author: osipovartem

datafusion/execution/src/disk_manager.rs

@@ -391,7 +391,6 @@ impl RefCountedTempFile {
         self.current_file_disk_usage.load(Ordering::Relaxed)
     }
 
-
     pub fn clone_refcounted(&self) -> Result<Self> {
         let reopened = std::fs::File::open(self.path())?;
         let temp_path = TempPath::from_path(self.path());

datafusion/physical-plan/src/joins/grace_hash_join/exec.rs

@@ -20,9 +20,7 @@ use crate::filter_pushdown::{
     ChildPushdownResult, FilterDescription, FilterPushdownPhase,
     FilterPushdownPropagation,
 };
-use crate::joins::utils::{
-    reorder_output_after_swap, swap_join_projection, OnceFut,
-};
+use crate::joins::utils::{reorder_output_after_swap, swap_join_projection, OnceFut};
 use crate::joins::{JoinOn, JoinOnRef, PartitionMode};
 use crate::projection::{
     try_embed_projection, try_pushdown_through_join, EmbeddedProjection, JoinData,
@@ -33,12 +31,12 @@ use crate::{
     common::can_project,
     joins::utils::{
         build_join_schema, check_join_is_valid, estimate_join_statistics,
-        symmetric_join_output_partitioning,
-        BuildProbeJoinMetrics, ColumnIndex, JoinFilter,
+        symmetric_join_output_partitioning, BuildProbeJoinMetrics, ColumnIndex,
+        JoinFilter,
     },
     metrics::{ExecutionPlanMetricsSet, MetricsSet},
-    DisplayAs, DisplayFormatType, Distribution, ExecutionPlan,
-    PlanProperties, SendableRecordBatchStream, Statistics,
+    DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, PlanProperties,
+    SendableRecordBatchStream, Statistics,
 };
 use crate::{ExecutionPlanProperties, SpillManager};
 use std::fmt;
@@ -52,8 +50,7 @@ use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::{
-    internal_err, plan_err, project_schema, JoinSide, JoinType,
-    NullEquality, Result,
+    internal_err, plan_err, project_schema, JoinSide, JoinType, NullEquality, Result,
 };
 use datafusion_execution::TaskContext;
 use datafusion_functions_aggregate_common::min_max::{MaxAccumulator, MinAccumulator};
@@ -670,7 +667,7 @@ impl ExecutionPlan for GraceHashJoinExec {
                 spill_right_clone,
                 partition,
             )
-                .await?;
+            .await?;
             accumulator_clone
                 .report_partition(partition, left_idx.clone(), right_idx.clone())
                 .await;
@@ -730,11 +727,11 @@ impl ExecutionPlan for GraceHashJoinExec {
         }
 
         if let Some(JoinData {
-                        projected_left_child,
-                        projected_right_child,
-                        join_filter,
-                        join_on,
-                    }) = try_pushdown_through_join(
+            projected_left_child,
+            projected_right_child,
+            join_filter,
+            join_on,
+        }) = try_pushdown_through_join(
             projection,
             self.left(),
             self.right(),
@@ -820,7 +817,7 @@ impl ExecutionPlan for GraceHashJoinExec {
         let mut result = FilterPushdownPropagation::if_any(child_pushdown_result.clone());
         assert_eq!(child_pushdown_result.self_filters.len(), 2); // Should always be 2, we have 2 children
         let right_child_self_filters = &child_pushdown_result.self_filters[1]; // We only push down filters to the right child
-        // We expect 0 or 1 self filters
+                                                                               // We expect 0 or 1 self filters
         if let Some(filter) = right_child_self_filters.first() {
             // Note that we don't check PushdDownPredicate::discrimnant because even if nothing said
             // "yes, I can fully evaluate this filter" things might still use it for statistics -> it's worth updating
@@ -855,7 +852,6 @@ impl ExecutionPlan for GraceHashJoinExec {
     }
 }
 
-
 #[allow(clippy::too_many_arguments)]
 pub async fn partition_and_spill(
     random_state: RandomState,
@@ -883,7 +879,7 @@ pub async fn partition_and_spill(
         &join_metrics,
         enable_dynamic_filter_pushdown,
     )
-        .await?;
+    .await?;
 
     // RIGHT side partitioning
     let right_index = partition_and_spill_one_side(
@@ -896,7 +892,7 @@ pub async fn partition_and_spill(
         &join_metrics,
         enable_dynamic_filter_pushdown,
     )
-        .await?;
+    .await?;
     Ok((left_index, right_index))
 }
 

datafusion/physical-plan/src/spill/in_memory_spill_buffer.rs

@@ -1,3 +1,20 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
 use crate::memory::MemoryStream;
 use crate::spill::spill_manager::GetSlicedSize;
 use arrow::array::RecordBatch;

datafusion/physical-plan/src/spill/spill_manager.rs

@@ -17,11 +17,11 @@
 
 //! Define the `SpillManager` struct, which is responsible for reading and writing `RecordBatch`es to raw files based on the provided configurations.
 
-use std::slice;
 use arrow::array::StringViewArray;
 use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
 use datafusion_execution::runtime_env::RuntimeEnv;
+use std::slice;
 use std::sync::Arc;
 
 use datafusion_common::{config::SpillCompression, DataFusionError, Result};
@@ -30,8 +30,8 @@ use datafusion_execution::SendableRecordBatchStream;
 
 use super::{in_progress_spill_file::InProgressSpillFile, SpillReaderStream};
 use crate::coop::cooperative;
-use crate::{common::spawn_buffered, metrics::SpillMetrics};
 use crate::spill::in_memory_spill_buffer::InMemorySpillBuffer;
+use crate::{common::spawn_buffered, metrics::SpillMetrics};
 
 /// The `SpillManager` is responsible for the following tasks:
 /// - Reading and writing `RecordBatch`es to raw files based on the provided configurations.
@@ -177,7 +177,11 @@ impl SpillManager {
 
     /// Automatically decides whether to spill the given RecordBatch to memory or disk,
     /// depending on available memory pool capacity.
-    pub(crate) fn spill_batch_auto(&self, batch: &RecordBatch, request_msg: &str) -> Result<SpillLocation> {
+    pub(crate) fn spill_batch_auto(
+        &self,
+        batch: &RecordBatch,
+        request_msg: &str,
+    ) -> Result<SpillLocation> {
         let size = batch.get_sliced_size()?;
 
         // Check current memory usage and total limit from the runtime memory pool
@@ -188,14 +192,16 @@ impl SpillManager {
         };
 
         // If there's enough memory (with a safety margin), keep it in memory
-        if used + size * 3 / 2  <= limit {
+        if used + size * 3 / 2 <= limit {
             let buf = Arc::new(InMemorySpillBuffer::from_batch(batch)?);
             self.metrics.spilled_bytes.add(size);
             self.metrics.spilled_rows.add(batch.num_rows());
             Ok(SpillLocation::Memory(buf))
         } else {
             // Otherwise spill to disk using the existing SpillManager logic
-            let Some(file) = self.spill_record_batch_and_finish(slice::from_ref(batch), request_msg)? else {
+            let Some(file) =
+                self.spill_record_batch_and_finish(slice::from_ref(batch), request_msg)?
+            else {
                 return Err(DataFusionError::Execution(
                     "failed to spill batch to disk".into(),
                 ));
@@ -251,13 +257,14 @@ impl SpillManager {
         spill: &SpillLocation,
     ) -> Result<SendableRecordBatchStream> {
         match spill {
-            SpillLocation::Memory(buf) => Ok(Arc::clone(buf).as_stream(Arc::clone(&self.schema))?),
+            SpillLocation::Memory(buf) => {
+                Ok(Arc::clone(buf).as_stream(Arc::clone(&self.schema))?)
+            }
             SpillLocation::Disk(file) => self.read_spill_as_stream_ref(file),
         }
     }
 }
 
-
 #[derive(Debug, Clone)]
 pub enum SpillLocation {
     Memory(Arc<InMemorySpillBuffer>),