test: sample current-process RSS before query execution

datafusion/core/tests/memory_limit/memory_limit_validation/utils.rs

@@ -43,30 +43,35 @@ where
     F: FnOnce() -> Fut,
     Fut: Future<Output = T>,
 {
-    // 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 {
             interval.tick().await;
             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();

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<Expr>, 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>,