chore: fix some flood logs with batch progress (#15492)

Author: BohuTANG

src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_spill_reader.rs

@@ -15,6 +15,7 @@
 use std::any::Any;
 use std::collections::VecDeque;
 use std::sync::Arc;
+use std::time::Duration;
 use std::time::Instant;
 
 use databend_common_base::runtime::profile::Profile;
@@ -199,6 +200,11 @@ impl<Method: HashMethodBounds, V: Send + Sync + 'static> Processor
                     self.deserializing_meta = Some((block_meta, VecDeque::from(vec![data])));
                 }
                 AggregateMeta::Partitioned { data, .. } => {
+                    // For log progress.
+                    let mut total_elapsed = Duration::default();
+                    let log_interval = 100;
+                    let mut processed_count = 0;
+
                     let mut read_data = Vec::with_capacity(data.len());
                     for meta in data {
                         if let AggregateMeta::BucketSpilled(payload) = meta {
@@ -229,11 +235,18 @@ impl<Method: HashMethodBounds, V: Send + Sync + 'static> Processor
                                     );
                                 }
 
-                                info!(
-                                    "Read aggregate spill {} successfully, elapsed: {:?}",
-                                    location,
-                                    instant.elapsed()
-                                );
+                                total_elapsed += instant.elapsed();
+                                processed_count += 1;
+
+                                // log the progress
+                                if processed_count % log_interval == 0 {
+                                    info!(
+                                        "Read aggregate {}/{} spilled buckets, elapsed: {:?}",
+                                        processed_count,
+                                        data.len(),
+                                        total_elapsed
+                                    );
+                                }
 
                                 Ok(data)
                             }));
@@ -251,6 +264,11 @@ impl<Method: HashMethodBounds, V: Send + Sync + 'static> Processor
                             self.deserializing_meta = Some((block_meta, read_data?));
                         }
                     };
+
+                    info!(
+                        "Read {} aggregate spills successfully, total elapsed: {:?}",
+                        processed_count, total_elapsed
+                    );
                 }
             }
         }

src/query/sql/src/evaluator/cse.rs

@@ -15,7 +15,7 @@
 use std::collections::HashMap;
 
 use databend_common_expression::Expr;
-use log::info;
+use log::debug;
 
 use super::BlockOperator;
 use crate::optimizer::ColumnSet;
@@ -63,7 +63,7 @@ pub fn apply_cse(
                         let mut expr_cloned = cse_candidate.clone();
                         perform_cse_replacement(&mut expr_cloned, &cse_replacements);
 
-                        info!(
+                        debug!(
                             "cse_candidate: {}, temp_expr: {}",
                             expr_cloned.sql_display(),
                             temp_expr.sql_display()