feat: slow query thread pool

[Swiddis]

Description

Following discussion on #5608, this PR implements a dedicated thread pool for slow queries. This helps keep the cluster responsive in cases of specific expensive queries being run, as the main worker thread pool will still be available to process fast queries. This comes with a plugins.sql.slow_worker_pool.enabled setting (default true).

The pool exists as a separate resource that the worker can hand off to. So the normal fast query path is unchanged, we only hand off in known special cases. This also means we only pay rescheduling overhead for queries that are already slow.

I've tested this by saturating a cluster with >20sec queries (until all slow threads are active and requests are queued), and then verified that normal cheap queries are still responsive even if slow queries are at capacity.

Guide for reviewers:

• The detection rules live in ScriptDetector.java. We primary look for specific expensive UDFs (rex, parse), window functions, joins, or any script pushdown contexts.
  • The biggest class of false-negatives is aggregations with high cardinality, we can't determine from the query plan alone that an aggregation field is very large. We'd need some sort of statistics tracking on the index. I don't want to put all aggregations in the slow pool since they're very common for dashboarding and many aggregations are low-cardinality.
• Execution dispatch is in ThreadPoolExecutionDispatcher, if we hit the ScriptDetector check we do a handoff involving a lot of thread context propagation and cleanup. This is the riskiest part of the code, I'm still leaving in draft as I test that all the context work here is correct and cleans up properly.

Related Issues

Resolves #5608

Check List

• New functionality includes testing.
• New functionality has been documented.
• New functionality has javadoc added.
• New functionality has a user manual doc added.
• New PPL command checklist all confirmed.
• API changes companion pull request created.
• Commits are signed per the DCO using --signoff or -s.
• Public documentation issue/PR created.

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

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue The CancellableTask is captured on the calling thread but may be cleared before the scheduled task runs. If another request on the same thread sets a different task between line 52 and the scheduled execution, the wrong task will be propagated to the slow pool. This occurs because getCancellableTask() reads from a ThreadLocal that can be modified by subsequent requests on the same thread before the scheduled runnable executes. CancellableTask task = OpenSearchQueryManager.getCancellableTask(); Possible Issue The CancellableTask is not propagated to the slow worker pool in this execution path. Unlike ThreadPoolExecutionDispatcher which captures and restores the task, this code only propagates metadata provider and context but omits the cancellable task. If a user cancels a slow query dispatched via this path, the cancellation will not reach the executing thread, leaving the query running indefinitely. LOG.debug("Query contains scripts, routing to slow worker pool"); JaninoRelMetadataProvider metadataProvider = RelMetadataQueryBase.THREAD_PROVIDERS.get(); long currentTime = Hook.CURRENT_TIME.get(-1L); boolean stripNullCols = CalcitePlanContext.stripNullColumns.get(); String twUnitName = CalcitePlanContext.timewrapUnitName.get(); String twSeries = CalcitePlanContext.timewrapSeries.get(); client .threadPool() .schedule( withCurrentContext( () -> { if (metadataProvider != null) { RelMetadataQueryBase.THREAD_PROVIDERS.set(metadataProvider); } Hook.Closeable hookHandle = null; if (currentTime >= 0) { hookHandle = Hook.CURRENT_TIME.addThread( (Consumer<Holder<Long>>) h -> h.set(currentTime)); } CalcitePlanContext.stripNullColumns.set(stripNullCols); CalcitePlanContext.timewrapUnitName.set(twUnitName); CalcitePlanContext.timewrapSeries.set(twSeries); try { executeTask.run(); } catch (Exception e) { closingListener.onFailure(e); } finally { if (hookHandle != null) { hookHandle.close(); } RelMetadataQueryBase.THREAD_PROVIDERS.remove(); CalcitePlanContext.clearTimewrapSignals(); } }), new TimeValue(0), SQL_SLOW_WORKER_THREAD_POOL_NAME); Resource Leak If executeTask.run() throws an exception at line 295, the hookHandle and thread-locals are never cleaned up because the finally block only exists inside the scheduled task (lines 284-290), not around the inline execution path. This leaks Calcite metadata providers and Hook handles when queries fail on the fast path. } else { executeTask.run(); }

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Security	Clear ThreadContext after execution The scheduled task does not clear ThreadContext after execution, which can cause context pollution across queries. Add ThreadContext.clearAll() in the finally block to ensure proper cleanup of security and logging context between requests. opensearch/src/main/java/org/opensearch/sql/opensearch/executor/ThreadPoolExecutionDispatcher.java [59-89] threadPool.schedule( () -> { ThreadContext.putAll(ctx); OpenSearchQueryManager.setCancellableTask(task); if (metadataProvider != null) { RelMetadataQueryBase.THREAD_PROVIDERS.set(metadataProvider); } Hook.Closeable hookHandle = null; if (currentTime >= 0) { hookHandle = Hook.CURRENT_TIME.addThread( (Consumer<org.apache.calcite.util.Holder<Long>>) h -> h.set(currentTime)); } CalcitePlanContext.stripNullColumns.set(stripNullCols); CalcitePlanContext.timewrapUnitName.set(twUnitName); CalcitePlanContext.timewrapSeries.set(twSeries); try { engine.execute(plan, context, listener); } catch (Exception e) { listener.onFailure(e); } finally { if (hookHandle != null) { hookHandle.close(); } OpenSearchQueryManager.clearCancellableTask(); RelMetadataQueryBase.THREAD_PROVIDERS.remove(); CalcitePlanContext.clearTimewrapSignals(); + ThreadContext.clearAll(); } }, new TimeValue(0), SQL_SLOW_WORKER_THREAD_POOL_NAME); Suggestion importance[1-10]: 9 __ Why: This is a critical security issue. Without clearing ThreadContext after execution, security context (DLS/FLS filters) can leak between queries on the same thread pool worker. This could allow unauthorized data access across different users' queries.	High