diff --git a/datafusion/core/tests/memory_limit/memory_limit_validation/utils.rs b/datafusion/core/tests/memory_limit/memory_limit_validation/utils.rs index aae30418b25f9..6997cfefb348e 100644 --- a/datafusion/core/tests/memory_limit/memory_limit_validation/utils.rs +++ b/datafusion/core/tests/memory_limit/memory_limit_validation/utils.rs @@ -43,22 +43,27 @@ where F: FnOnce() -> Fut, Fut: Future, { - // Initialize system information - let mut system = System::new_all(); - system.refresh_all(); - // Get the current process ID let pid = sysinfo::get_current_pid().expect("Failed to get current PID"); - - // Shared atomic variable to store max RSS - let max_rss = Arc::new(AtomicUsize::new(0)); + let mut system = System::new(); + system.refresh_processes_specifics( + ProcessesToUpdate::Some(&[pid]), + true, + ProcessRefreshKind::nothing().with_memory().without_tasks(), + ); + let initial_rss = system + .process(pid) + .expect("Current process is missing") + .memory(); + // Capture a baseline before execution; monitor startup may outlast a short query. + let max_rss = Arc::new(AtomicUsize::new(initial_rss as usize)); // Clone for the monitoring task let max_rss_clone = Arc::clone(&max_rss); // Spawn a monitoring task let monitor_handle = SpawnedTask::spawn(async move { - let mut sys = System::new_all(); + let mut sys = system; let mut interval = interval(Duration::from_millis(7)); loop { @@ -66,7 +71,7 @@ where sys.refresh_processes_specifics( ProcessesToUpdate::Some(&[pid]), true, - ProcessRefreshKind::nothing().with_memory(), + ProcessRefreshKind::nothing().with_memory().without_tasks(), ); if let Some(process) = sys.process(pid) { let rss_bytes = process.memory(); diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs index efd5a61e4ed9f..d9c6feb4cfb08 100644 --- a/datafusion/optimizer/src/push_down_filter.rs +++ b/datafusion/optimizer/src/push_down_filter.rs @@ -1135,10 +1135,13 @@ impl OptimizerRule for PushDownFilter { // multiple window functions, each with potentially different partition keys. // Therefore, we need to ensure that any potential partition key returned is used in // ALL window functions. Otherwise, filters cannot be pushed by through that column. + #[allow(clippy::allow_attributes, clippy::mutable_key_type)] // Expr's memory reservations do not affect equality or hashing fn extract_partition_keys(func: &WindowFunction) -> HashSet<&Expr> { func.params.partition_by.iter().collect() } + #[allow(clippy::allow_attributes, clippy::mutable_key_type)] + // Expr's memory reservations do not affect equality or hashing let potential_partition_keys = window .window_expr .iter() @@ -1601,6 +1604,7 @@ fn with_filters(predicates: Vec, plan: LogicalPlan) -> LogicalPlan { /// /// A predicate containing a subquery is never pushed: what the subquery reads is /// not visible from the expression tree. +#[allow(clippy::allow_attributes, clippy::mutable_key_type)] // Expr's memory reservations do not affect equality or hashing fn reads_only_partition_keys( expr: &Expr, partition_keys: &HashSet<&Expr>,