Fix native memory duress probe

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsSearchBackendPlugin.java

@@ -8,7 +8,9 @@
 
 package org.opensearch.analytics.spi;
 
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 /**
  * SPI extension point for backend query engine plugins.
@@ -114,6 +116,24 @@ default void configureFilterDelegation(FilterDelegationHandle handle, BackendExe
         throw new UnsupportedOperationException("configureFilterDelegation not implemented for [" + name() + "]");
     }
 
+    /**
+     * Returns a snapshot of this backend's currently-tracked queries, keyed by {@code contextId}.
+     *
+     * <p>The map is a point-in-time view — entries can register or drain concurrently on the
+     * backend side. Implementations MUST return a non-null map (empty when nothing is tracked)
+     * and SHOULD make it unmodifiable so callers cannot mutate backend state.
+     *
+     * <p>Implementations MAY cap the result to a top-N subset by current memory usage to bound
+     * the FFI cost (the DataFusion backend caps at the heaviest 10 live queries). Callers that
+     * need a complete enumeration should not rely on this method.
+     *
+     * <p>Default implementation returns an empty map so backends that do not track per-query
+     * metrics don't have to opt in.
+     */
+    default Map<Long, QueryExecutionMetrics> getTopQueriesByMemory() {
+        return Collections.emptyMap();
+    }
+
     /**
      * Install a thread tracker for attribution of delegation callbacks executing on foreign threads.
      * Called after {@link #configureFilterDelegation}. Pass {@code null} to clear.

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/QueryExecutionMetrics.java

@@ -0,0 +1,21 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.analytics.spi;
+
+/**
+ * Snapshot of per-query execution metrics reported by an analytics backend.
+ *
+ * <p>The context id (the map key in {@link AnalyticsSearchBackendPlugin#getTopQueriesByMemory()})
+ * is not repeated here — this record holds only the memory accounting fields.
+ *
+ * @param currentBytes bytes currently reserved by the query's memory pool
+ *
+ * @opensearch.internal
+ */
+public record QueryExecutionMetrics(long currentBytes) {}

sandbox/plugins/analytics-backend-datafusion/rust/src/ffm.rs

@@ -206,6 +206,47 @@ pub extern "C" fn df_cancel_query(context_id: i64) {
     api::cancel_query(context_id);
 }
 
+// ---------------------------------------------------------------------------
+// Per-query registry top-N snapshot
+//
+// One FFM call: Java allocates a buffer sized for `N` entries, Rust selects
+// the heaviest live queries by `current_bytes` (bounded min-heap of size N)
+// and writes them back-to-back. See `query_tracker::WireQueryMetric` for the
+// wire layout.
+// ---------------------------------------------------------------------------
+
+/// Copies up to `cap_entries` of the heaviest live queries (by
+/// `current_bytes` desc) as `WireQueryMetric`s into the caller-provided buffer.
+/// Returns the number of entries actually written.
+///
+/// Order of entries within the buffer is unspecified. Completed and zero-byte
+/// trackers are filtered out.
+///
+/// Safety: `out_ptr` must be non-null, 8-byte aligned, and point to storage
+/// for at least `cap_entries * size_of::<WireQueryMetric>()` bytes.
+#[ffm_safe]
+#[no_mangle]
+pub unsafe extern "C" fn df_query_registry_top_n_by_current(
+    out_ptr: *mut u8,
+    cap_entries: i64,
+) -> i64 {
+    use crate::query_tracker::{snapshot_top_n_by_current, WireQueryMetric};
+
+    if cap_entries < 0 {
+        return Err(format!("negative capacity: {cap_entries}"));
+    }
+    if cap_entries == 0 {
+        return Ok(0);
+    }
+    if out_ptr.is_null() {
+        return Err("null snapshot buffer".to_string());
+    }
+    let out: &mut [WireQueryMetric] =
+        slice::from_raw_parts_mut(out_ptr as *mut WireQueryMetric, cap_entries as usize);
+    let written = snapshot_top_n_by_current(out);
+    Ok(written as i64)
+}
+
 #[ffm_safe]
 #[no_mangle]
 pub unsafe extern "C" fn df_sql_to_substrait(

sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs

@@ -35,4 +35,4 @@ pub mod session_context;
 pub mod statistics_cache;
 pub mod udf;
 pub mod stats;
-pub mod task_monitors;
+pub mod task_monitors;