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've been carefully checking that the context propagation is working correctly and that we handle things like auth context (field level security) and cancellation.

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 🔍

(Review updated until commit 434021f)

Here are some key observations to aid the review process:

🧪 No relevant tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue The dispatchInternal method catches exceptions during task execution and calls failureListener.onFailure(e), but failureListener can be null (as seen in dispatchTask). If an exception occurs when failureListener is null, the exception is silently swallowed. This means errors in analytics engine execution via dispatchTask will not be reported, potentially hiding failures. try { task.run(); } catch (Exception e) { if (failureListener != null) { failureListener.onFailure(e); } } finally { Possible Issue The optimization step was moved from CalciteToolsHelper.run to executeCalcitePlan, meaning the plan is now optimized twice: once in executeCalcitePlan (line 242) and again inside CalciteToolsHelper.run (which is called by OpenSearchRelRunners.run). This duplication wastes CPU cycles and may cause unexpected behavior if optimization is not idempotent. RelNode optimizedPlan = CalciteToolsHelper.optimize(calcitePlan, context); // Wrap execution with EXECUTING stage tracking — dispatch via // ExecutionDispatcher which may route to a slow worker pool StageErrorHandler.executeStageVoid( QueryProcessingStage.EXECUTING, () -> executionDispatcher.dispatch(optimizedPlan, context, listener, executionEngine), Resource Leak If Hook.CURRENT_TIME.addThread is called (line 82-84) but an exception occurs before the finally block, hookHandle.close() may not be called if hookHandle assignment fails. While the finally block does check for null, if the exception happens during addThread itself, the hook may not be properly cleaned up. 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 { task.run(); } catch (Exception e) { if (failureListener != null) { failureListener.onFailure(e); } } finally { if (hookHandle != null) { hookHandle.close(); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 434021f

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Add null check for PushDownContext The method doesn't handle the case where getPushDownContext() returns null. If a scan node has no push-down context, calling methods on null will throw NullPointerException. Add a null check before accessing context methods. opensearch/src/main/java/org/opensearch/sql/opensearch/executor/ScriptDetector.java [73-83] private static boolean scanHasScripts(AbstractCalciteIndexScan scan) { PushDownContext ctx = scan.getPushDownContext(); + if (ctx == null) { + return false; + } if (ctx.isScriptPushed()) { return true; } if (ctx.isSortExprPushed()) { return true; } AggSpec aggSpec = ctx.getAggSpec(); return aggSpec != null && aggSpec.getScriptCount() > 0; } Suggestion importance[1-10]: 8 __ Why: The suggestion identifies a potential NullPointerException if getPushDownContext() returns null. This is a valid concern that could cause runtime failures. Adding a null check prevents crashes and improves robustness.	Medium
	Log exceptions before conditional handling The exception caught during task execution is swallowed when failureListener is null. This can hide critical errors in the dispatchTask path. Consider logging the exception at ERROR level before the null check to ensure visibility of failures. opensearch/src/main/java/org/opensearch/sql/opensearch/executor/ThreadPoolExecutionDispatcher.java [88-94] } catch (Exception e) { + LOG.error("Exception during task execution on slow pool", e); if (failureListener != null) { failureListener.onFailure(e); } } finally { Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that exceptions are swallowed when failureListener is null in the dispatchTask path. Adding logging ensures visibility of failures, which is important for debugging. However, this is a moderate improvement rather than a critical bug fix.	Medium

---

Previous suggestions

Suggestions up to commit 6fdd82e

Category	Suggestion	Impact
Possible issue	Add null check for PushDownContext The method doesn't handle the case where getPushDownContext() returns null. If scan.getPushDownContext() is null, calling methods on ctx will throw a NullPointerException. Add a null check before accessing the context. opensearch/src/main/java/org/opensearch/sql/opensearch/executor/ScriptDetector.java [73-83] private static boolean scanHasScripts(AbstractCalciteIndexScan scan) { PushDownContext ctx = scan.getPushDownContext(); + if (ctx == null) { + return false; + } if (ctx.isScriptPushed()) { return true; } if (ctx.isSortExprPushed()) { return true; } AggSpec aggSpec = ctx.getAggSpec(); return aggSpec != null && aggSpec.getScriptCount() > 0; } Suggestion importance[1-10]: 8 __ Why: This is a valid defensive programming suggestion. While AbstractCalciteIndexScan implementations likely always return a non-null PushDownContext, adding a null check prevents potential NullPointerException and makes the code more robust. The suggestion correctly identifies a potential issue and provides appropriate handling.	Medium
	Log unhandled exceptions in slow pool When failureListener is null and an exception occurs, the error is silently swallowed. This can hide critical failures in the slow pool execution path. Consider logging the exception or re-throwing it to ensure visibility of errors. opensearch/src/main/java/org/opensearch/sql/opensearch/executor/ThreadPoolExecutionDispatcher.java [88-94] } catch (Exception e) { if (failureListener != null) { failureListener.onFailure(e); + } else { + LOG.error("Unhandled exception in slow pool execution", e); } } finally { Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that exceptions are silently swallowed when failureListener is null. However, looking at the code context, failureListener is only null when called via dispatchTask, which is used for the analytics engine path that has its own error handling. The suggestion is valid but the impact is moderate since the analytics path handles errors separately.	Medium
	Handle exceptions in analytics engine execution The executeTask runnable doesn't handle exceptions that may occur during analyticsEngine.execute or executeWithProfile. If an exception is thrown, it won't be caught and passed to closingListener.onFailure. Wrap the engine calls in try-catch blocks within the runnable. plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java [240-254] Runnable executeTask = profiling - ? () -> - analyticsEngine.executeWithProfile( - finalPlan, - planContext, - queryCtx, - createQueryListener(queryType, profileCtx, closingListener)) - : () -> - analyticsEngine.execute( - finalPlan, - planContext, - queryCtx, - createQueryListener(queryType, profileCtx, closingListener)); + ? () -> { + try { + analyticsEngine.executeWithProfile( + finalPlan, + planContext, + queryCtx, + createQueryListener(queryType, profileCtx, closingListener)); + } catch (Exception e) { + closingListener.onFailure(e); + } + } + : () -> { + try { + analyticsEngine.execute( + finalPlan, + planContext, + queryCtx, + createQueryListener(queryType, profileCtx, closingListener)); + } catch (Exception e) { + closingListener.onFailure(e); + } + }; executionDispatcher.dispatchTask(finalPlan, planContext, executeTask); Suggestion importance[1-10]: 6 __ Why: The suggestion identifies a potential issue with exception handling in the executeTask runnable. However, examining the broader context shows that dispatchTask already wraps execution in a try-catch block (lines 88-94 in ThreadPoolExecutionDispatcher), and the outer try-catch at line 255 also catches exceptions. While adding explicit exception handling in the runnable would be more defensive, the current implementation already has error handling coverage.	Low

Suggestions up to commit 46c3cbe

Category	Suggestion	Impact
Possible issue	Add null check for PushDownContext The method doesn't handle the case where getPushDownContext() returns null, which could cause a NullPointerException. Add a null check for ctx before accessing its methods to prevent potential crashes during plan inspection. opensearch/src/main/java/org/opensearch/sql/opensearch/executor/ScriptDetector.java [73-83] private static boolean scanHasScripts(AbstractCalciteIndexScan scan) { PushDownContext ctx = scan.getPushDownContext(); + if (ctx == null) { + return false; + } if (ctx.isScriptPushed()) { return true; } if (ctx.isSortExprPushed()) { return true; } AggSpec aggSpec = ctx.getAggSpec(); return aggSpec != null && aggSpec.getScriptCount() > 0; } Suggestion importance[1-10]: 8 __ Why: Adding a null check for ctx prevents potential NullPointerException crashes during plan inspection. This is a defensive programming practice that improves robustness, especially since getPushDownContext() could return null in edge cases.	Medium
General	Move optimization inside error tracking stage The optimization step is outside the EXECUTING stage error handler, meaning optimization failures won't be properly tracked or reported. Move CalciteToolsHelper.optimize inside the executeStageVoid lambda to ensure optimization errors are caught and attributed to the correct stage. core/src/main/java/org/opensearch/sql/executor/QueryService.java [212-223] -// Optimize before dispatch so the dispatcher's ScriptDetector -// sees the post-optimization plan for accurate routing. -RelNode optimizedPlan = CalciteToolsHelper.optimize(calcitePlan, context); - // Wrap execution with EXECUTING stage tracking — dispatch via // ExecutionDispatcher which may route to a slow worker pool StageErrorHandler.executeStageVoid( QueryProcessingStage.EXECUTING, - () -> - executionDispatcher.dispatch( - optimizedPlan, context, listener, executionEngine), + () -> { + // Optimize before dispatch so the dispatcher's ScriptDetector + // sees the post-optimization plan for accurate routing. + RelNode optimizedPlan = CalciteToolsHelper.optimize(calcitePlan, context); + executionDispatcher.dispatch( + optimizedPlan, context, listener, executionEngine); + }, "while running the query"); Suggestion importance[1-10]: 7 __ Why: Moving CalciteToolsHelper.optimize inside the EXECUTING stage handler ensures optimization errors are properly tracked and attributed. This improves error reporting consistency, though the current placement doesn't cause functional issues since optimization failures would still be caught by the outer try-catch.	Medium
	Add exception logging in slow pool The exception handler only notifies the listener but doesn't log the error. Add logging before calling failureListener.onFailure(e) to ensure exceptions in the slow worker pool are visible for debugging and monitoring. opensearch/src/main/java/org/opensearch/sql/opensearch/executor/ThreadPoolExecutionDispatcher.java [88-94] } catch (Exception e) { + LOG.error("Exception during slow pool execution", e); if (failureListener != null) { failureListener.onFailure(e); } } finally { Suggestion importance[1-10]: 6 __ Why: Adding logging for exceptions in the slow worker pool improves observability and debugging. However, the suggestion is a minor enhancement rather than fixing a critical issue, as the exception is already propagated to the failureListener.	Low

Suggestions up to commit 5b0d7b2

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

[github-actions]

Persistent review updated to latest commit 46c3cbe

[github-actions]

Persistent review updated to latest commit 6fdd82e

[github-actions]

Persistent review updated to latest commit 434021f