Native allocator follow-up: dynamic settings, query pool, plugin nodeStats

Builds on #21703 to add the pieces a real production deployment needs:

* Make pool min/max settings dynamic and grouped-validate cross-setting
  invariants on every cluster-state update (sum of pool mins <= root,
  per-pool min <= max). FLIGHT_MIN/INGEST_MIN defaults are 0L (the prior
  Long.MAX_VALUE defaults caused the new validator to reject any non-MAX
  root). setPoolMin now updates the live BufferAllocator.setLimit so the
  Dynamic property has observable effect even when the rebalancer is off.
* Derive ROOT_LIMIT_SETTING default from the AC node-native-memory budget
  (limit minus buffer-percent) so the framework respects the same budget
  AC throttles on, with a Long.MAX_VALUE fallback when AC is unconfigured.
* Add QUERY pool alongside FLIGHT and INGEST. Pools init at min when the
  rebalancer is enabled; otherwise at max so non-rebalanced nodes can
  still allocate.
* Wire arrow-flight-rpc to the FLIGHT pool, parquet-data-format to the
  INGEST pool, and analytics-engine to the QUERY pool via the framework's
  allocator service. Hard-fail if the framework plugin is missing —
  silently skipping the wire-up is the silent-misconfiguration class of
  bug we want to prevent.
* Cleanup ad-hoc allocator fallbacks in parquet-data-format / analytics-engine
  so all Arrow consumers go through the unified pool hierarchy.
* Rebalancer now distributes headroom across all pools (not only those with
  current allocation > 0). Avoids the dead-pool corner case where a pool
  with min=0 starts at limit=0, can never make a first allocation, and
  never receives a bonus.

DataFusion runtime memory accounting stays separate. The Rust-side
DataFusion MemoryPool is governed by datafusion.memory_pool_limit_bytes
(unchanged from before), which is the right knob: DataFusion's internal
sort/hash/group-by working memory is allocated by Rust, not through the
Arrow Java BufferAllocator hierarchy, so a Java-side pool would resize a
ceiling no allocator routes through. The framework's QUERY pool covers
the cross-plugin Arrow allocations the analytics-engine plumbing makes,
which is what we want bounded centrally.

Plugin _nodes/stats integration:
* Add Plugin#nodeStats() hook + PluginNodeStats interface in server.
* Wire NodeStats to carry Map<String, PluginNodeStats> with version-gated
  ser/deser at V_3_7_0 and top-level rendering under nodes.<id>.<name>.
* Each entry is wire-framed as (name, length-prefixed bytes); the receiver
  wraps the inner payload with NamedWriteableAwareStreamInput and drops
  entries whose subtype is not registered locally. This makes mixed-version
  rolling upgrades safe — a coordinator that lacks the plugin a data node
  is running keeps decoding the rest of NodeStats instead of failing the
  whole response.
* NativeAllocatorPluginStats adapter wraps NativeAllocatorPoolStats so the
  framework contributes to _nodes/stats. The dedicated _native_allocator/stats
  REST endpoint is gone — one observability surface, not two.
* Plugin stats are emitted on every _nodes/stats request regardless of the
  ?metric= filter; matches RemoteStoreNodeStats precedent.

DataFusion spill memory limit:
* Promote datafusion.spill_memory_limit_bytes to Setting.Property.Dynamic.
* Wire addSettingsUpdateConsumer that branches on
  NativeBridge.isSpillLimitDynamic(): mirrors live when df_set_spill_limit
  is exported, otherwise warns and waits for next node restart.

API hygiene:
* parquet.max_per_vsr_allocation_ratio is a divisor (limit/N), not a ratio.
  Renamed to parquet.max_per_vsr_allocation_divisor with a hard upper
  bound of 100 to reject fat-finger PUTs that would starve every VSR.

Real regressions caught during self-audit:
* Existing arrow-flight-rpc internalClusterTests and the sandbox coordinator
  ITs were not declaring the framework plugin in nodePlugins(). After the
  flight transport got wired to the FLIGHT pool, those ITs fail at node
  startup with "ArrowNativeAllocator not initialized". Each IT now installs
  the framework plugin and lists it as an extendedPlugin.

Signed-off-by: Gaurav Singh <snghsvn@amazon.com>

Author: gaurav-amz

libs/arrow-spi/src/main/java/org/opensearch/arrow/spi/NativeAllocator.java

@@ -55,6 +55,24 @@ public interface NativeAllocator extends Closeable {
      */
     NativeAllocatorPoolStats stats();
 
+    /**
+     * Registers a listener that is invoked after a pool's limit changes.
+     *
+     * <p>Listener invocation is synchronous on the caller thread. See
+     * {@link NativeAllocatorListener} for threading constraints.
+     *
+     * @param listener the listener to register
+     */
+    void addListener(NativeAllocatorListener listener);
+
+    /**
+     * Unregisters a previously registered listener. No-op if the listener
+     * was not registered.
+     *
+     * @param listener the listener to remove
+     */
+    void removeListener(NativeAllocatorListener listener);
+
     /**
      * Opaque handle to a memory pool. Plugins downcast to the concrete type
      * (e.g., Arrow's {@code BufferAllocator}) in the implementation layer.

libs/arrow-spi/src/main/java/org/opensearch/arrow/spi/NativeAllocatorListener.java

@@ -0,0 +1,39 @@
+/*
+ * 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.arrow.spi;
+
+/**
+ * Callback invoked when a pool's limit changes.
+ *
+ * <p>Consumers that mirror a Java-side pool limit to a separate accountant
+ * (for example, a native runtime that holds its own memory pool) implement
+ * this interface and register it via {@link NativeAllocator#addListener} so
+ * resize events propagate without polling.
+ *
+ * <p>The interface is allocator-agnostic — no Arrow types appear in the
+ * signature — so non-Arrow consumers can also subscribe.
+ *
+ * <p><b>Threading:</b> listeners are invoked synchronously on the thread that
+ * caused the resize (a settings-update thread or the rebalancer thread).
+ * Implementations must not block or take locks held by the caller; if either
+ * is required, dispatch the work to another executor.
+ *
+ * @opensearch.api
+ */
+@FunctionalInterface
+public interface NativeAllocatorListener {
+
+    /**
+     * Invoked after a pool's limit has been updated.
+     *
+     * @param poolName logical pool name
+     * @param newLimit the new limit in bytes
+     */
+    void onPoolLimitChanged(String poolName, long newLimit);
+}

libs/arrow-spi/src/main/java/org/opensearch/arrow/spi/NativeAllocatorPoolConfig.java

@@ -30,6 +30,8 @@ public final class NativeAllocatorPoolConfig {
     public static final String POOL_FLIGHT = "flight";
     /** Pool name for ingest pipeline memory. */
     public static final String POOL_INGEST = "ingest";
+    /** Pool name for query-execution memory (analytics-engine fragments and per-query allocators). */
+    public static final String POOL_QUERY = "query";
 
     /** Setting key for the root allocator limit. */
     public static final String SETTING_ROOT_LIMIT = "native.allocator.root.limit";
@@ -42,6 +44,10 @@ public final class NativeAllocatorPoolConfig {
     public static final String SETTING_INGEST_MIN = "native.allocator.pool.ingest.min";
     /** Setting key for the ingest pool maximum. */
     public static final String SETTING_INGEST_MAX = "native.allocator.pool.ingest.max";
+    /** Setting key for the query pool minimum. */
+    public static final String SETTING_QUERY_MIN = "native.allocator.pool.query.min";
+    /** Setting key for the query pool maximum. */
+    public static final String SETTING_QUERY_MAX = "native.allocator.pool.query.max";
 
     private NativeAllocatorPoolConfig() {}
 }

libs/arrow-spi/src/test/java/org/opensearch/arrow/spi/NativeAllocatorPoolConfigTests.java

@@ -15,13 +15,16 @@ public class NativeAllocatorPoolConfigTests extends OpenSearchTestCase {
     public void testPoolConstants() {
         assertEquals("flight", NativeAllocatorPoolConfig.POOL_FLIGHT);
         assertEquals("ingest", NativeAllocatorPoolConfig.POOL_INGEST);
+        assertEquals("query", NativeAllocatorPoolConfig.POOL_QUERY);
     }
 
     public void testSettingKeys() {
         assertEquals("native.allocator.pool.flight.min", NativeAllocatorPoolConfig.SETTING_FLIGHT_MIN);
         assertEquals("native.allocator.pool.flight.max", NativeAllocatorPoolConfig.SETTING_FLIGHT_MAX);
         assertEquals("native.allocator.pool.ingest.min", NativeAllocatorPoolConfig.SETTING_INGEST_MIN);
         assertEquals("native.allocator.pool.ingest.max", NativeAllocatorPoolConfig.SETTING_INGEST_MAX);
+        assertEquals("native.allocator.pool.query.min", NativeAllocatorPoolConfig.SETTING_QUERY_MIN);
+        assertEquals("native.allocator.pool.query.max", NativeAllocatorPoolConfig.SETTING_QUERY_MAX);
     }
 
     public void testRootSettingKey() {

plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/ArrowBasePlugin.java

@@ -15,14 +15,18 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.AbstractModule;
 import org.opensearch.common.inject.Module;
+import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
+import org.opensearch.core.common.unit.ByteSizeValue;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
+import org.opensearch.node.resource.tracker.ResourceTrackerSettings;
 import org.opensearch.plugins.ExtensiblePlugin;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.plugins.PluginNodeStats;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.script.ScriptService;
 import org.opensearch.threadpool.ThreadPool;
@@ -50,19 +54,49 @@ public class ArrowBasePlugin extends Plugin implements ExtensiblePlugin {
     /** Creates the plugin. */
     public ArrowBasePlugin() {}
 
-    /** Maximum bytes for the root Arrow allocator. */
-    public static final Setting<Long> ROOT_LIMIT_SETTING = Setting.longSetting(
+    /**
+     * Maximum bytes for the root Arrow allocator.
+     *
+     * <p>When unset, the default is derived from the admission-control budget
+     * {@link ResourceTrackerSettings#NODE_NATIVE_MEMORY_LIMIT_SETTING} reduced by
+     * {@link ResourceTrackerSettings#NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING} —
+     * the same budget AC throttles on. If AC is unconfigured (limit = 0), the
+     * default is {@link Long#MAX_VALUE}, preserving pre-AC behaviour.
+     */
+    public static final Setting<Long> ROOT_LIMIT_SETTING = new Setting<>(
         NativeAllocatorPoolConfig.SETTING_ROOT_LIMIT,
-        Long.MAX_VALUE,
-        0L,
+        ArrowBasePlugin::deriveRootLimitDefault,
+        s -> {
+            long v = Long.parseLong(s);
+            if (v < 0) {
+                throw new IllegalArgumentException(
+                    "Setting [" + NativeAllocatorPoolConfig.SETTING_ROOT_LIMIT + "] must be >= 0, got " + v
+                );
+            }
+            return v;
+        },
         Setting.Property.NodeScope,
         Setting.Property.Dynamic
     );
 
+    /**
+     * Computes the default for {@link #ROOT_LIMIT_SETTING} from the AC native-memory budget.
+     * Returns the bytes-as-string representation expected by the Setting parser.
+     */
+    static String deriveRootLimitDefault(Settings settings) {
+        ByteSizeValue acLimit = ResourceTrackerSettings.NODE_NATIVE_MEMORY_LIMIT_SETTING.get(settings);
+        if (acLimit.getBytes() <= 0) {
+            return Long.toString(Long.MAX_VALUE);
+        }
+        int bufferPercent = ResourceTrackerSettings.NODE_NATIVE_MEMORY_BUFFER_PERCENT_SETTING.get(settings);
+        long usable = acLimit.getBytes() - (acLimit.getBytes() * bufferPercent / 100L);
+        return Long.toString(Math.max(0L, usable));
+    }
+
     /** Minimum guaranteed bytes for the Flight pool. */
     public static final Setting<Long> FLIGHT_MIN_SETTING = Setting.longSetting(
         NativeAllocatorPoolConfig.SETTING_FLIGHT_MIN,
-        Long.MAX_VALUE,
+        0L,
         0L,
         Setting.Property.NodeScope,
         Setting.Property.Dynamic
@@ -80,7 +114,7 @@ public ArrowBasePlugin() {}
     /** Minimum guaranteed bytes for the ingest pool. */
     public static final Setting<Long> INGEST_MIN_SETTING = Setting.longSetting(
         NativeAllocatorPoolConfig.SETTING_INGEST_MIN,
-        Long.MAX_VALUE,
+        0L,
         0L,
         Setting.Property.NodeScope,
         Setting.Property.Dynamic
@@ -95,6 +129,38 @@ public ArrowBasePlugin() {}
         Setting.Property.Dynamic
     );
 
+    /**
+     * Minimum guaranteed bytes for the query pool. Honored by the rebalancer (when
+     * enabled) — sets a floor below which the rebalancer will not shrink the pool.
+     * Has no effect when rebalancing is disabled.
+     */
+    public static final Setting<Long> QUERY_MIN_SETTING = Setting.longSetting(
+        NativeAllocatorPoolConfig.SETTING_QUERY_MIN,
+        0L,
+        0L,
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
+
+    /**
+     * Maximum bytes the query pool can allocate. Enforced by Arrow's child-allocator
+     * limit — analytics-engine's per-query allocators are children of this pool, so the
+     * sum of in-flight per-query allocations is capped here.
+     *
+     * <p>Note: each individual analytics query is also bounded by
+     * {@code analytics.exec.QueryContext} per-query limit (currently the constant
+     * {@code DEFAULT_PER_QUERY_MEMORY_LIMIT = 256 MB}). Lowering {@code QUERY_MAX}
+     * below {@code 256 MB × concurrent-queries} can starve queries even when each
+     * individual query is within its per-query limit.
+     */
+    public static final Setting<Long> QUERY_MAX_SETTING = Setting.longSetting(
+        NativeAllocatorPoolConfig.SETTING_QUERY_MAX,
+        Long.MAX_VALUE,
+        0L,
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
+
     /** Interval in seconds between pool rebalance cycles. 0 disables rebalancing. */
     public static final Setting<Long> REBALANCE_INTERVAL_SETTING = Setting.longSetting(
         "native.allocator.rebalance.interval_seconds",
@@ -127,20 +193,27 @@ public Collection<Object> createComponents(
         allocator = new ArrowNativeAllocator(rootLimit);
         allocator.setRebalanceInterval(REBALANCE_INTERVAL_SETTING.get(settings));
 
-        long flightMin = FLIGHT_MIN_SETTING.get(settings);
-        long flightMax = FLIGHT_MAX_SETTING.get(settings);
-        long ingestMin = INGEST_MIN_SETTING.get(settings);
-        long ingestMax = INGEST_MAX_SETTING.get(settings);
+        // Single source of truth for cross-setting invariants — same logic runs on
+        // dynamic updates via the grouped consumer below.
+        validateUpdate(settings);
 
-        validateMinMax(NativeAllocatorPoolConfig.POOL_FLIGHT, flightMin, flightMax);
-        validateMinMax(NativeAllocatorPoolConfig.POOL_INGEST, ingestMin, ingestMax);
-        validateMinSum(rootLimit, flightMin, ingestMin);
+        allocator.getOrCreatePool(NativeAllocatorPoolConfig.POOL_FLIGHT, FLIGHT_MIN_SETTING.get(settings), FLIGHT_MAX_SETTING.get(settings));
+        allocator.getOrCreatePool(NativeAllocatorPoolConfig.POOL_INGEST, INGEST_MIN_SETTING.get(settings), INGEST_MAX_SETTING.get(settings));
+        allocator.getOrCreatePool(NativeAllocatorPoolConfig.POOL_QUERY, QUERY_MIN_SETTING.get(settings), QUERY_MAX_SETTING.get(settings));
 
-        allocator.getOrCreatePool(NativeAllocatorPoolConfig.POOL_FLIGHT, flightMin, flightMax);
-        allocator.getOrCreatePool(NativeAllocatorPoolConfig.POOL_INGEST, ingestMin, ingestMax);
+        ClusterSettings cs = clusterService.getClusterSettings();
+        cs.addSettingsUpdateConsumer(ROOT_LIMIT_SETTING, allocator::setRootLimit);
+        cs.addSettingsUpdateConsumer(REBALANCE_INTERVAL_SETTING, allocator::setRebalanceInterval);
+        cs.addSettingsUpdateConsumer(FLIGHT_MAX_SETTING, v -> allocator.setPoolLimit(NativeAllocatorPoolConfig.POOL_FLIGHT, v));
+        cs.addSettingsUpdateConsumer(FLIGHT_MIN_SETTING, v -> allocator.setPoolMin(NativeAllocatorPoolConfig.POOL_FLIGHT, v));
+        cs.addSettingsUpdateConsumer(INGEST_MAX_SETTING, v -> allocator.setPoolLimit(NativeAllocatorPoolConfig.POOL_INGEST, v));
+        cs.addSettingsUpdateConsumer(INGEST_MIN_SETTING, v -> allocator.setPoolMin(NativeAllocatorPoolConfig.POOL_INGEST, v));
+        cs.addSettingsUpdateConsumer(QUERY_MAX_SETTING, v -> allocator.setPoolLimit(NativeAllocatorPoolConfig.POOL_QUERY, v));
+        cs.addSettingsUpdateConsumer(QUERY_MIN_SETTING, v -> allocator.setPoolMin(NativeAllocatorPoolConfig.POOL_QUERY, v));
 
-        clusterService.getClusterSettings().addSettingsUpdateConsumer(ROOT_LIMIT_SETTING, allocator::setRootLimit);
-        clusterService.getClusterSettings().addSettingsUpdateConsumer(REBALANCE_INTERVAL_SETTING, allocator::setRebalanceInterval);
+        // Grouped validator runs across the related settings on every dynamic update so cross-setting
+        // invariants (sum of pool mins ≤ root, per-pool min ≤ max) are enforced post-startup.
+        cs.addSettingsUpdateConsumer(s -> {}, MIN_MAX_SETTINGS, ArrowBasePlugin::validateUpdate);
 
         // ArrowBasePlugin component: node-level Arrow allocator service for cross-plugin zero-copy
         this.allocatorService = new DefaultArrowAllocatorService(allocator);
@@ -161,6 +234,30 @@ protected void configure() {
         });
     }
 
+    private static final List<Setting<Long>> MIN_MAX_SETTINGS = List.of(
+        ROOT_LIMIT_SETTING,
+        FLIGHT_MIN_SETTING,
+        FLIGHT_MAX_SETTING,
+        INGEST_MIN_SETTING,
+        INGEST_MAX_SETTING,
+        QUERY_MIN_SETTING,
+        QUERY_MAX_SETTING
+    );
+
+    private static void validateUpdate(Settings settings) {
+        long rootLimit = ROOT_LIMIT_SETTING.get(settings);
+        long flightMin = FLIGHT_MIN_SETTING.get(settings);
+        long flightMax = FLIGHT_MAX_SETTING.get(settings);
+        long ingestMin = INGEST_MIN_SETTING.get(settings);
+        long ingestMax = INGEST_MAX_SETTING.get(settings);
+        long queryMin = QUERY_MIN_SETTING.get(settings);
+        long queryMax = QUERY_MAX_SETTING.get(settings);
+        validateMinMax(NativeAllocatorPoolConfig.POOL_FLIGHT, flightMin, flightMax);
+        validateMinMax(NativeAllocatorPoolConfig.POOL_INGEST, ingestMin, ingestMax);
+        validateMinMax(NativeAllocatorPoolConfig.POOL_QUERY, queryMin, queryMax);
+        validateMinSum(rootLimit, flightMin, ingestMin, queryMin);
+    }
+
     @Override
     public List<Setting<?>> getSettings() {
         return List.of(
@@ -169,10 +266,27 @@ public List<Setting<?>> getSettings() {
             FLIGHT_MAX_SETTING,
             INGEST_MIN_SETTING,
             INGEST_MAX_SETTING,
+            QUERY_MIN_SETTING,
+            QUERY_MAX_SETTING,
             REBALANCE_INTERVAL_SETTING
         );
     }
 
+    @Override
+    public List<PluginNodeStats> nodeStats() {
+        if (allocator == null) {
+            return List.of();
+        }
+        return List.of(new NativeAllocatorPluginStats(allocator.stats()));
+    }
+
+    @Override
+    public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
+        return List.of(
+            new NamedWriteableRegistry.Entry(PluginNodeStats.class, NativeAllocatorPluginStats.NAME, NativeAllocatorPluginStats::new)
+        );
+    }
+
     private static void validateMinMax(String poolName, long min, long max) {
         if (min > max) {
             throw new IllegalArgumentException("Pool '" + poolName + "' min (" + min + ") exceeds max (" + max + ")");
@@ -185,10 +299,10 @@ private static void validateMinSum(long rootLimit, long... mins) {
         }
         long sum = 0;
         for (long min : mins) {
-            long prev = sum;
-            sum += min;
-            if (sum < prev) {
-                throw new IllegalArgumentException("Sum of pool minimums overflows.");
+            try {
+                sum = Math.addExact(sum, min);
+            } catch (ArithmeticException overflow) {
+                throw new IllegalArgumentException("Sum of pool minimums overflows.", overflow);
             }
         }
         if (sum > rootLimit) {