replace err with warn log

Author: ding-young

datafusion/core/tests/sql/runtime_config.rs

@@ -193,7 +193,7 @@ async fn test_max_temp_directory_size_enforcement() {
         .unwrap();
 
     let result = ctx.sql(query).await.unwrap().collect().await;
-    println!("result is {result:?}");
+
     assert!(
         result.is_ok(),
         "Should not fail due to max temp directory size limit"

datafusion/execution/src/memory_pool/pool.rs

@@ -44,7 +44,6 @@ impl MemoryPool for UnboundedMemoryPool {
 
     fn try_grow(&self, reservation: &MemoryReservation, additional: usize) -> Result<()> {
         self.grow(reservation, additional);
-        println!("[mem pool] {} used", self.reserved());
         Ok(())
     }
 
@@ -71,8 +70,7 @@ pub struct GreedyMemoryPool {
 impl GreedyMemoryPool {
     /// Create a new pool that can allocate up to `pool_size` bytes
     pub fn new(pool_size: usize) -> Self {
-        // debug!("Created new GreedyMemoryPool(pool_size={pool_size})");
-        println!("Created new GreedyMemoryPool(pool_size={pool_size})");
+        debug!("Created new GreedyMemoryPool(pool_size={pool_size})");
         Self {
             pool_size,
             used: AtomicUsize::new(0),

datafusion/physical-plan/src/sorts/streaming_merge.rs

@@ -29,9 +29,8 @@ use arrow::array::*;
 use arrow::datatypes::{DataType, SchemaRef};
 use datafusion_common::{internal_err, Result};
 use datafusion_execution::disk_manager::RefCountedTempFile;
-#[allow(unused_imports)]
 use datafusion_execution::memory_pool::{
-    human_readable_size, GreedyMemoryPool, MemoryConsumer, MemoryPool, MemoryReservation,
+    human_readable_size, MemoryConsumer, MemoryPool, MemoryReservation,
     UnboundedMemoryPool,
 };
 use datafusion_physical_expr_common::sort_expr::LexOrdering;

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

@@ -43,6 +43,7 @@ use datafusion_common_runtime::SpawnedTask;
 use datafusion_execution::disk_manager::RefCountedTempFile;
 use datafusion_execution::RecordBatchStream;
 use futures::{FutureExt as _, Stream};
+use log::warn;
 
 /// Stream that reads spill files from disk where each batch is read in a spawned blocking task
 /// It will read one batch at a time and will not do any buffering, to buffer data use [`crate::common::spawn_buffered`]
@@ -54,12 +55,15 @@ use futures::{FutureExt as _, Stream};
 struct SpillReaderStream {
     schema: SchemaRef,
     state: SpillReaderStreamState,
-    /// how much memory the largest memory batch is taking
-    pub max_record_batch_memory: Option<usize>,
+    /// Maximum memory size observed among spilling sorted record batches.
+    /// This is used for validation purposes during reading each RecordBatch from spill.
+    /// For context on why this value is recorded and validated,
+    /// see `physical_plan/sort/multi_level_merge.rs`.
+    max_record_batch_memory: Option<usize>,
 }
 
 // Small margin allowed to accommodate slight memory accounting variation
-const MEMORY_MARGIN: usize = 4096;
+const SPILL_BATCH_MEMORY_MARGIN: usize = 4096;
 
 /// When we poll for the next batch, we will get back both the batch and the reader,
 /// so we can call `next` again.
@@ -141,19 +145,15 @@ impl SpillReaderStream {
                                     let actual_size =
                                         get_record_batch_memory_size(&batch);
                                     if actual_size
-                                        > max_record_batch_memory + MEMORY_MARGIN
+                                        > max_record_batch_memory
+                                            + SPILL_BATCH_MEMORY_MARGIN
                                     {
-                                        return Poll::Ready(Some(Err(
-                                            DataFusionError::ResourcesExhausted(
-                                                format!(
-                                                    "Record batch memory usage ({actual_size} bytes) exceeds the expected limit ({max_record_batch_memory} bytes)\n
-                                                    by more than the allowed tolerance ({MEMORY_MARGIN} bytes).\n
-                                                    This likely indicates a bug in memory accounting during spilling.\n
-                                                    Please report this issue",
-                                                )
-                                                .to_owned(),
-                                            ),
-                                        )));
+                                        warn!(
+                                                "Record batch memory usage ({actual_size} bytes) exceeds the expected limit ({max_record_batch_memory} bytes) \n\
+                                                by more than the allowed tolerance ({SPILL_BATCH_MEMORY_MARGIN} bytes).\n\
+                                                This likely indicates a bug in memory accounting during spilling.\n\
+                                                Please report this issue in https://github.com/apache/datafusion/issues/17340."
+                                            );
                                     }
                                 }
                                 self.state = SpillReaderStreamState::Waiting(reader);