Tighten memory guard to prevent OOM

[Bukhtawar]

---

DataFusion's hash aggregation allocates via jemalloc before consulting the memory pool (hashbrown::reserve() → malloc → then try_grow()). Under concurrent load, 20 queries burst past the 30.9GB pool limit simultaneously → OOM kills the node.

The Solution: Layered Defense

  Query arrives
        │
        ▼
  ┌─────────────────────────────────────────────────────────────────────┐
  │  ADMISSION (query_budget.rs) — once per query, before execution     │
  │                                                                     │
  │  cached_resident_bytes() vs pool_limit:                             │
  │    RSS < 70%  → admit at full parallelism                           │
  │    RSS 70-85% → admit at reduced parallelism (fewer partitions)     │
  │    RSS > 85%  → reject query (429 backpressure)                     │
  │                                                                     │
  │  Also reserves a "phantom" for untracked memory (in-flight batches, │
  │  decode buffers) so the pool accurately reflects true usage.         │
  └─────────────────────────────────────────────────────────────────────┘
        │
        ▼  query admitted, operators start processing batches
        │
  ┌─────────────────────────────────────────────────────────────────────┐
  │  try_grow() — called per batch per operator (hot path)              │
  │                                                                     │
  │  ┌─ HARD GUARD (pre-CAS) ────────────────────────────────────────┐  │
  │  │  cached_resident_bytes() > 95% of pool_limit?                 │  │
  │  │    YES → reject → triggers spill (operator flushes to disk)   │  │
  │  │    NO  → continue to CAS                                      │  │
  │  └───────────────────────────────────────────────────────────────┘  │
  │                                                                     │
  │  ┌─ POOL CAS ────────────────────────────────────────────────────┐  │
  │  │  pool_used + additional <= pool_limit?                        │  │
  │  │    YES → allow (fast path, no jemalloc call)                  │  │
  │  │    NO  → CAS fails, check override                           │  │
  │  └───────────────────────────────────────────────────────────────┘  │
  │                                                                     │
  │  ┌─ OPERATOR OVERRIDE (post-CAS-fail) ──────────────────────────┐  │
  │  │  cached_resident_bytes() < 85% of pool_limit?                 │  │
  │  │    YES → pool is lying (stale phantoms) → allow allocation    │  │
  │  │    NO  → pressure is real, continue to cancel check           │  │
  │  │                                                               │  │
  │  │  This is how spill sort buffers succeed: after the hash table │  │
  │  │  is freed during spill, RSS drops below 85%, override fires,  │  │
  │  │  and the sort allocation goes through.                        │  │
  │  └───────────────────────────────────────────────────────────────┘  │
  │                                                                     │
  │  ┌─ CANCEL (post-CAS-fail, post-override-denied) ───────────────┐  │
  │  │  cached_resident_bytes() > 95% of pool_limit?                 │  │
  │  │    YES → cancel query (ResourcesExhausted, protect node)      │  │
  │  │    NO  → reject → triggers spill                              │  │
  │  └───────────────────────────────────────────────────────────────┘  │
  └─────────────────────────────────────────────────────────────────────┘

  Thresholds (configurable at runtime via cluster settings)

  ┌─────────────────────────────────────────────┬─────────┬───────────────────────────────────────────────────────────┐
  │                   Setting                   │ Default │                           Role                            │
  ├─────────────────────────────────────────────┼─────────┼───────────────────────────────────────────────────────────┤
  │ datafusion.memory_guard.admission_threshold │ 0.70    │ Reduce/reject new queries                                 │
  ├─────────────────────────────────────────────┼─────────┼───────────────────────────────────────────────────────────┤
  │ datafusion.memory_guard.operator_threshold  │ 0.85    │ Override: allow if RSS below (spill buffers succeed here) │
  ├─────────────────────────────────────────────┼─────────┼───────────────────────────────────────────────────────────┤
  │ datafusion.memory_guard.critical_threshold  │ 0.95    │ Hard guard + cancel: protect node from OOM                │
  └─────────────────────────────────────────────┴─────────┴───────────────────────────────────────────────────────────┘

The Adaptive Cache

cached_resident_bytes() avoids calling jemalloc epoch.advance() (~1-5µs) on every try_grow:

• Happy path (RSS < 85%): returns cached value, refreshed every 100ms. Cost: one atomic load (<1ns).
• Under pressure (cached ≥ 85%): bypasses cache, reads fresh. This prevents a stale-high value from blocking the override when RSS has actually dropped (e.g., after spill frees memory).

How Spill Works End-to-End

1. Batch processes → hash table grows via hashbrown::reserve() (outside pool)
2. try_grow() called → hard guard or CAS rejects
3. DataFusion catches ResourcesExhausted → calls spill()
4. spill() emits hash table into a RecordBatch
5. spill() calls clear_shrink(0) → frees hash table memory → RSS drops
6. spill() calls try_grow(sort_memory) for sort buffer
7. CAS fails (pool accounting still high from phantoms)
8. Override: cached_resident (now fresh, RSS dropped) < 85% → ALLOWS ✓
9. Sort buffer allocated → data sorted → written to disk
10. Query continues with next batch (hash table rebuilt incrementally)

What's NOT fixed (separate issue)

18 high-cardinality queries (17M-100M groups) fail at the Arrow C Data import layer — when Rust exports partial aggregation results to Java, the Arrow flight allocator (~1.8GB) is exhausted. This is upstream of the memory pool, in the transport layer.
Needs backpressure at the Rust→Java boundary or a larger flight allocator budget.

Test Results

┌─────────────────────────────────────┬────────────────────────┬──────────────────────────────────────────────────┐
│                                     │       Unpatched        │            Patched (commit faa08e2c)             │
├─────────────────────────────────────┼────────────────────────┼──────────────────────────────────────────────────┤
│ OOM trigger (25 concurrent queries) │ OOM at 60.7GB in 8s    │ Survived at 46.5GB, all complete in 19s          │
├─────────────────────────────────────┼────────────────────────┼──────────────────────────────────────────────────┤
│ Full ClickBench (42 queries)        │ OOM kills node         │ 23/41 pass, 18 fail at Arrow import (node alive) │
├─────────────────────────────────────┼────────────────────────┼──────────────────────────────────────────────────┤
│ Peak native RSS                     │ 50.7 GB (uncontrolled) │ 27.9 GB (with spill)                             │
├─────────────────────────────────────┼────────────────────────┼──────────────────────────────────────────────────┤
│ Spill                               │ Never triggered        │ 1.4 GB peak                                      │
└─────────────────────────────────────┴────────────────────────┴──────────────────────────────────────────────────┘

Description

[Describe what this change achieves]

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 083578e)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue The cached_resident_bytes() function bypasses the cache when cached > threshold, but the threshold calculation uses spill_x1000 which may be zero if set_thresholds() was never called (static initialization is 850, but if Java calls setMemoryGuardThresholds with zero values before any query runs, EXECUTION_SPILL_X1000 could be zero). Division by zero is avoided, but threshold becomes zero, causing every call to bypass the cache and read fresh RSS (defeating the cache). This triggers on every try_grow call under load, adding ~1-5µs per batch per operator. let spill_x1000 = EXECUTION_SPILL_X1000.load(Ordering::Relaxed); let limit = pool_limit_for_guard(); if limit > 0 { let threshold = (limit as u64 * spill_x1000 / 1000) as i64; if cached >= threshold { let fresh = native_bridge_common::allocator::resident_bytes(); CACHED_RESIDENT.store(fresh, Ordering::Relaxed); return fresh; } } } Logic Error The proactive admission guard checks reserved >= admission_bytes before consulting jemalloc, but then compares resident >= spill_bytes (line 223) where spill_bytes uses admission_reject threshold (0.85), not admission_throttle (0.75). The variable name spill_bytes is misleading—it should be reject_bytes. More critically, the condition reserved >= admission_bytes (line 219) uses admission_throttle (0.75), but the subsequent RSS check at line 239 also uses admission_throttle. This means the RSS check at line 239 is redundant: if reserved >= admission_bytes (0.75 threshold) is true, and RSS is also checked against the same 0.75 threshold, the second check adds no value. The intent appears to be: check pool accounting first (cheap), then confirm with RSS (expensive), but both use the same threshold, making the flow confusing. if let Some(limit) = pool_limit(pool) { let reserved = pool.reserved(); let thresholds = crate::memory_guard::get_thresholds(); let admission_bytes = (limit as f64 * thresholds.admission_throttle) as usize; if reserved >= admission_bytes { let resident = crate::memory_guard::cached_resident_bytes(); if resident > 0 { let spill_bytes = (limit as f64 * thresholds.admission_reject) as i64; if resident >= spill_bytes { // RSS at spill threshold (85%) — reject immediately. // Even at min partitions this query will hit spill on first batch. // Better to reject with clear backpressure than admit and fail slowly. native_bridge_common::log_info!( "Admission REJECTED: pool reserved={}B, RSS={}B >= spill threshold ({:.0}% of {}B). Node under memory pressure.", reserved, resident, thresholds.admission_reject * 100.0, limit ); return Err(crate::native_error::admission_rejected_error( compute_untracked_bytes_with_columns(min_partitions, MIN_BATCH_SIZE, avg_row_bytes, num_columns), min_partitions, MIN_BATCH_SIZE, avg_row_bytes, )); } // RSS between admission (70%) and operator (85%) — reduce partitions let admission_threshold_bytes = (limit as f64 * thresholds.admission_throttle) as i64; if resident >= admission_threshold_bytes { native_bridge_common::log_info!( "Admission: pool reserved={}B, RSS={}B >= admission threshold ({:.0}%) — reducing to min partitions={}", reserved, resident, thresholds.admission_throttle * 100.0, min_partitions ); target_partitions = min_partitions; } } } } Possible Issue The hard guard (lines 139-149) rejects when resident_usize > critical_bytes, but the subsequent operator guard (lines 154-164) also rejects when resident_usize > spill_bytes. If execution_critical (0.95) and execution_spill (0.85) are both set to the same value (e.g., both 0.90 via dynamic settings), the critical guard fires first and the operator guard becomes unreachable. While the default values differ, the code does not enforce execution_critical > execution_spill, so a misconfiguration (or a race during settings update) could cause the operator guard to never fire, bypassing the intended spill-before-cancel logic. if resident > 0 && limit >= 16 * 1024 * 1024 { let thresholds = crate::memory_guard::get_thresholds(); let critical_bytes = (limit as f64 * thresholds.execution_critical) as usize; let spill_bytes = (limit as f64 * thresholds.execution_spill) as usize; let resident_usize = resident as usize; // Critical (95%): hard reject — OOM imminent, protect the node. if resident_usize > critical_bytes { self.tripped_count.fetch_add(1, Ordering::Relaxed); let used = self.used.load(Ordering::Relaxed); return Err(crate::native_error::pool_limit_error( additional, reservation.consumer().name(), reservation.size(), 0, limit, )); } // Operator (85%): soft reject — triggers spill. The operator will // flush to disk, free memory, then retry. Spill sort buffers will // succeed on retry because RSS drops and the override allows them. if resident_usize > spill_bytes { self.tripped_count.fetch_add(1, Ordering::Relaxed); let used = self.used.load(Ordering::Relaxed); return Err(crate::native_error::pool_limit_error( additional, reservation.consumer().name(), reservation.size(), 0, limit, )); } }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 083578e

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Prevent integer overflow in threshold comparison The comparison resident < threshold_bytes as i64 can overflow when threshold_bytes exceeds i64::MAX. Since resident is i64 and threshold_bytes is u64, casting a large u64 to i64 wraps to negative, causing incorrect comparisons. Cast resident to u64 before comparison to avoid overflow. sandbox/plugins/analytics-backend-datafusion/rust/src/memory_guard.rs [213-214] let threshold_bytes = (pool_limit_bytes as u64).saturating_mul(threshold_x1000) / 1000; -resident < threshold_bytes as i64 +(resident as u64) < threshold_bytes Suggestion importance[1-10]: 8 __ Why: Casting a large u64 to i64 can wrap to negative values, causing incorrect threshold comparisons. Since resident is i64 and threshold_bytes is u64, casting resident to u64 before comparison prevents overflow and ensures correct behavior.	Medium
	Fix overflow in critical threshold comparison The cast critical_bytes as i64 can overflow when critical_bytes exceeds i64::MAX, wrapping to negative and causing incorrect comparisons. Cast resident to u64 before comparison to prevent overflow and ensure correct threshold checks. sandbox/plugins/analytics-backend-datafusion/rust/src/memory_guard.rs [179-180] let critical_bytes = (pool_limit_bytes as u64).saturating_mul(EXECUTION_CRITICAL_X1000.load(Ordering::Acquire)) / 1000; -resident >= critical_bytes as i64 +(resident as u64) >= critical_bytes Suggestion importance[1-10]: 8 __ Why: The cast critical_bytes as i64 can overflow when critical_bytes exceeds i64::MAX, wrapping to negative and causing incorrect comparisons. Casting resident to u64 before comparison prevents this overflow issue.	Medium
	Prevent negative-to-unsigned cast wraparound Casting resident (i64) to usize can wrap negative values to large positive numbers on 64-bit systems, causing incorrect threshold comparisons. Check that resident > 0 before casting, or use resident.max(0) as usize to clamp negatives to zero. sandbox/plugins/analytics-backend-datafusion/rust/src/memory.rs [130-136] let critical_bytes = (limit as f64 * thresholds.execution_critical) as usize; let spill_bytes = (limit as f64 * thresholds.execution_spill) as usize; -let resident_usize = resident as usize; +let resident_usize = resident.max(0) as usize; Suggestion importance[1-10]: 7 __ Why: Casting resident (i64) to usize can wrap negative values to large positive numbers, causing incorrect threshold comparisons. Using resident.max(0) as usize prevents this wraparound and ensures correct behavior.	Medium
General	Rename misleading variable for clarity The variable name spill_bytes is misleading—it represents the admission reject threshold (85%), not the execution spill threshold. Rename to reject_bytes to match its semantic purpose and avoid confusion with the execution spill threshold used elsewhere. sandbox/plugins/analytics-backend-datafusion/rust/src/query_budget.rs [222-223] -let spill_bytes = (limit as f64 * thresholds.admission_reject) as i64; -if resident >= spill_bytes { +let reject_bytes = (limit as f64 * thresholds.admission_reject) as i64; +if resident >= reject_bytes { Suggestion importance[1-10]: 6 __ Why: The variable name spill_bytes is misleading as it represents the admission reject threshold (85%), not the execution spill threshold. Renaming to reject_bytes improves code clarity and prevents confusion.	Low

---

Previous suggestions

Suggestions up to commit bafa0f9

Category	Suggestion	Impact
General	Use cached RSS in admission logic The admission logic calls native_bridge_common::allocator::resident_bytes() directly instead of using cached_resident_bytes(). This bypasses the 100ms cache and forces an expensive jemalloc epoch advance on every query admission when the pool is under pressure. Use cached_resident_bytes() to maintain consistency with the rest of the guard logic and avoid unnecessary overhead. sandbox/plugins/analytics-backend-datafusion/rust/src/query_budget.rs [215-249] if reserved >= admission_bytes { - let resident = native_bridge_common::allocator::resident_bytes(); + let resident = crate::memory_guard::cached_resident_bytes(); if resident > 0 { let spill_bytes = (limit as f64 * thresholds.admission_reject) as i64; if resident >= spill_bytes { ... return Err(...); } let admission_threshold_bytes = (limit as f64 * thresholds.admission_throttle) as i64; if resident >= admission_threshold_bytes { ... target_partitions = min_partitions; } } } Suggestion importance[1-10]: 8 __ Why: The admission logic calls native_bridge_common::allocator::resident_bytes() directly instead of using cached_resident_bytes(), bypassing the 100ms cache and forcing expensive jemalloc epoch advances on every query admission under pressure. Using cached_resident_bytes() would maintain consistency with the rest of the guard logic and significantly reduce overhead.	Medium
	Pass actual pool usage to error The critical threshold rejection path loads used but never uses it in the error. This creates a misleading error message where the pool accounting shows 0 bytes used. Either pass used to the error constructor or remove the unused load to avoid confusion during debugging. sandbox/plugins/analytics-backend-datafusion/rust/src/memory.rs [139-149] if resident_usize > critical_bytes { self.tripped_count.fetch_add(1, Ordering::Relaxed); let used = self.used.load(Ordering::Relaxed); return Err(crate::native_error::pool_limit_error( additional, reservation.consumer().name(), reservation.size(), - 0, + used, limit, )); } Suggestion importance[1-10]: 6 __ Why: The used variable is loaded but not passed to the error constructor, resulting in 0 being passed instead. This creates misleading error messages during debugging. The suggestion correctly identifies this issue and proposes passing the actual used value to improve error reporting.	Low
Possible issue	Use stronger memory ordering for cache bypass The cache bypass logic uses Ordering::Relaxed for both load and store operations. When bypassing the cache due to high memory pressure, use Ordering::Acquire for the load and Ordering::Release for the store to ensure proper synchronization across threads. This prevents stale reads when multiple threads are checking memory pressure simultaneously. sandbox/plugins/analytics-backend-datafusion/rust/src/memory_guard.rs [43-54] if cached > 0 { - let spill_x1000 = EXECUTION_SPILL_X1000.load(Ordering::Relaxed); + let spill_x1000 = EXECUTION_SPILL_X1000.load(Ordering::Acquire); let limit = pool_limit_for_guard(); if limit > 0 { let threshold = (limit as u64 * spill_x1000 / 1000) as i64; if cached >= threshold { let fresh = native_bridge_common::allocator::resident_bytes(); - CACHED_RESIDENT.store(fresh, Ordering::Relaxed); + CACHED_RESIDENT.store(fresh, Ordering::Release); return fresh; } } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that Ordering::Relaxed may be insufficient when bypassing the cache under memory pressure. Using Ordering::Acquire for loads and Ordering::Release for stores ensures proper synchronization across threads, preventing potential race conditions where stale values could be read.	Medium

Suggestions up to commit e72d255

Category	Suggestion	Impact
Possible issue	Prevent integer overflow in threshold calculation The multiplication limit as u64 * spill_x1000 can overflow when limit is large (e.g., 1TB pool with threshold 850). Use saturating_mul to prevent silent wraparound that would cause incorrect threshold comparisons and bypass the cache refresh logic. sandbox/plugins/analytics-backend-datafusion/rust/src/memory_guard.rs [44-54] if cached > 0 { let spill_x1000 = EXECUTION_SPILL_X1000.load(Ordering::Relaxed); let limit = pool_limit_for_guard(); if limit > 0 { - let threshold = (limit as u64 * spill_x1000 / 1000) as i64; + let threshold = ((limit as u64).saturating_mul(spill_x1000) / 1000) as i64; if cached >= threshold { let fresh = native_bridge_common::allocator::resident_bytes(); CACHED_RESIDENT.store(fresh, Ordering::Relaxed); return fresh; } } } Suggestion importance[1-10]: 8 __ Why: Valid overflow concern. With a 1TB pool (limit = 1099511627776) and spill_x1000 = 850, the multiplication limit as u64 * spill_x1000 could overflow u64. Using saturating_mul prevents silent wraparound that would cause incorrect threshold comparisons and break the cache refresh logic.	Medium
	Prevent race condition in limit reads The hard guard checks are performed before the CAS operation, but the limit value is loaded with Acquire ordering later in the function. This creates a potential race where the limit could change between the guard checks and the actual CAS. Load limit once at the start of try_grow with Acquire ordering and reuse it throughout to ensure consistency. sandbox/plugins/analytics-backend-datafusion/rust/src/memory.rs [130-165] +let limit = self.dynamic_limit.load(Ordering::Acquire); let resident = crate::memory_guard::cached_resident_bytes(); if resident > 0 && limit >= 16 * 1024 * 1024 { let thresholds = crate::memory_guard::get_thresholds(); let critical_bytes = (limit as f64 * thresholds.execution_critical) as usize; let spill_bytes = (limit as f64 * thresholds.execution_spill) as usize; let resident_usize = resident as usize; // Critical (95%): hard reject — OOM imminent, protect the node. if resident_usize > critical_bytes { ... } // Operator (85%): soft reject — triggers spill. if resident_usize > spill_bytes { ... } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that limit is loaded twice (once at line 130 for the guard checks, then again at line 167 for the CAS). Loading it once with Acquire ordering ensures consistency across all checks. However, the existing code already loads limit at line 130 before the guard checks, so the issue is less severe than described.	Medium
General	Clarify threshold naming in admission logic The admission guard compares reserved (usize) against admission_bytes (usize), then compares resident (i64) against spill_bytes (i64). However, it uses admission_throttle threshold for the first check but admission_reject threshold for the second. This creates a logic gap where queries can slip through when RSS is between throttle and reject thresholds. Use consistent threshold types or verify the intended behavior. sandbox/plugins/analytics-backend-datafusion/rust/src/query_budget.rs [215-237] if let Some(limit) = pool_limit(pool) { let reserved = pool.reserved(); let thresholds = crate::memory_guard::get_thresholds(); - let admission_bytes = (limit as f64 * thresholds.admission_throttle) as usize; - if reserved >= admission_bytes { + let admission_throttle_bytes = (limit as f64 * thresholds.admission_throttle) as usize; + if reserved >= admission_throttle_bytes { let resident = native_bridge_common::allocator::resident_bytes(); if resident > 0 { - let spill_bytes = (limit as f64 * thresholds.admission_reject) as i64; - if resident >= spill_bytes { + let admission_reject_bytes = (limit as f64 * thresholds.admission_reject) as i64; + if resident >= admission_reject_bytes { ... } Suggestion importance[1-10]: 3 __ Why: The suggestion misunderstands the logic. The code intentionally uses admission_throttle (75%) to decide when to check RSS, then uses admission_reject (85%) to decide whether to reject. This is a two-tier check, not a logic gap. The variable naming admission_bytes could be clearer (e.g., admission_throttle_bytes), but the logic is correct.	Low

Suggestions up to commit b09174e

Category	Suggestion	Impact
General	Optimize cache bypass condition The cache bypass logic reads spill_x1000 and pool_limit_for_guard() on every call when cached > 0, even when the cached value is well below the threshold. This adds unnecessary atomic loads to the hot path. Cache the threshold value or check it only when the cached value is close to the threshold. sandbox/plugins/analytics-backend-datafusion/rust/src/memory_guard.rs [38-54] pub fn cached_resident_bytes() -> i64 { let cached = CACHED_RESIDENT.load(Ordering::Relaxed); // If last known value was above spill threshold, bypass cache and read fresh. if cached > 0 { - let spill_x1000 = EXECUTION_SPILL_X1000.load(Ordering::Relaxed); let limit = pool_limit_for_guard(); if limit > 0 { + let spill_x1000 = EXECUTION_SPILL_X1000.load(Ordering::Relaxed); let threshold = (limit as u64 * spill_x1000 / 1000) as i64; - if cached >= threshold { + // Only bypass cache if cached value is within 5% of threshold + if cached >= (threshold * 95 / 100) { let fresh = native_bridge_common::allocator::resident_bytes(); CACHED_RESIDENT.store(fresh, Ordering::Relaxed); return fresh; } } } + ... +} Suggestion importance[1-10]: 3 __ Why: The suggestion correctly identifies that the bypass logic reads atomics on every call when cached > 0. However, the proposed 5% buffer is arbitrary and may not provide meaningful optimization. The impact is minor since atomic loads are fast (~1ns).	Low

Suggestions up to commit 9c5ed7c

Category	Suggestion	Impact
Possible issue	Fix cache refresh race condition The cache refresh logic has a race condition. If the CAS fails, the function returns the stale cached value instead of the fresh value that another thread just stored. This can cause admission decisions to use outdated RSS data. After a failed CAS, reload the cached value to get the fresh update. sandbox/plugins/analytics-backend-datafusion/rust/src/memory_guard.rs [60-67] if now_ms.wrapping_sub(last) >= RESIDENT_CACHE_INTERVAL_MS { if LAST_CHECK_MS.compare_exchange(last, now_ms, Ordering::Relaxed, Ordering::Relaxed).is_ok() { let r = native_bridge_common::allocator::resident_bytes(); CACHED_RESIDENT.store(r, Ordering::Relaxed); return r; } + // CAS failed — another thread refreshed. Return the fresh value. + return CACHED_RESIDENT.load(Ordering::Relaxed); } CACHED_RESIDENT.load(Ordering::Relaxed) Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a race condition where a failed CAS returns a stale cached value instead of the fresh value another thread just stored. Returning the fresh value after a failed CAS ensures admission decisions use up-to-date RSS data, improving correctness under concurrent access.	Medium
General	Align RSS check with threshold The admission guard compares reserved (pool accounting) against admission_throttle to decide whether to check RSS, but then compares RSS against admission_reject (85%) for rejection. This creates a gap: when reserved is between 75-85%, RSS is checked but the rejection threshold is 85%, not 75%. Use admission_throttle for the RSS comparison to match the intent. sandbox/plugins/analytics-backend-datafusion/rust/src/query_budget.rs [215-224] if let Some(limit) = pool_limit(pool) { let reserved = pool.reserved(); let thresholds = crate::memory_guard::get_thresholds(); let admission_bytes = (limit as f64 * thresholds.admission_throttle) as usize; if reserved >= admission_bytes { let resident = native_bridge_common::allocator::resident_bytes(); if resident > 0 { - let spill_bytes = (limit as f64 * thresholds.admission_reject) as i64; - if resident >= spill_bytes { + let reject_bytes = (limit as f64 * thresholds.admission_reject) as i64; + let throttle_bytes = (limit as f64 * thresholds.admission_throttle) as i64; + if resident >= reject_bytes { Suggestion importance[1-10]: 5 __ Why: The suggestion identifies a potential inconsistency where reserved is compared against admission_throttle (75%) but RSS rejection uses admission_reject (85%). However, the code appears intentional: it checks RSS only when pool accounting shows pressure (75%), then applies a higher threshold (85%) for actual rejection. The suggestion may improve clarity but the current logic seems deliberate for a two-tier approach.	Low

Suggestions up to commit b741411

Category	Suggestion	Impact
Possible issue	Prevent overflow in threshold calculations Use saturating_mul for threshold calculations to prevent potential overflow when limit is large. This matches the pattern used elsewhere in the codebase (e.g., memory_guard.rs) and ensures safe arithmetic. sandbox/plugins/analytics-backend-datafusion/rust/src/memory.rs [133-135] let thresholds = crate::memory_guard::get_thresholds(); -let critical_bytes = (limit as f64 * thresholds.execution_critical) as usize; -let spill_bytes = (limit as f64 * thresholds.execution_spill) as usize; +let critical_bytes = ((limit as u64).saturating_mul((thresholds.execution_critical * 1000.0) as u64) / 1000) as usize; +let spill_bytes = ((limit as u64).saturating_mul((thresholds.execution_spill * 1000.0) as u64) / 1000) as usize; Suggestion importance[1-10]: 4 __ Why: The suggestion to use saturating_mul is valid for preventing overflow, but the impact is low since limit is typically a reasonable memory size (not near u64::MAX). The suggested code is more complex than necessary.	Low
	Prevent overflow in admission threshold calculation Use saturating_mul for the spill_bytes calculation to prevent overflow when limit is large. This ensures consistent safe arithmetic across all threshold calculations. sandbox/plugins/analytics-backend-datafusion/rust/src/query_budget.rs [219-222] if reserved >= admission_bytes { let resident = native_bridge_common::allocator::resident_bytes(); if resident > 0 { - let spill_bytes = (limit as f64 * thresholds.admission_reject) as i64; + let spill_bytes = ((limit as u64).saturating_mul((thresholds.admission_reject * 1000.0) as u64) / 1000) as i64; Suggestion importance[1-10]: 4 __ Why: Similar to suggestion 1, using saturating_mul prevents overflow but has low practical impact. The limit value is typically a reasonable memory size, making overflow unlikely in production scenarios.	Low

[github-actions]

Persistent review updated to latest commit 4703262

[github-actions]

Persistent review updated to latest commit ed68a0b

[github-actions]

✅ Gradle check result for ed68a0b: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.39%. Comparing base (71c0afb) to head (083578e).

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #21814      +/-   ##
============================================
- Coverage     73.40%   73.39%   -0.02%     
- Complexity    75366    75385      +19     
============================================
  Files          6029     6029              
  Lines        342164   342164              
  Branches      49204    49204              
============================================
- Hits         251178   251117      -61     
- Misses        71051    71099      +48     
- Partials      19935    19948      +13     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[github-actions]

Persistent review updated to latest commit 11be13c

[github-actions]

Persistent review updated to latest commit 2e32d1c

[github-actions]

✅ Gradle check result for 2e32d1c: SUCCESS

[github-actions]

Persistent review updated to latest commit 3706521

[github-actions]

❌ Gradle check result for 3706521: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit 525b207

[github-actions]

Persistent review updated to latest commit 5e1ee57

[github-actions]

✅ Gradle check result for 5e1ee57: SUCCESS

[github-actions]

Persistent review updated to latest commit c7ca448

[github-actions]

Persistent review updated to latest commit a32359f

[github-actions]

✅ Gradle check result for a32359f: SUCCESS

[github-actions]

Persistent review updated to latest commit 922c4ab

[github-actions]

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit 2980913.

Path	Line	Severity	Description
sandbox/Cross.toml	2	high	Container image 'cross-aarch64-with-protoc' is referenced without a registry prefix or digest pin. Without a fully qualified registry path and SHA digest, this image resolves via Docker's default search order (Docker Hub or local daemon), making it susceptible to namespace hijacking or image substitution during cross-compilation builds. Maintainers should verify this image's source, pin it to a specific digest (e.g., registry.example.com/cross-aarch64-with-protoc@sha256:...), and confirm it is from a trusted internal registry.
sandbox/libs/dataformat-native/rust/Cross.toml	2	high	Same unverifiable container image 'cross-aarch64-with-protoc' introduced in a second location without registry or digest pin. Identical supply chain risk as sandbox/Cross.toml — any compromise of this build image would affect the dataformat-native Rust library cross-compilation, potentially injecting malicious code into the native shared library artifact.

The table above displays the top 10 most important findings.

Total: 2 | Critical: 0 | High: 2 | Medium: 0 | Low: 0

---

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

---

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

[github-actions]

Persistent review updated to latest commit 509e6e5

[github-actions]

Persistent review updated to latest commit 2980913

[github-actions]

Persistent review updated to latest commit b741411

[github-actions]

Persistent review updated to latest commit 9c5ed7c

[github-actions]

Persistent review updated to latest commit b09174e

[github-actions]

Persistent review updated to latest commit e72d255

[github-actions]

Persistent review updated to latest commit bafa0f9

[github-actions]

✅ Gradle check result for bafa0f9: SUCCESS

[github-actions]

Persistent review updated to latest commit 083578e

[github-actions]

✅ Gradle check result for 083578e: SUCCESS