Propagate security context to sql-worker thread pool

Wrap scheduled lambdas in both execute() and executeViaTransport() with
withCurrentContext() to capture and restore ThreadContext (user identity,
permissions, audit trail) on the worker thread. Follows the same pattern
as OpenSearchQueryManager.withCurrentContext().

Signed-off-by: Kai Huang <kaihuang@amazon.com>
Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java

@@ -11,10 +11,12 @@
 import static org.opensearch.sql.opensearch.executor.OpenSearchQueryManager.SQL_WORKER_THREAD_POOL_NAME;
 import static org.opensearch.sql.protocol.response.format.JsonResponseFormatter.Style.PRETTY;
 
+import java.util.Map;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.schema.impl.AbstractSchema;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.ThreadContext;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.rest.RestStatus;
@@ -84,7 +86,7 @@ public void execute(String query, QueryType queryType, RestChannel channel) {
     client
         .threadPool()
         .schedule(
-            () -> doExecute(query, queryType, channel),
+            withCurrentContext(() -> doExecute(query, queryType, channel)),
             new TimeValue(0),
             SQL_WORKER_THREAD_POOL_NAME);
   }
@@ -107,7 +109,7 @@ public void executeViaTransport(
     client
         .threadPool()
         .schedule(
-            () -> doExecuteViaTransport(query, queryType, pplRequest, listener),
+            withCurrentContext(() -> doExecuteViaTransport(query, queryType, pplRequest, listener)),
             new TimeValue(0),
             SQL_WORKER_THREAD_POOL_NAME);
   }
@@ -298,6 +300,19 @@ private static void recordFailureMetric(QueryType queryType, Exception e) {
     }
   }
 
+  /**
+   * Capture current thread context and restore it on the worker thread. Ensures security context
+   * (user identity, permissions) is propagated. Same pattern as {@link
+   * org.opensearch.sql.opensearch.executor.OpenSearchQueryManager#withCurrentContext}.
+   */
+  private static Runnable withCurrentContext(final Runnable task) {
+    final Map<String, String> currentContext = ThreadContext.getImmutableContext();
+    return () -> {
+      ThreadContext.putAll(currentContext);
+      task.run();
+    };
+  }
+
   private static void reportError(RestChannel channel, Exception e) {
     RestStatus status =
         isClientError(e) ? RestStatus.BAD_REQUEST : RestStatus.INTERNAL_SERVER_ERROR;