Wire Arrow consumers to unified native allocator

libs/arrow-spi/build.gradle

@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+
+apply plugin: 'opensearch.publish'
+
+dependencies {
+    api project(':libs:opensearch-core')
+    api project(':libs:opensearch-common')
+    testImplementation project(':test:framework')
+}
+
+tasks.named('forbiddenApisMain').configure {
+    replaceSignatureFiles 'jdk-signatures'
+}
+
+tasks.named('forbiddenApisTest').configure {
+    replaceSignatureFiles 'jdk-signatures'
+}

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

@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import java.io.Closeable;
+
+/**
+ * Arrow-agnostic interface for a hierarchical native memory allocator.
+ *
+ * <p>The implementation (backed by Arrow's {@code RootAllocator}) is provided by
+ * a plugin. The SPI allows other subsystems to interact with the allocator
+ * without depending on Arrow classes.
+ *
+ * <p>Plugins that need Arrow allocators obtain the implementation via
+ * service lookup or plugin extension and call {@link #getOrCreatePool} to
+ * register their pool.
+ *
+ * @opensearch.api
+ */
+public interface NativeAllocator extends Closeable {
+
+    /**
+     * Returns the named pool, creating it on first access with the given limit.
+     * Subsequent calls with the same name return the same pool (first-call limit wins).
+     *
+     * @param poolName logical pool name (e.g., "query", "flight")
+     * @param limit maximum bytes this pool can allocate in aggregate
+     * @return an opaque pool handle
+     */
+    PoolHandle getOrCreatePool(String poolName, long limit);
+
+    /**
+     * Updates the limit of an existing pool.
+     *
+     * @param poolName logical pool name
+     * @param newLimit new maximum bytes for the pool
+     */
+    void setPoolLimit(String poolName, long newLimit);
+
+    /**
+     * Sets the root-level memory limit for the entire allocator.
+     *
+     * @param limit new maximum bytes for the root allocator
+     */
+    void setRootLimit(long limit);
+
+    /**
+     * Collects a point-in-time stats snapshot across all pools.
+     */
+    NativeAllocatorPoolStats stats();
+
+    /**
+     * Registers a listener that fires whenever a pool's limit is updated via
+     * {@link #setPoolLimit(String, long)}. Used by consumers (e.g., DataFusion)
+     * that need to mirror the Java-side limit to a separate native accountant.
+     *
+     * @param listener callback to invoke on pool resize
+     */
+    default void addListener(NativeAllocatorListener listener) {
+        // Default no-op for impls that don't support listeners.
+    }
+
+    /**
+     * Opaque handle to a memory pool. Plugins downcast to the concrete type
+     * (e.g., Arrow's {@code BufferAllocator}) in the implementation layer.
+     */
+    interface PoolHandle {
+
+        /**
+         * Creates a named child allocation within this pool.
+         *
+         * @param childName name for debugging
+         * @param childLimit maximum bytes for the child
+         * @return an opaque child handle (downcast to BufferAllocator in Arrow impl)
+         */
+        PoolHandle newChild(String childName, long childLimit);
+
+        /**
+         * Returns the current allocated bytes for this pool/child.
+         */
+        long allocatedBytes();
+
+        /**
+         * Returns the peak memory allocation.
+         */
+        long peakBytes();
+
+        /**
+         * Returns the configured limit.
+         */
+        long limit();
+
+        /**
+         * Releases this allocation handle.
+         */
+        void close();
+    }
+}

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

@@ -0,0 +1,29 @@
+/*
+ * 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. Consumers that mirror the
+ * Java-side pool limit to a separate accountant (e.g., a native runtime that
+ * holds its own memory pool) implement this and register it with the
+ * allocator so resize events propagate.
+ *
+ * @opensearch.api
+ */
+@FunctionalInterface
+public interface NativeAllocatorListener {
+
+    /**
+     * Invoked after a pool's limit has been updated.
+     *
+     * @param poolName logical pool name
+     * @param newLimit new limit in bytes
+     */
+    void onPoolLimitChanged(String poolName, long newLimit);
+}

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

@@ -0,0 +1,59 @@
+/*
+ * 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;
+
+/**
+ * Pool name constants and setting keys for native allocator pools.
+ *
+ * <p>Each pool has a min (guaranteed reservation) and max (burst limit).
+ * The rebalancer ensures every pool can always allocate up to its min,
+ * and distributes unused capacity up to each pool's max.
+ *
+ * <p>Limits are provided via {@code opensearch.yml} or the cluster settings API.
+ * The setting keys follow the pattern
+ * {@code native.allocator.pool.<name>.min} and {@code native.allocator.pool.<name>.max}.
+ *
+ * <p>This class is Arrow-agnostic — it defines the logical pool topology
+ * without referencing any Arrow classes.
+ *
+ * @opensearch.api
+ */
+public final class NativeAllocatorPoolConfig {
+
+    /** Pool name for Arrow Flight RPC memory. */
+    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";
+    /** Pool name for DataFusion native runtime memory. */
+    public static final String POOL_DATAFUSION = "datafusion";
+
+    /** Setting key for the root allocator limit. */
+    public static final String SETTING_ROOT_LIMIT = "native.allocator.root.limit";
+
+    /** Setting key for the Flight pool minimum. */
+    public static final String SETTING_FLIGHT_MIN = "native.allocator.pool.flight.min";
+    /** Setting key for the Flight pool maximum. */
+    public static final String SETTING_FLIGHT_MAX = "native.allocator.pool.flight.max";
+    /** Setting key for the ingest pool minimum. */
+    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";
+    /** Setting key for the DataFusion pool minimum. */
+    public static final String SETTING_DATAFUSION_MIN = "native.allocator.pool.datafusion.min";
+    /** Setting key for the DataFusion pool maximum. */
+    public static final String SETTING_DATAFUSION_MAX = "native.allocator.pool.datafusion.max";
+
+    private NativeAllocatorPoolConfig() {}
+}