[analytics-engine] Fix query-pool import leak: register POOL_QUERY as unmanaged/unbounded

[sandeshkr419]

The arrow-java C-Data importer retains a native batch reference before calling allocateBytes. If POOL_QUERY is bounded and throws OutOfMemoryException mid-array, the refcount stays elevated, the C-Data release callback never fires, and the native batch leaks permanently. These bytes are zero-copy foreign wraps of pre-existing native memory — the Java-side limit guarded nothing physical; real enforcement lives Rust-side (DataFusion MemoryPool → CircuitBreakingException → 429).

POOL_QUERY is now registered as a special unmanaged pool: fixed at Long.MAX_VALUE, excluded from the rebalancer, budget validation, and pool-group limit sums. The pool.query.max and pool.query.min settings are deprecated — they have no effect and log a warning if configured.

This also reverts the per-batch staging workaround from #22448. That workaround prevented the trigger by routing imports through an unbounded root child; with POOL_QUERY permanently unbounded the trigger cannot fire, making the staging path unnecessary. The definitive upstream fix for the arrow-java bug is bug: apache/arrow-java#1239 pr: apache/arrow-java#1240.

Tested on a single-node cluster and 2 node cluster with 3 shard setup:

• pool.query.max=239MB setting accepted but pool stays MAX (no-op confirmed)
• deprecation warn fires on dynamic update
• 16×5 wide chained-concat storm: allocated_bytes drains to baseline, 0 wrapForeignAllocation
• rebalancer runs multiple ticks: query pool untouched

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 🔍

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 Multiple PR themes Sub-PR theme: Register POOL_QUERY as unmanaged/unbounded in allocator framework Relevant files: plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/ArrowBasePlugin.java plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/ArrowNativeAllocator.java plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/NativeMemoryRebalancer.java plugins/arrow-base/src/test/java/org/opensearch/arrow/allocator/ArrowBasePluginTests.java plugins/arrow-base/src/test/java/org/opensearch/arrow/allocator/ArrowNativeAllocatorTests.java plugins/arrow-base/src/test/java/org/opensearch/arrow/allocator/NativeMemoryRebalancerTests.java plugins/arrow-flight-rpc/src/internalClusterTest/java/org/opensearch/arrow/flight/NativeMemoryRebalancerIT.java Sub-PR theme: Revert per-batch staging import workaround in analytics backends Relevant files: sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionResultStream.java sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/DatafusionResultStreamTests.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneResultStream.java

⚡ Recommended focus areas for review Missing setting registration QUERY_MIN_SETTING and QUERY_MAX_SETTING were changed to add Setting.Property.Dynamic and Setting.Property.Deprecated, but the diff removed the trailing Setting.Property.Dynamic line from the original definition and now ends with a trailing comma before ). Verify the settings compile and are still registered in getSettings() — a syntactically valid but incorrectly-declared setting could cause the deprecation warning wiring (addSettingsUpdateConsumer) to fail at startup if the setting isn't registered as dynamic/updatable, since these consumers require the setting to be registered with the ClusterSettings. public static final Setting<Long> QUERY_MIN_SETTING = new Setting<>( NativeAllocatorPoolConfig.SETTING_QUERY_MIN, s -> derivePoolMinDefault(s, 2), s -> parseNonNegativeLong(s, NativeAllocatorPoolConfig.SETTING_QUERY_MIN), Setting.Property.NodeScope, Setting.Property.Dynamic, Setting.Property.Deprecated ); /** @deprecated The query pool is unbounded (registered as a special unmanaged pool); this has no effect. */ @Deprecated public static final Setting<Long> QUERY_MAX_SETTING = new Setting<>( NativeAllocatorPoolConfig.SETTING_QUERY_MAX, s -> derivePoolMaxDefault(s, 5), s -> parseNonNegativeLong(s, NativeAllocatorPoolConfig.SETTING_QUERY_MAX), Setting.Property.NodeScope, Setting.Property.Dynamic, Setting.Property.Deprecated ); Race in registerUnmanagedPool registerUnmanagedPool checks pools.containsKey(poolName) and then calls pools.computeIfAbsent, but unmanagedPools.add(poolName) occurs between them. If a concurrent caller invokes getOrCreatePool with the same name after the containsKey check but before computeIfAbsent, the pool ends up flagged as unmanaged while its underlying child allocator is bounded — the exact state the guard is meant to prevent. Consider making the check-and-register atomic (e.g., add to unmanagedPools inside the computeIfAbsent lambda, and validate absence there). public PoolHandle registerUnmanagedPool(String poolName, PoolGroup group) { if (pools.containsKey(poolName)) { throw new IllegalStateException( "Pool [" + poolName + "] already exists as a managed pool and cannot be re-registered as unmanaged; " + "its child allocator has a bounded limit. Register it unmanaged from the start." ); } unmanagedPools.add(poolName); poolConfigs.putIfAbsent(poolName, new PoolConfig(0, Long.MAX_VALUE, group)); return pools.computeIfAbsent(poolName, name -> { BufferAllocator child = root.newChildAllocator(name, 0, Long.MAX_VALUE); return new ArrowPoolHandle(child); }); }

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Close root on import failure If Data.importIntoVectorSchemaRoot throws, freshRoot is never closed and its already-allocated vectors leak against the caller allocator. Wrap the import in a try/catch that closes freshRoot on failure before rethrowing. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DatafusionResultStream.java [137-143] VectorSchemaRoot freshRoot = VectorSchemaRoot.create(schema, allocator); try (ArrowArray arrowArray = ArrowArray.wrap(arrayAddr)) { Data.importIntoVectorSchemaRoot(allocator, arrowArray, freshRoot, dictionaryProvider); +} catch (RuntimeException e) { + freshRoot.close(); + throw e; } nextBatch = freshRoot; batchEmitted = true; return true; Suggestion importance[1-10]: 7 __ Why: Legitimate resource leak concern: if Data.importIntoVectorSchemaRoot throws, the pre-allocated freshRoot vectors would leak. Adding a try/catch to close on failure is a reasonable defensive measure.	Medium
General	Make unmanaged pool registration atomic The pre-check pools.containsKey(poolName) and the mutations that follow are not atomic: two concurrent callers (or a concurrent getOrCreatePool on the same name) can both pass the check, and the unmanagedPools.add/poolConfigs.putIfAbsent/computeIfAbsent sequence can then end up marking a managed pool as unmanaged. Move the entire registration inside a single computeIfAbsent and only add to unmanagedPools/poolConfigs from within the mapping function. plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/ArrowNativeAllocator.java [291-296] -unmanagedPools.add(poolName); -poolConfigs.putIfAbsent(poolName, new PoolConfig(0, Long.MAX_VALUE, group)); -return pools.computeIfAbsent(poolName, name -> { +final boolean[] created = { false }; +ArrowPoolHandle handle = (ArrowPoolHandle) pools.computeIfAbsent(poolName, name -> { BufferAllocator child = root.newChildAllocator(name, 0, Long.MAX_VALUE); + created[0] = true; + poolConfigs.put(name, new PoolConfig(0, Long.MAX_VALUE, group)); + unmanagedPools.add(name); return new ArrowPoolHandle(child); }); +if (!created[0] && !unmanagedPools.contains(poolName)) { + throw new IllegalStateException("Pool [" + poolName + "] already exists as a managed pool ..."); +} +return handle; Suggestion importance[1-10]: 6 __ Why: Valid concurrency concern about the non-atomic check-then-act between the containsKey check and subsequent mutations, which could allow a race with concurrent getOrCreatePool calls. The improved code is a reasonable atomic alternative.	Low
	Avoid consumers on deprecated no-op settings Registering dynamic update consumers for deprecated no-op settings still forces them to be declared as registered settings on the node; if either QUERY_MIN/MAX_SETTING is not declared via getSettings() this addSettingsUpdateConsumer call will throw at startup. Verify these are still included in the plugin's registered settings list, or drop the consumers entirely and rely solely on Setting.Property.Deprecated for the warning. plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/ArrowBasePlugin.java [341-354] if (QUERY_MIN_SETTING.exists(settings) || QUERY_MAX_SETTING.exists(settings)) { logger.warn( "native.allocator.pool.query.min/max are configured but have NO EFFECT: the query pool is " + "unbounded (native enforcement lives Rust-side). Remove these settings." ); } -cs.addSettingsUpdateConsumer( - QUERY_MIN_SETTING, - v -> logger.warn("native.allocator.pool.query.min has no effect: query pool is unbounded") -); -cs.addSettingsUpdateConsumer( - QUERY_MAX_SETTING, - v -> logger.warn("native.allocator.pool.query.max has no effect: query pool is unbounded") -); +// Rely on Setting.Property.Deprecated for dynamic-update warnings; avoid registering +// no-op consumers that require the settings to remain in getSettings(). Suggestion importance[1-10]: 5 __ Why: Valid concern that addSettingsUpdateConsumer requires the settings to be registered, and if they are, the consumer's warning is somewhat redundant with Setting.Property.Deprecated. However, the suggestion is speculative ("Verify...") without concrete evidence of a bug.	Low

[github-actions]

❌ Gradle check result for 7aacbeb: 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?