diff --git a/common/src/main/java/org/opensearch/sql/common/setting/Settings.java b/common/src/main/java/org/opensearch/sql/common/setting/Settings.java
index 67643a80add..fcdf84d9e8f 100644
--- a/common/src/main/java/org/opensearch/sql/common/setting/Settings.java
+++ b/common/src/main/java/org/opensearch/sql/common/setting/Settings.java
@@ -38,6 +38,18 @@ public enum Key {
PPL_JOIN_SUBSEARCH_MAXOUT("plugins.ppl.join.subsearch_maxout"),
PPL_REST_ALLOWED_ENDPOINTS("plugins.ppl.rest.allowed_endpoints"),
+ /** Maximum number of asynchronous PPL queries running concurrently on one node. */
+ PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES("plugins.ppl.async.node_concurrent_running_queries"),
+
+ /** Maximum number of asynchronous PPL jobs retained on one node. */
+ PPL_ASYNC_MAX_RETAINED_JOBS("plugins.ppl.async.max_retained_jobs"),
+
+ /** Maximum accepted submit wait-for-completion timeout. */
+ PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT("plugins.ppl.async.max_wait_for_completion_timeout"),
+
+ /** Maximum accepted asynchronous PPL job lease. */
+ PPL_ASYNC_MAX_KEEP_ALIVE("plugins.ppl.async.max_keep_alive"),
+
/** Enable Calcite as execution engine */
CALCITE_ENGINE_ENABLED("plugins.calcite.enabled"),
CALCITE_FALLBACK_ALLOWED("plugins.calcite.fallback.allowed"),
diff --git a/core/src/main/java/org/opensearch/sql/executor/AsyncQueryExecution.java b/core/src/main/java/org/opensearch/sql/executor/AsyncQueryExecution.java
new file mode 100644
index 00000000000..c52c124a099
--- /dev/null
+++ b/core/src/main/java/org/opensearch/sql/executor/AsyncQueryExecution.java
@@ -0,0 +1,34 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.executor;
+
+import java.util.Optional;
+import java.util.concurrent.CompletionStage;
+import org.opensearch.sql.executor.ExecutionEngine.QueryResponse;
+
+/**
+ * Lifecycle-facing handle for one asynchronous query execution.
+ *
+ * The execution module owns result production and execution-specific resources. The lifecycle
+ * module owns this handle after submission and uses it to read the current result, observe terminal
+ * completion, and release those resources.
+ *
+ *
On successful completion, the authoritative final result must be visible through {@link
+ * #currentResult()} before {@link #completion()} completes normally. Implementations must make
+ * {@link #close()} idempotent and safe to call concurrently with {@link #currentResult()}.
+ */
+public interface AsyncQueryExecution extends AutoCloseable {
+
+ /** Returns the complete result currently visible, or empty before a result is available. */
+ Optional currentResult();
+
+ /** Completes normally on query success and exceptionally on query failure. */
+ CompletionStage completion();
+
+ /** Releases execution-owned result resources. */
+ @Override
+ void close();
+}
diff --git a/docs/dev/query-job-mvp-design.md b/docs/dev/query-job-mvp-design.md
new file mode 100644
index 00000000000..0a534014804
--- /dev/null
+++ b/docs/dev/query-job-mvp-design.md
@@ -0,0 +1,433 @@
+# Query Job Lifecycle — MVP Redesign
+
+> Response to review of [PR #5809](https://github.com/opensearch-project/sql/pull/5809).
+> This document is design-only; no code has been changed.
+
+## 1. What we heard from review
+
+Three concerns from @dai-chen shaped this redesign.
+
+| # | Concern | Where it landed |
+|---|---------|-----------------|
+| 1 | Make the lifecycle layer language-neutral, similar to Livy's control plane over Spark. | The `job` package no longer names PPL. It talks only to a `QueryRunner` SPI. Language modules (`ppl`, `sql`, `async-query`) implement `QueryRunnerFactory`. |
+| 2 | The execution boundary can be engine-neutral too — SQL, PPL, and AE all run behind it. | `QueryRunner` returns a neutral `QueryResult`. It knows nothing about ANTLR, `PPLService`, `SQLService`, or Spark. The same interface fits AE. |
+| 3 | Separate execution state from retention. In a BigQuery-style model the job stays `RUNNING` regardless of whether a caller is still waiting. Treat `wait_for_completion_timeout` as submit-response behavior only. | The state machine has five values (`PENDING`, `RUNNING`, `SUCCEEDED`, `FAILED`, `CANCELLED`) and no retention axis. Waiting is a *transport* concern, not a *lifecycle* concern. |
+| 3a | Extract internal responsibilities (JobStore, LeaseManager, …) from the service class. | `QueryJobService`, `QueryJobStore`, and `QueryJob` split along Single-Responsibility lines. `LeaseManager` is intentionally out of MVP; it can be added without changing the public interfaces (§7). |
+
+## 2. MVP scope
+
+**In**
+
+- One neutral job model that carries both sync and async queries.
+- Submit / get / cancel over a small, engine-agnostic control plane.
+- Owner-node routing via an opaque `QueryJobId`.
+- Caller identity captured at submit and checked at get / cancel.
+
+**Out (deferred without breaking the API)**
+
+- `wait_for_completion_timeout` — MVP treats every submission as `wait_for_completion=forever`. The transport blocks on `QueryJob#completion()` until it resolves.
+- `keep_alive` and retention leases — no expiry timer, no keep-alive renewal. Terminal jobs live only long enough for the caller to observe them.
+- Admission control (`maxRunningQueries`, `maxRetainedJobs`).
+- Persistent job store.
+- Partial-result streaming (already deferred by the current PR).
+
+Everything under **Out** is a store, service, or transport concern. None of it changes the class surface described in §4.
+
+## 3. Livy inspiration in one paragraph
+
+Livy is a REST control plane that owns *sessions* and *statements* on top of Spark. Livy has no compiler and no query engine — it accepts a language tag (`scala`, `python`, `sql`, `r`), forwards the statement to the right session, and tracks a state machine on top. The session state machine is small: waiting → running → available | error | cancelled. Language is data; the control plane is code. We adopt the same split: `QueryJob` is language-blind, and `QueryRunner` is the per-engine adapter.
+
+## 4. Class model
+
+### 4.1 Package layout
+
+```
+core/src/main/java/org/opensearch/sql/job/
+├── QueryJob.java // active object (state machine)
+├── QueryJobId.java // opaque, node-routable identifier
+├── QueryJobState.java // PENDING, RUNNING, SUCCEEDED, FAILED, CANCELLED
+├── QueryJobStatus.java // immutable snapshot
+├── QueryJobStore.java // registry SPI (in-memory MVP impl provided)
+├── QueryJobService.java // control plane: submit / get / cancel
+├── QueryRunner.java // engine SPI: run() + cancel()
+├── QueryRunnerFactory.java // engine SPI: build a runner from a request
+├── QueryResult.java // neutral final result
+├── QueryFailure.java // neutral failure descriptor
+├── Principal.java // caller identity + authorization
+└── SubmitRequest.java // neutral submission payload
+```
+
+`core` already has no dependency on other modules, so `job` is free of language and engine imports. `ppl`, `sql`, `async-query`, and future engines depend on `job` — not the other way around.
+
+### 4.2 Class diagram
+
+```mermaid
+classDiagram
+ class QueryJobService {
+ <>
+ +submit(SubmitRequest) QueryJob
+ +get(QueryJobId, Principal) QueryJobStatus
+ +cancel(QueryJobId, Principal) QueryJobStatus
+ }
+
+ class QueryJob {
+ -QueryJobId id
+ -Principal owner
+ -QueryRunner runner
+ -QueryJobState state
+ -CompletableFuture~QueryResult~ completion
+ +id() QueryJobId
+ +owner() Principal
+ +status() QueryJobStatus
+ +completion() CompletionStage~QueryResult~
+ +cancel() void
+ }
+
+ class QueryJobStore {
+ <>
+ +register(QueryJob) QueryJob
+ +find(QueryJobId) Optional~QueryJob~
+ +remove(QueryJobId, QueryJob) boolean
+ +jobs() Collection~QueryJob~
+ +close() void
+ }
+
+ class QueryRunner {
+ <>
+ +run() CompletionStage~QueryResult~
+ +cancel() void
+ }
+
+ class QueryRunnerFactory {
+ <>
+ +create(SubmitRequest) QueryRunner
+ }
+
+ class QueryJobId {
+ <>
+ +ownerNodeId String
+ +contextId String
+ +encode() String
+ +parse(String)$ QueryJobId
+ }
+
+ class QueryJobStatus {
+ <>
+ +id QueryJobId
+ +state QueryJobState
+ +submittedAtMillis long
+ +startedAtMillis OptionalLong
+ +completedAtMillis OptionalLong
+ +failure Optional~QueryFailure~
+ }
+
+ class Principal {
+ <>
+ +name String
+ +tenant String
+ +backendRoles List~String~
+ +authorize(Principal) void
+ }
+
+ class QueryResult {
+ <>
+ +schema Schema
+ +rows List~ExprValue~
+ +tookMillis long
+ }
+
+ class QueryFailure {
+ <>
+ +type String
+ +reason String
+ }
+
+ QueryJobService --> QueryJobStore : uses
+ QueryJobService --> QueryRunnerFactory : uses
+ QueryJobService --> QueryJob : creates
+ QueryJob --> QueryRunner : delegates execution
+ QueryJob --> Principal : owner
+ QueryJob --> QueryJobId : id
+ QueryJob --> QueryJobStatus : projects
+ QueryJobStore --> QueryJob : holds
+ QueryJobStatus --> QueryJobId
+ QueryJobStatus --> QueryJobState
+ QueryJobStatus --> QueryFailure
+```
+
+### 4.3 State machine
+
+Retention is not part of state. Waiting is not part of state.
+
+```
+ submit()
+ │
+ ▼
+ ┌───────┐ runner starts ┌────────┐
+ │PENDING│─────────────────▶│RUNNING │
+ └───────┘ └────┬───┘
+ │
+ ┌───────────────────────┼───────────────────────┐
+ ▼ ▼ ▼
+ ┌─────────┐ ┌────────┐ ┌─────────┐
+ │SUCCEEDED│ │ FAILED │ │CANCELLED│
+ └─────────┘ └────────┘ └─────────┘
+
+Terminal transitions are idempotent. cancel() from any non-terminal state
+moves to CANCELLED; cancel() from a terminal state is a no-op.
+```
+
+## 5. Interface contracts
+
+Signatures below define the contract only. JavaDoc phrased for readers of Effective Java: each interface has one reason to exist (Item 20) and each record documents its invariants (Item 17).
+
+### 5.1 `QueryJobId`
+
+```java
+public record QueryJobId(String ownerNodeId, String contextId) {
+ public QueryJobId {
+ // both fields must be non-blank
+ }
+ /** Fresh ID with a random context. */
+ public static QueryJobId create(String ownerNodeId) { … }
+ /** URL-safe, versioned, opaque encoding. */
+ public String encode() { … }
+ /** Uniform error on any parse failure. */
+ public static QueryJobId parse(String encoded) { … }
+}
+```
+
+Kept intentionally identical to today's `QueryJobId`. The encoding is a public wire format; a design-level rewrite must not disturb it.
+
+### 5.2 `Principal`
+
+```java
+public record Principal(String name, String tenant, List backendRoles) {
+ public static final Principal UNSECURED = new Principal(null, null, List.of());
+ public static Principal current(ThreadContext ctx) { … }
+ public void authorize(Principal caller) { … }
+}
+```
+
+Renamed from `QueryJobOwner` for neutrality — the identity is not job-specific and will be reused by any future admin action.
+
+### 5.3 `QueryJobState`, `QueryJobStatus`, `QueryFailure`
+
+```java
+public enum QueryJobState {
+ PENDING, RUNNING, SUCCEEDED, FAILED, CANCELLED;
+ public boolean isTerminal() { … }
+}
+
+public record QueryJobStatus(
+ QueryJobId id,
+ QueryJobState state,
+ long submittedAtMillis,
+ OptionalLong startedAtMillis,
+ OptionalLong completedAtMillis,
+ Optional failure,
+ Optional result) {
+ // result present iff state == SUCCEEDED
+ // failure present iff state == FAILED
+}
+
+public record QueryFailure(String type, String reason) {
+ public static QueryFailure of(Throwable t) { … }
+}
+```
+
+`QueryJobStatus` collapses the previous `Snapshot` sealed hierarchy. The three record variants (`Running`, `Succeeded`, `Failed`) exist today only because retention returned different shapes; without retention, one record with a state discriminator is smaller and enforces the same invariants via record validation.
+
+### 5.4 `QueryResult`
+
+```java
+public record QueryResult(Schema schema, List rows, long tookMillis) {}
+```
+
+Uses the existing `core` types (`ExecutionEngine.Schema`, `ExprValue`) so all engines already implementing them can produce a `QueryResult` without introducing new dependencies.
+
+### 5.5 `SubmitRequest`
+
+```java
+public record SubmitRequest(
+ String language, // "ppl" | "sql" | ...
+ String statement,
+ Map params, // engine-specific, opaque to core
+ Principal submitter) {}
+```
+
+Language is data. The service uses it to pick a `QueryRunnerFactory`. Neither the service nor the job parses `statement` or reads `params`.
+
+### 5.6 `QueryRunner` and `QueryRunnerFactory`
+
+```java
+public interface QueryRunner {
+ /** Single-use. Multiple invocations must throw IllegalStateException. */
+ CompletionStage run();
+
+ /** Idempotent cooperative cancel. Safe to call before run() and after completion. */
+ void cancel();
+}
+
+public interface QueryRunnerFactory {
+ /** Language handled by this factory, matched against SubmitRequest#language. */
+ String language();
+ QueryRunner create(SubmitRequest request);
+}
+```
+
+Two-method interface (Item 21: interfaces are for use, not for reuse). PPL, SQL, and AE each provide one implementation and register it via Guice. The lifecycle package never imports any of them.
+
+### 5.7 `QueryJob`
+
+```java
+public final class QueryJob {
+
+ /** Package-private: only QueryJobService instantiates jobs. */
+ QueryJob(QueryJobId id, Principal owner, QueryRunner runner, Clock clock) { … }
+
+ public QueryJobId id() { … }
+ public Principal owner() { … }
+ public QueryJobStatus status() { … } // snapshot, never throws
+ public CompletionStage completion() { … }
+ public void cancel() { … } // idempotent
+}
+```
+
+- Thread safety documented at the class level (Effective Java Item 82). All mutable fields are guarded by `this`; every side effect (runner cancel, store removal, listener notification) happens **after** the monitor is released, exactly as today's `QueryJob` does.
+- `completion()` returns a `CompletionStage` view. The internal `CompletableFuture` is never leaked, so callers cannot complete the job externally (Item 15).
+- `cancel()` is the *only* public mutator. Retention, expiry, and admission are not job concerns.
+
+### 5.8 `QueryJobStore`
+
+```java
+public interface QueryJobStore extends Closeable {
+ QueryJob register(QueryJob job); // returns existing on duplicate ID
+ Optional find(QueryJobId id);
+ boolean remove(QueryJobId id, QueryJob job); // conditional
+ Collection jobs();
+ @Override void close(); // drops residual jobs
+}
+```
+
+The MVP ships one implementation: `InMemoryQueryJobStore`, a thin wrapper around `ConcurrentHashMap`. A persistent (system-index) implementation can be added later without touching `QueryJob` or `QueryJobService`.
+
+### 5.9 `QueryJobService`
+
+```java
+public interface QueryJobService {
+ QueryJob submit(SubmitRequest request);
+ QueryJobStatus get(QueryJobId id, Principal caller);
+ QueryJobStatus cancel(QueryJobId id, Principal caller);
+}
+```
+
+A single production implementation, `LocalQueryJobService`, wires:
+
+- `QueryJobStore` for the registry,
+- a `Map` keyed by language,
+- a `Clock` supplier,
+- a local-node-id supplier for job ID minting.
+
+The service is **not** an `AbstractLifecycleComponent`. Lifecycle wiring (start / stop) lives in the plugin module (`AsyncQueryLifecycle`), keeping `core` free of OpenSearch node types.
+
+## 6. Sync = async with `wait_for_timeout=forever`
+
+Only the transport layer knows whether a caller is willing to wait.
+
+```mermaid
+sequenceDiagram
+ autonumber
+ participant Client
+ participant Transport as SyncTransport
+ participant Service as QueryJobService
+ participant Job as QueryJob
+ participant Runner as QueryRunner
+
+ Client->>Transport: POST /_plugins/_ppl {query}
+ Transport->>Service: submit(request)
+ Service->>Job: new QueryJob(id, owner, runner)
+ Service->>Runner: run()
+ Service-->>Transport: QueryJob
+ Runner-->>Job: result / failure
+ Transport->>Job: completion().join()
+ Job-->>Transport: QueryResult
+ Transport-->>Client: formatted response
+```
+
+The async transport differs only after step 5:
+
+```mermaid
+sequenceDiagram
+ autonumber
+ participant Client
+ participant Transport as AsyncTransport
+ participant Service as QueryJobService
+ participant Job as QueryJob
+
+ Client->>Transport: POST /_plugins/_async_ppl {query}
+ Transport->>Service: submit(request)
+ Service-->>Transport: QueryJob
+ Transport-->>Client: 202 {jobId}
+ Note over Client,Transport: later …
+ Client->>Transport: GET /_plugins/_async_ppl/{jobId}
+ Transport->>Service: get(jobId, caller)
+ Service-->>Transport: QueryJobStatus
+ Transport-->>Client: status (+ result if SUCCEEDED)
+```
+
+Both paths use the same `QueryJob`. Nothing about the job knows which one was chosen.
+
+## 7. Extensibility without churn
+
+Every deferred feature lands behind an interface that already exists.
+
+| Feature | Change surface |
+|---------|----------------|
+| `wait_for_completion_timeout` | Transport chooses `completion().orTimeout(...)` and returns `QueryJobStatus.RUNNING` on timeout. Job unchanged. |
+| `keep_alive` | New `RetentionPolicy` composed into `QueryJobService`. Adds an `AutoCloseable` timer per terminal job; store gains `evict()`. Job unchanged. |
+| Admission control | `AdmissionController` interface, invoked before `store.register`. Job unchanged. |
+| Persistent job metadata | Alternate `QueryJobStore` implementation. Service and job unchanged. |
+| SQL and AE support | New `QueryRunnerFactory` implementations. Zero core changes. |
+| Progress / partial results | `QueryRunner` gains a second method (default returning empty). All existing engines compile without change. |
+
+Because each seam is an interface with one reason to change, the deferrals really are deferrals — not "we'll rewrite later".
+
+## 8. Effective Java / Clean Code checklist
+
+- **Item 1** — Static factory methods where they help: `QueryJobId.create`, `QueryFailure.of`, `Principal.current`. Constructors are hidden or package-private.
+- **Item 15** — `QueryJob`'s `CompletableFuture` is never exposed; only its `CompletionStage` view is.
+- **Item 17** — `QueryJobId`, `QueryJobStatus`, `QueryFailure`, `QueryResult`, `Principal`, `SubmitRequest` are `record`s with compact-constructor validation.
+- **Item 18** — `QueryJobService` composes a store, factories, and a clock. No inheritance chain.
+- **Item 20** — Every seam is an interface. Alternate engines and stores drop in.
+- **Item 22** — Each interface expresses one role; no marker or constant interfaces.
+- **Item 24** — `QueryJobStatus` is a top-level record, not a nested type of `QueryJob`.
+- **Item 55** — `Optional` on values that legitimately may be absent (`find`, `failure`, `result`, `startedAtMillis`). `OptionalLong` for the primitive fields (Item 55 forbids `Optional`).
+- **Item 82** — `QueryJob` documents thread-safety at class level and lists guarded fields.
+- **Clean Code, small classes** — the biggest class in the package is `QueryJob`; its public API is four methods.
+- **Clean Code, intent-revealing names** — `submit` / `get` / `cancel`, `register` / `find` / `remove`. No `handleXxx`, no `processYyy`.
+- **Clean Code, no duplication** — sync and async share one `QueryJob`; only the transport layer differs.
+
+## 9. What migrates from the current PR
+
+The existing branch already ships the plumbing this design keeps. The redesign is refactor-shaped, not rewrite-shaped:
+
+| Today (`feat/query-job-refactor`) | MVP redesign |
+|-----------------------------------|--------------|
+| `plugin.transport.asyncquery.QueryJob` (983 lines) | `core.job.QueryJob` (~250 lines after retention drops out). |
+| `QueryJob.Snapshot` sealed hierarchy (`Running`, `Succeeded`, `Failed`) | `QueryJobStatus` record with a `state` discriminator. |
+| `QueryJobRegistry` | `QueryJobStore` interface + `InMemoryQueryJobStore`. |
+| `QueryJobOwner` | `Principal`. |
+| `AsyncQueryExecution` (in `core.executor`) | `QueryRunner` (in `core.job`). Same shape, engine-neutral name and location. |
+| `DefaultAsyncQueryExecution` (in `ppl`) | `PPLQueryRunner` in `ppl` implementing `QueryRunner`; symmetrical `SQLQueryRunner` in `sql`. |
+| `PPLAsyncQueryService` (mentioned in dai-chen's comment #3, not yet in-tree) | `LocalQueryJobService` — no retention, no leases, no admission for MVP. |
+| `wait_for_completion_timeout` state axis | Transport-level `completion().orTimeout(...)` in a later PR. |
+| `keep_alive` expiry timer | Deferred; lands as a `RetentionPolicy` composed into the service. |
+
+## 10. Open questions
+
+1. Do we want `QueryResult` to model streaming from day one (a `Publisher` instead of `List`) so partial results are additive later? MVP says no; two engines that already produce final results would need adaptation. Flagging for review.
+2. Should `Principal.current` live on `Principal` or in a `SecurityAdapter` interface? Today it reaches into `ThreadContext`; that couples `core.job` to OpenSearch. A `SecurityAdapter` SPI is cleaner and keeps `core` engine-neutral, but adds one more file. Leaning `SecurityAdapter`.
+3. Is one job store per node acceptable in MVP, or does the first release need cluster-visible metadata (backed by a system index)?
+
+Feedback welcome on any of the above.
diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java
index ad6bdcb7c9c..97f4c390b74 100644
--- a/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java
+++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java
@@ -36,6 +36,8 @@
/** Setting implementation on OpenSearch. */
@Log4j2
public class OpenSearchSettings extends Settings {
+ private static final TimeValue MAX_PPL_ASYNC_KEEP_ALIVE = TimeValue.timeValueHours(24);
+
/** Default settings. */
private final Map> defaultSettings;
@@ -86,6 +88,42 @@ public class OpenSearchSettings extends Settings {
Setting.Property.NodeScope,
Setting.Property.Dynamic);
+ /** Maximum number of asynchronous PPL queries running concurrently on one node. */
+ public static final Setting PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING =
+ Setting.intSetting(
+ Key.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES.getKeyValue(),
+ 20,
+ 1,
+ Setting.Property.NodeScope,
+ Setting.Property.Dynamic);
+
+ /** Maximum number of asynchronous PPL jobs retained on one node. */
+ public static final Setting PPL_ASYNC_MAX_RETAINED_JOBS_SETTING =
+ Setting.intSetting(
+ Key.PPL_ASYNC_MAX_RETAINED_JOBS.getKeyValue(),
+ 100,
+ 1,
+ Setting.Property.NodeScope,
+ Setting.Property.Dynamic);
+
+ /** Maximum accepted submit wait-for-completion timeout. */
+ public static final Setting PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING =
+ Setting.positiveTimeSetting(
+ Key.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT.getKeyValue(),
+ TimeValue.timeValueSeconds(60),
+ Setting.Property.NodeScope,
+ Setting.Property.Dynamic);
+
+ /** Configurable asynchronous PPL job lease limit, capped at 24 hours. */
+ public static final Setting PPL_ASYNC_MAX_KEEP_ALIVE_SETTING =
+ Setting.timeSetting(
+ Key.PPL_ASYNC_MAX_KEEP_ALIVE.getKeyValue(),
+ MAX_PPL_ASYNC_KEEP_ALIVE,
+ TimeValue.ZERO,
+ MAX_PPL_ASYNC_KEEP_ALIVE,
+ Setting.Property.NodeScope,
+ Setting.Property.Dynamic);
+
public static final Setting PPL_SYNTAX_LEGACY_PREFERRED_SETTING =
Setting.boolSetting(
Key.PPL_SYNTAX_LEGACY_PREFERRED.getKeyValue(),
@@ -444,6 +482,30 @@ public OpenSearchSettings(ClusterSettings clusterSettings) {
Key.PPL_QUERY_TIMEOUT,
PPL_QUERY_TIMEOUT_SETTING,
new Updater(Key.PPL_QUERY_TIMEOUT));
+ register(
+ settingBuilder,
+ clusterSettings,
+ Key.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES,
+ PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING,
+ new Updater(Key.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES));
+ register(
+ settingBuilder,
+ clusterSettings,
+ Key.PPL_ASYNC_MAX_RETAINED_JOBS,
+ PPL_ASYNC_MAX_RETAINED_JOBS_SETTING,
+ new Updater(Key.PPL_ASYNC_MAX_RETAINED_JOBS));
+ register(
+ settingBuilder,
+ clusterSettings,
+ Key.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT,
+ PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING,
+ new Updater(Key.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT));
+ register(
+ settingBuilder,
+ clusterSettings,
+ Key.PPL_ASYNC_MAX_KEEP_ALIVE,
+ PPL_ASYNC_MAX_KEEP_ALIVE_SETTING,
+ new Updater(Key.PPL_ASYNC_MAX_KEEP_ALIVE));
register(
settingBuilder,
clusterSettings,
@@ -767,6 +829,10 @@ public static List> pluginSettings() {
.add(DESERIALIZATION_MAX_BYTES_SETTING)
.add(PPL_ENABLED_SETTING)
.add(PPL_QUERY_TIMEOUT_SETTING)
+ .add(PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING)
+ .add(PPL_ASYNC_MAX_RETAINED_JOBS_SETTING)
+ .add(PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING)
+ .add(PPL_ASYNC_MAX_KEEP_ALIVE_SETTING)
.add(PPL_SYNTAX_LEGACY_PREFERRED_SETTING)
.add(CALCITE_ENGINE_ENABLED_SETTING)
.add(CALCITE_FALLBACK_ALLOWED_SETTING)
diff --git a/opensearch/src/test/java/org/opensearch/sql/opensearch/setting/OpenSearchSettingsTest.java b/opensearch/src/test/java/org/opensearch/sql/opensearch/setting/OpenSearchSettingsTest.java
index 63678051d3d..7c9e3cb3f9e 100644
--- a/opensearch/src/test/java/org/opensearch/sql/opensearch/setting/OpenSearchSettingsTest.java
+++ b/opensearch/src/test/java/org/opensearch/sql/opensearch/setting/OpenSearchSettingsTest.java
@@ -9,6 +9,7 @@
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.AdditionalMatchers.not;
import static org.mockito.AdditionalMatchers.or;
@@ -28,6 +29,7 @@
import org.opensearch.cluster.ClusterName;
import org.opensearch.common.settings.ClusterSettings;
import org.opensearch.common.settings.Setting;
+import org.opensearch.common.unit.TimeValue;
import org.opensearch.core.common.unit.ByteSizeValue;
import org.opensearch.sql.common.setting.Settings;
import org.opensearch.sql.utils.DeserializationFilterUtil;
@@ -168,6 +170,50 @@ void deserializationStructuralLimitSettingsAreDynamicAndRegistered() {
assertTrue(pluginSettings.contains(OpenSearchSettings.DESERIALIZATION_MAX_BYTES_SETTING));
}
+ @Test
+ void pplAsyncSettingsHaveExpectedDefaultsAndAreDynamic() {
+ org.opensearch.common.settings.Settings empty = org.opensearch.common.settings.Settings.EMPTY;
+
+ assertEquals(
+ 20, OpenSearchSettings.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING.get(empty));
+ assertEquals(100, OpenSearchSettings.PPL_ASYNC_MAX_RETAINED_JOBS_SETTING.get(empty));
+ assertEquals(
+ TimeValue.timeValueSeconds(60),
+ OpenSearchSettings.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING.get(empty));
+ assertEquals(
+ TimeValue.timeValueHours(24),
+ OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING.get(empty));
+
+ List> pluginSettings = OpenSearchSettings.pluginSettings();
+ assertTrue(
+ pluginSettings.contains(
+ OpenSearchSettings.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING));
+ assertTrue(pluginSettings.contains(OpenSearchSettings.PPL_ASYNC_MAX_RETAINED_JOBS_SETTING));
+ assertTrue(
+ pluginSettings.contains(
+ OpenSearchSettings.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING));
+ assertTrue(pluginSettings.contains(OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING));
+ assertTrue(OpenSearchSettings.PPL_ASYNC_NODE_CONCURRENT_RUNNING_QUERIES_SETTING.isDynamic());
+ assertTrue(OpenSearchSettings.PPL_ASYNC_MAX_RETAINED_JOBS_SETTING.isDynamic());
+ assertTrue(OpenSearchSettings.PPL_ASYNC_MAX_WAIT_FOR_COMPLETION_TIMEOUT_SETTING.isDynamic());
+ assertTrue(OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING.isDynamic());
+ }
+
+ @Test
+ void pplAsyncMaxKeepAliveIsCappedAt24Hours() {
+ String key = Settings.Key.PPL_ASYNC_MAX_KEEP_ALIVE.getKeyValue();
+
+ assertEquals(
+ TimeValue.timeValueHours(24),
+ OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING.get(
+ org.opensearch.common.settings.Settings.builder().put(key, "24h").build()));
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ OpenSearchSettings.PPL_ASYNC_MAX_KEEP_ALIVE_SETTING.get(
+ org.opensearch.common.settings.Settings.builder().put(key, "25h").build()));
+ }
+
@Test
void getSparkExecutionEngineConfigSetting() {
// Default is empty string
diff --git a/plugin/build.gradle b/plugin/build.gradle
index d0c825424be..64ac97f561a 100644
--- a/plugin/build.gradle
+++ b/plugin/build.gradle
@@ -168,6 +168,7 @@ dependencies {
api project(':datasources')
api project(':async-query')
api project(':direct-query')
+ implementation "org.opensearch:common-utils:${opensearch_build}"
testImplementation group: 'net.bytebuddy', name: 'byte-buddy-agent', version: '1.15.11'
testImplementation group: 'org.hamcrest', name: 'hamcrest-library', version: "${hamcrest_version}"
diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/PPLQueryErrorHandler.java b/plugin/src/main/java/org/opensearch/sql/plugin/PPLQueryErrorHandler.java
new file mode 100644
index 00000000000..9d690dd3e67
--- /dev/null
+++ b/plugin/src/main/java/org/opensearch/sql/plugin/PPLQueryErrorHandler.java
@@ -0,0 +1,67 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.OpenSearchException;
+import org.opensearch.core.rest.RestStatus;
+import org.opensearch.index.IndexNotFoundException;
+import org.opensearch.sql.common.antlr.SyntaxCheckException;
+import org.opensearch.sql.common.error.ErrorReport;
+import org.opensearch.sql.datasources.exceptions.DataSourceClientException;
+import org.opensearch.sql.exception.QueryEngineException;
+import org.opensearch.sql.legacy.metrics.MetricName;
+import org.opensearch.sql.legacy.metrics.Metrics;
+
+/** Classifies PPL failures and records the corresponding customer or system error metric. */
+public final class PPLQueryErrorHandler {
+ private static final Logger LOG = LogManager.getLogger(PPLQueryErrorHandler.class);
+
+ private PPLQueryErrorHandler() {}
+
+ /**
+ * Records a PPL failure and returns the HTTP status associated with it.
+ *
+ * @param exception query failure
+ * @return client or system error status
+ */
+ public static RestStatus recordFailure(Exception exception) {
+ int code = rawStatusCode(exception);
+ if (400 <= code && code < 500) {
+ increment(MetricName.PPL_FAILED_REQ_COUNT_CUS);
+ } else if (500 <= code && code < 600) {
+ increment(MetricName.PPL_FAILED_REQ_COUNT_SYS);
+ } else {
+ LOG.warn(
+ "Got an exception returning non-error status {}", RestStatus.fromCode(code), exception);
+ }
+ return RestStatus.fromCode(code);
+ }
+
+ private static int rawStatusCode(Exception exception) {
+ if (exception instanceof ErrorReport errorReport) {
+ return rawStatusCode(errorReport.getCause());
+ }
+ if (exception instanceof OpenSearchException openSearchException) {
+ return openSearchException.status().getStatus();
+ }
+ return isClientError(exception) ? 400 : 500;
+ }
+
+ private static boolean isClientError(Exception exception) {
+ return exception instanceof IllegalArgumentException
+ || exception instanceof IndexNotFoundException
+ || exception instanceof QueryEngineException
+ || exception instanceof SyntaxCheckException
+ || exception instanceof DataSourceClientException
+ || exception instanceof IllegalAccessException;
+ }
+
+ private static void increment(MetricName metricName) {
+ Metrics.getInstance().getNumericalMetric(metricName).increment();
+ }
+}
diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJob.java b/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJob.java
new file mode 100644
index 00000000000..f105047ebe0
--- /dev/null
+++ b/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJob.java
@@ -0,0 +1,983 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin.transport.asyncquery;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletionException;
+import java.util.function.Function;
+import java.util.function.LongSupplier;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.ResourceNotFoundException;
+import org.opensearch.cluster.node.DiscoveryNode;
+import org.opensearch.common.lease.Releasable;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.action.ActionListener;
+import org.opensearch.core.tasks.TaskId;
+import org.opensearch.sql.executor.AsyncQueryExecution;
+import org.opensearch.sql.executor.ExecutionEngine.QueryResponse;
+import org.opensearch.sql.executor.ExecutionEngine.Schema;
+import org.opensearch.sql.plugin.PPLQueryErrorHandler;
+import org.opensearch.sql.plugin.transport.PPLQueryAction;
+import org.opensearch.sql.plugin.transport.PPLQueryTask;
+import org.opensearch.sql.plugin.transport.TransportPPLQueryRequest;
+import org.opensearch.tasks.CancellableTask;
+import org.opensearch.tasks.Task;
+import org.opensearch.tasks.TaskManager;
+import org.opensearch.threadpool.Scheduler.ScheduledCancellable;
+import org.opensearch.threadpool.ThreadPool;
+
+/**
+ * Active object representing one asynchronous PPL query on the owner node.
+ *
+ * Each job owns its execution handle, cancellable task, retention deadline, and lease timer. It
+ * publishes itself to a {@link QueryJobRegistry} at creation and removes itself when it becomes
+ * unreachable through DELETE, expiry, direct completion, or shutdown. All state transitions are
+ * synchronized on the job; every observable side effect (registry mutation, listener callback, task
+ * cancellation, execution close) happens after the lock has been released.
+ *
+ *
The state machine tracks two axes: retention and completion.
+ *
+ *
+ * Current state Event Next state Response
+ * RUNNING direct success REMOVED final snapshot without ID
+ * RUNNING direct failure REMOVED failure without ID
+ * RUNNING retain deadline RETAINED_RUNNING running snapshot with ID
+ * RETAINED_RUNNING success RETAINED_SUCCEEDED none (final result kept)
+ * RETAINED_RUNNING failure RETAINED_FAILED none (failure kept)
+ * * DELETE REMOVED status observed at DELETE
+ * RETAINED_* expire / shutdown REMOVED none
+ *
+ */
+public final class QueryJob {
+ private static final Logger LOG = LogManager.getLogger(QueryJob.class);
+
+ private static final Cancellable NO_TIMEOUT = () -> {};
+
+ /** Lifecycle state exposed in asynchronous PPL responses. */
+ public enum Status {
+ /** Query execution is still running. */
+ RUNNING,
+
+ /** Query execution completed successfully. */
+ SUCCEEDED,
+
+ /** Query execution failed. */
+ FAILED,
+
+ /** Query execution was cancelled by DELETE. */
+ CANCELLED
+ }
+
+ /**
+ * Immutable point-in-time response for a job.
+ *
+ * Each variant exposes only the data valid for its lifecycle state.
+ */
+ public sealed interface Snapshot {
+ /**
+ * Returns the retained job ID.
+ *
+ * @return job ID, or empty for a final response returned directly by POST
+ */
+ Optional id();
+
+ /**
+ * Returns the public lifecycle state.
+ *
+ * @return response status
+ */
+ Status status();
+
+ /**
+ * Snapshot of a query that is still running.
+ *
+ * @param jobId retained job ID
+ * @param response current result, or empty before a result is available
+ */
+ record Running(QueryJobId jobId, Optional response) implements Snapshot {
+ @Override
+ public Optional id() {
+ return Optional.of(jobId);
+ }
+
+ @Override
+ public Status status() {
+ return Status.RUNNING;
+ }
+ }
+
+ /**
+ * Snapshot of a successfully completed query.
+ *
+ * @param id retained job ID, or empty for a direct POST response
+ * @param response final query result
+ * @param tookMillis elapsed execution time
+ */
+ record Succeeded(Optional id, QueryResponse response, long tookMillis)
+ implements Snapshot {
+ @Override
+ public Status status() {
+ return Status.SUCCEEDED;
+ }
+ }
+
+ /**
+ * Snapshot of a failed query.
+ *
+ * @param id retained job ID, or empty for a direct POST response
+ * @param failure client-visible failure
+ * @param tookMillis elapsed execution time
+ */
+ record Failed(Optional id, Failure failure, long tookMillis) implements Snapshot {
+ @Override
+ public Status status() {
+ return Status.FAILED;
+ }
+ }
+ }
+
+ /**
+ * Client-visible failure retained with a failed job.
+ *
+ * @param type exception type
+ * @param reason client-facing failure reason
+ */
+ public record Failure(String type, String reason) {
+ /**
+ * Derives a client-visible failure from an exception.
+ *
+ * @param exception exception raised by execution
+ * @return failure with the exception's simple type and message
+ */
+ public static Failure from(Exception exception) {
+ String type =
+ exception.getClass().getSimpleName().isBlank()
+ ? exception.getClass().getName()
+ : exception.getClass().getSimpleName();
+ String reason =
+ exception.getMessage() == null || exception.getMessage().isBlank()
+ ? "query execution failed"
+ : exception.getMessage();
+ return new Failure(type, reason);
+ }
+ }
+
+ /** Cancels a previously scheduled timer. Multiple cancellations are safe. */
+ @FunctionalInterface
+ public interface Cancellable {
+ /** Cancels the timer if it has not already fired. */
+ void cancel();
+ }
+
+ /** Schedules a one-shot runnable. Production wraps {@link ThreadPool}; tests use a fake. */
+ @FunctionalInterface
+ public interface Scheduler {
+ /**
+ * Schedules {@code task} to run once after {@code delayMillis} milliseconds.
+ *
+ * @param delayMillis delay before firing
+ * @param task action to run
+ * @return handle used to cancel the scheduled task
+ */
+ Cancellable schedule(long delayMillis, Runnable task);
+ }
+
+ /** Owned cancellable task and its registration cleanup. Package-private for tests. */
+ record JobTask(CancellableTask task, Runnable release) {
+ /** Releases task-manager registrations owned by this wrapper. Safe to call once. */
+ void close() {
+ release.run();
+ }
+ }
+
+ /** Internal lifecycle; unlike {@link Status}, this includes retention and removal. */
+ private enum State {
+ RUNNING,
+ RETAINED_RUNNING,
+ RETAINED_SUCCEEDED,
+ RETAINED_FAILED,
+ REMOVED
+ }
+
+ private final QueryJobId id;
+ private final QueryJobRegistry registry;
+ private final LongSupplier clock;
+ private final Scheduler scheduler;
+ private final TaskManager taskManager;
+ private final QueryJobOwner owner;
+ private final long startTimeMillis;
+ private final ActionListener responseListener;
+
+ private State state = State.RUNNING;
+ private long keepAliveMillis;
+ private long expirationTimeMillis;
+ private long completionTimeMillis = -1L;
+ private JobTask task;
+ private AsyncQueryExecution execution;
+ private Failure failure;
+ private Cancellable retentionTimer = NO_TIMEOUT;
+ private Cancellable expiryTimer = NO_TIMEOUT;
+
+ private QueryJob(
+ QueryJobId id,
+ QueryJobRegistry registry,
+ LongSupplier clock,
+ Scheduler scheduler,
+ TaskManager taskManager,
+ QueryJobOwner owner,
+ long keepAliveMillis,
+ JobTask task,
+ ActionListener responseListener) {
+ this.id = id;
+ this.registry = registry;
+ this.clock = clock;
+ this.scheduler = scheduler;
+ this.taskManager = taskManager;
+ this.owner = owner;
+ this.startTimeMillis = clock.getAsLong();
+ this.keepAliveMillis = keepAliveMillis;
+ this.expirationTimeMillis = startTimeMillis + keepAliveMillis;
+ this.task = task;
+ this.responseListener = responseListener;
+ }
+
+ /**
+ * Starts an asynchronous PPL query in production wiring.
+ *
+ * {@link TaskManager} registers the job's cancellable task as a child of {@code parentTask} so
+ * cancellation of the POST request cascades. The job schedules its own retention deadline on
+ * {@code threadPool} and calls {@code executionStarter} inline. If execution finishes before the
+ * deadline, {@code responseListener} receives the final result without a job ID. Otherwise the
+ * job becomes retained and the listener receives a running snapshot with the ID.
+ *
+ * @param registry registry that will hold the job while it is discoverable
+ * @param threadPool schedules the retention deadline and keep-alive expiry
+ * @param taskManager registers and cancels the job's own task
+ * @param parentTask POST request task; the job task becomes its child
+ * @param request transport request used to register the job task
+ * @param owner caller retained with the job for later authorization
+ * @param keepAlive lease duration applied when the job is retained
+ * @param waitForCompletion maximum time to wait for a direct result
+ * @param executionStarter starts execution using the job-owned cancellable task
+ * @param responseListener receives either the direct result or the retained running snapshot
+ * @return newly created job, already published to {@code registry}
+ * @throws NullPointerException if any required argument is null
+ */
+ public static QueryJob create(
+ QueryJobRegistry registry,
+ ThreadPool threadPool,
+ TaskManager taskManager,
+ PPLQueryTask parentTask,
+ TransportPPLQueryRequest request,
+ QueryJobOwner owner,
+ TimeValue keepAlive,
+ TimeValue waitForCompletion,
+ Function executionStarter,
+ ActionListener responseListener) {
+ Objects.requireNonNull(threadPool);
+ Objects.requireNonNull(taskManager);
+ JobTask jobTask = registerJobTask(taskManager, parentTask, request);
+ try {
+ return create(
+ taskManager.localNode().getId(),
+ registry,
+ System::currentTimeMillis,
+ wrapScheduler(threadPool),
+ taskManager,
+ jobTask,
+ owner,
+ keepAlive,
+ waitForCompletion,
+ executionStarter,
+ responseListener);
+ } catch (RuntimeException | Error e) {
+ jobTask.close();
+ throw e;
+ }
+ }
+
+ /**
+ * Test entry point that accepts injectable clock, scheduler, and pre-registered task.
+ *
+ * @param localNodeId local node ID used to route the ID back on future requests
+ * @param registry registry that will hold the job while it is discoverable
+ * @param clock time source
+ * @param scheduler timer used for retention deadline and keep-alive expiry
+ * @param taskManager cancels the job task with descendant cascade; may be {@code null} when the
+ * job task should be cancelled directly
+ * @param task pre-registered cancellable task owned by the job
+ * @param owner caller retained with the job for later authorization
+ * @param keepAlive lease duration applied when the job is retained
+ * @param waitForCompletion maximum time to wait for a direct result
+ * @param executionStarter starts execution using the job-owned cancellable task
+ * @param responseListener receives either the direct result or the retained running snapshot
+ * @return newly created job, already published to {@code registry}
+ */
+ static QueryJob create(
+ String localNodeId,
+ QueryJobRegistry registry,
+ LongSupplier clock,
+ Scheduler scheduler,
+ TaskManager taskManager,
+ JobTask task,
+ QueryJobOwner owner,
+ TimeValue keepAlive,
+ TimeValue waitForCompletion,
+ Function executionStarter,
+ ActionListener responseListener) {
+ Objects.requireNonNull(localNodeId);
+ Objects.requireNonNull(registry);
+ Objects.requireNonNull(clock);
+ Objects.requireNonNull(scheduler);
+ Objects.requireNonNull(task);
+ Objects.requireNonNull(owner);
+ Objects.requireNonNull(keepAlive);
+ Objects.requireNonNull(waitForCompletion);
+ Objects.requireNonNull(executionStarter);
+ Objects.requireNonNull(responseListener);
+
+ QueryJob job =
+ publish(
+ localNodeId,
+ registry,
+ clock,
+ scheduler,
+ taskManager,
+ owner,
+ keepAlive.millis(),
+ task,
+ responseListener);
+ try {
+ job.scheduleRetention(waitForCompletion);
+ } catch (RuntimeException | Error e) {
+ job.discard("PPL asynchronous query startup failed");
+ throw e;
+ }
+ job.startExecution(executionStarter);
+ return job;
+ }
+
+ private static QueryJob publish(
+ String localNodeId,
+ QueryJobRegistry registry,
+ LongSupplier clock,
+ Scheduler scheduler,
+ TaskManager taskManager,
+ QueryJobOwner owner,
+ long keepAliveMillis,
+ JobTask task,
+ ActionListener responseListener) {
+ while (true) {
+ QueryJobId id = QueryJobId.create(localNodeId);
+ QueryJob job =
+ new QueryJob(
+ id,
+ registry,
+ clock,
+ scheduler,
+ taskManager,
+ owner,
+ keepAliveMillis,
+ task,
+ responseListener);
+ if (registry.add(job) == null) {
+ return job;
+ }
+ }
+ }
+
+ /**
+ * Returns the opaque ID assigned to this job.
+ *
+ * @return job ID used as the registry key
+ */
+ public QueryJobId getJobId() {
+ return id;
+ }
+
+ /**
+ * Returns the immutable owner identity captured when this job was created.
+ *
+ * @return job owner used for GET and DELETE authorization
+ */
+ public QueryJobOwner getOwner() {
+ return owner;
+ }
+
+ /**
+ * Returns the current snapshot of the job.
+ *
+ * A non-null {@code keepAlive} starts a new lease from the current time. A call at or after
+ * the current expiration removes the job and reports it as not found.
+ *
+ * @param keepAlive new lease duration, or {@code null} to leave the current lease unchanged
+ * @return immutable snapshot of the current state
+ * @throws ResourceNotFoundException if the job has been removed or has just expired
+ */
+ public Snapshot get(TimeValue keepAlive) {
+ long now = clock.getAsLong();
+ AsyncQueryExecution executionForResult;
+ State stateForResult;
+ long tookMillis;
+ Failure failureForResult;
+ boolean expired = false;
+ Cancellable retentionToCancel = NO_TIMEOUT;
+ Cancellable expiryToCancel = NO_TIMEOUT;
+ Cancellable rescheduledExpiry = NO_TIMEOUT;
+ JobTask taskToClose = null;
+ AsyncQueryExecution executionToClose = null;
+
+ synchronized (this) {
+ ensurePresent();
+ if (state != State.RUNNING && now >= expirationTimeMillis) {
+ expired = true;
+ retentionToCancel = retentionTimer;
+ expiryToCancel = expiryTimer;
+ retentionTimer = NO_TIMEOUT;
+ expiryTimer = NO_TIMEOUT;
+ taskToClose = detachTask();
+ executionToClose = detachExecution();
+ state = State.REMOVED;
+ executionForResult = null;
+ stateForResult = null;
+ tookMillis = 0L;
+ failureForResult = null;
+ } else {
+ if (keepAlive != null && state != State.RUNNING) {
+ keepAliveMillis = keepAlive.millis();
+ expirationTimeMillis = now + keepAliveMillis;
+ expiryToCancel = expiryTimer;
+ expiryTimer = NO_TIMEOUT;
+ rescheduledExpiry = scheduler.schedule(keepAliveMillis, this::expire);
+ expiryTimer = rescheduledExpiry;
+ }
+ executionForResult = execution;
+ stateForResult = state;
+ tookMillis =
+ completionTimeMillis < 0 ? -1L : Math.max(0L, completionTimeMillis - startTimeMillis);
+ failureForResult = failure;
+ }
+ }
+
+ if (expired) {
+ retentionToCancel.cancel();
+ expiryToCancel.cancel();
+ registry.remove(id, this);
+ cancelTaskAsync(taskToClose, "PPL asynchronous query expired");
+ closeExecution(executionToClose);
+ throw notFound();
+ }
+
+ expiryToCancel.cancel();
+ return buildSnapshot(stateForResult, executionForResult, failureForResult, tookMillis, true);
+ }
+
+ /**
+ * Cancels this job and returns the status observed at cancellation.
+ *
+ * @param reason human-readable reason recorded on the underlying task
+ * @return the {@link Status} observed when the job was removed
+ * @throws ResourceNotFoundException if the job has been removed or has just expired
+ */
+ public Status cancel(String reason) {
+ Objects.requireNonNull(reason);
+ Status observed;
+ Cancellable retentionToCancel;
+ Cancellable expiryToCancel;
+ JobTask taskToClose;
+ AsyncQueryExecution executionToClose;
+ String cancelReason;
+ boolean cancelRunning;
+
+ synchronized (this) {
+ ensurePresent();
+ if (state != State.RUNNING && clock.getAsLong() >= expirationTimeMillis) {
+ // Late DELETE races expiry; the job is gone.
+ retentionToCancel = retentionTimer;
+ expiryToCancel = expiryTimer;
+ retentionTimer = NO_TIMEOUT;
+ expiryTimer = NO_TIMEOUT;
+ taskToClose = detachTask();
+ executionToClose = detachExecution();
+ state = State.REMOVED;
+ cancelReason = "PPL asynchronous query expired";
+ cancelRunning = true;
+ observed = null;
+ } else {
+ observed =
+ switch (state) {
+ case RUNNING, RETAINED_RUNNING -> Status.CANCELLED;
+ case RETAINED_SUCCEEDED -> Status.SUCCEEDED;
+ case RETAINED_FAILED -> Status.FAILED;
+ case REMOVED -> throw new IllegalStateException("PPL asynchronous query was removed");
+ };
+ cancelRunning = (state == State.RUNNING || state == State.RETAINED_RUNNING);
+ retentionToCancel = retentionTimer;
+ expiryToCancel = expiryTimer;
+ retentionTimer = NO_TIMEOUT;
+ expiryTimer = NO_TIMEOUT;
+ taskToClose = detachTask();
+ executionToClose = detachExecution();
+ state = State.REMOVED;
+ cancelReason = reason;
+ }
+ }
+
+ retentionToCancel.cancel();
+ expiryToCancel.cancel();
+ registry.remove(id, this);
+ if (cancelRunning) {
+ cancelTaskAsync(taskToClose, cancelReason);
+ } else {
+ closeTask(taskToClose);
+ }
+ closeExecution(executionToClose);
+
+ if (observed == null) {
+ throw notFound();
+ }
+ return observed;
+ }
+
+ /**
+ * Removes the job as part of an internal cleanup (shutdown or startup abort). Idempotent.
+ *
+ * @param reason reason recorded on the underlying task
+ */
+ void discard(String reason) {
+ Objects.requireNonNull(reason);
+ Cancellable retentionToCancel;
+ Cancellable expiryToCancel;
+ JobTask taskToClose;
+ AsyncQueryExecution executionToClose;
+ boolean wasExecuting;
+
+ synchronized (this) {
+ if (state == State.REMOVED) {
+ return;
+ }
+ wasExecuting = (state == State.RUNNING || state == State.RETAINED_RUNNING);
+ retentionToCancel = retentionTimer;
+ expiryToCancel = expiryTimer;
+ retentionTimer = NO_TIMEOUT;
+ expiryTimer = NO_TIMEOUT;
+ taskToClose = detachTask();
+ executionToClose = detachExecution();
+ state = State.REMOVED;
+ }
+
+ retentionToCancel.cancel();
+ expiryToCancel.cancel();
+ registry.remove(id, this);
+ if (wasExecuting) {
+ cancelTaskAsync(taskToClose, reason);
+ } else {
+ closeTask(taskToClose);
+ }
+ closeExecution(executionToClose);
+ }
+
+ // ---------- Retention deadline ----------
+
+ private void scheduleRetention(TimeValue waitForCompletion) {
+ if (waitForCompletion.millis() == 0) {
+ onRetentionDeadline();
+ return;
+ }
+ Cancellable timer = scheduler.schedule(waitForCompletion.millis(), this::onRetentionDeadline);
+ synchronized (this) {
+ if (state == State.RUNNING) {
+ retentionTimer = timer;
+ return;
+ }
+ }
+ // Retention state already changed while the timer was being wired up. Cancel it.
+ timer.cancel();
+ }
+
+ private void onRetentionDeadline() {
+ Snapshot snapshotToDeliver;
+ Cancellable rescheduledExpiry;
+
+ synchronized (this) {
+ if (state != State.RUNNING) {
+ return;
+ }
+ state = State.RETAINED_RUNNING;
+ long now = clock.getAsLong();
+ expirationTimeMillis = now + keepAliveMillis;
+ snapshotToDeliver = new Snapshot.Running(id, currentResultCopy());
+ retentionTimer = NO_TIMEOUT;
+ rescheduledExpiry = scheduler.schedule(keepAliveMillis, this::expire);
+ expiryTimer = rescheduledExpiry;
+ }
+
+ try {
+ responseListener.onResponse(snapshotToDeliver);
+ } catch (RuntimeException e) {
+ // Cannot deliver retained ID; discard the job so no state leaks.
+ discard("PPL asynchronous query startup failed");
+ safeNotifyFailure(e);
+ }
+ }
+
+ // ---------- Execution attachment and completion ----------
+
+ private void startExecution(Function executionStarter) {
+ AsyncQueryExecution started;
+ try {
+ started = Objects.requireNonNull(executionStarter.apply(task().task()));
+ } catch (RuntimeException e) {
+ fail(e);
+ return;
+ }
+ if (!tryAttachExecution(started)) {
+ closeExecution(started);
+ return;
+ }
+ started
+ .completion()
+ .whenComplete(
+ (ignored, err) -> {
+ if (err == null) {
+ complete();
+ } else {
+ fail(asException(err));
+ }
+ });
+ }
+
+ private synchronized JobTask task() {
+ return task;
+ }
+
+ private synchronized boolean tryAttachExecution(AsyncQueryExecution incoming) {
+ if (state != State.RUNNING && state != State.RETAINED_RUNNING) {
+ return false;
+ }
+ if (execution != null) {
+ return false;
+ }
+ execution = incoming;
+ return true;
+ }
+
+ private void complete() {
+ long now = clock.getAsLong();
+ boolean directResponse;
+ JobTask taskToClose;
+ AsyncQueryExecution executionForSnapshot;
+ Cancellable retentionToCancel;
+
+ synchronized (this) {
+ if (state != State.RUNNING && state != State.RETAINED_RUNNING) {
+ return;
+ }
+ if (execution == null) {
+ throw new IllegalStateException(
+ "PPL asynchronous execution must be attached before successful completion");
+ }
+ completionTimeMillis = now;
+ directResponse = (state == State.RUNNING);
+ retentionToCancel = retentionTimer;
+ retentionTimer = NO_TIMEOUT;
+ taskToClose = detachTask();
+ if (directResponse) {
+ executionForSnapshot = detachExecution();
+ state = State.REMOVED;
+ } else {
+ executionForSnapshot = null;
+ state = State.RETAINED_SUCCEEDED;
+ }
+ }
+
+ retentionToCancel.cancel();
+ if (directResponse) {
+ registry.remove(id, this);
+ Snapshot snapshot;
+ try {
+ QueryResponse result =
+ executionForSnapshot
+ .currentResult()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Successful PPL asynchronous execution completed without a final"
+ + " result"));
+ snapshot =
+ new Snapshot.Succeeded(Optional.empty(), defensiveCopy(result), now - startTimeMillis);
+ } catch (RuntimeException e) {
+ safeNotifyFailure(e);
+ closeTask(taskToClose);
+ closeExecution(executionForSnapshot);
+ return;
+ }
+ try {
+ responseListener.onResponse(snapshot);
+ } catch (RuntimeException e) {
+ safeNotifyFailure(e);
+ } finally {
+ closeTask(taskToClose);
+ closeExecution(executionForSnapshot);
+ }
+ } else {
+ closeTask(taskToClose);
+ }
+ }
+
+ private void fail(Exception cause) {
+ Objects.requireNonNull(cause);
+ long now = clock.getAsLong();
+ boolean directResponse;
+ JobTask taskToClose;
+ AsyncQueryExecution executionToClose;
+ Cancellable retentionToCancel;
+ Failure captured = Failure.from(cause);
+
+ synchronized (this) {
+ if (state != State.RUNNING && state != State.RETAINED_RUNNING) {
+ return;
+ }
+ completionTimeMillis = now;
+ failure = captured;
+ directResponse = (state == State.RUNNING);
+ retentionToCancel = retentionTimer;
+ retentionTimer = NO_TIMEOUT;
+ taskToClose = detachTask();
+ executionToClose = detachExecution();
+ state = directResponse ? State.REMOVED : State.RETAINED_FAILED;
+ }
+
+ retentionToCancel.cancel();
+ if (directResponse) {
+ registry.remove(id, this);
+ try {
+ responseListener.onFailure(cause);
+ } catch (RuntimeException e) {
+ LOG.warn("PPL asynchronous listener rejected failure ({})", e.getClass().getSimpleName());
+ } finally {
+ closeTask(taskToClose);
+ closeExecution(executionToClose);
+ }
+ } else {
+ PPLQueryErrorHandler.recordFailure(cause);
+ closeTask(taskToClose);
+ closeExecution(executionToClose);
+ }
+ }
+
+ private void expire() {
+ long now = clock.getAsLong();
+ Cancellable retentionToCancel;
+ JobTask taskToClose;
+ AsyncQueryExecution executionToClose;
+
+ synchronized (this) {
+ if (state == State.REMOVED || state == State.RUNNING || now < expirationTimeMillis) {
+ return;
+ }
+ retentionToCancel = retentionTimer;
+ retentionTimer = NO_TIMEOUT;
+ expiryTimer = NO_TIMEOUT;
+ taskToClose = detachTask();
+ executionToClose = detachExecution();
+ state = State.REMOVED;
+ }
+
+ retentionToCancel.cancel();
+ registry.remove(id, this);
+ cancelTaskAsync(taskToClose, "PPL asynchronous query expired");
+ closeExecution(executionToClose);
+ }
+
+ // ---------- Helpers ----------
+
+ private void ensurePresent() {
+ if (state == State.REMOVED) {
+ throw notFound();
+ }
+ }
+
+ private JobTask detachTask() {
+ JobTask detached = task;
+ task = null;
+ return detached;
+ }
+
+ private AsyncQueryExecution detachExecution() {
+ AsyncQueryExecution detached = execution;
+ execution = null;
+ return detached;
+ }
+
+ private Optional currentResultCopy() {
+ if (execution == null) {
+ return Optional.empty();
+ }
+ return execution.currentResult().map(QueryJob::defensiveCopy);
+ }
+
+ private Snapshot buildSnapshot(
+ State from,
+ AsyncQueryExecution execForResult,
+ Failure failureForResult,
+ long tookMillis,
+ boolean withId) {
+ Optional snapshotId = withId ? Optional.of(id) : Optional.empty();
+ return switch (from) {
+ case RUNNING, RETAINED_RUNNING -> {
+ Optional current =
+ execForResult == null
+ ? Optional.empty()
+ : execForResult.currentResult().map(QueryJob::defensiveCopy);
+ yield new Snapshot.Running(id, current);
+ }
+ case RETAINED_SUCCEEDED -> {
+ QueryResponse result =
+ Objects.requireNonNull(execForResult, "attached execution required")
+ .currentResult()
+ .map(QueryJob::defensiveCopy)
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Successful PPL asynchronous execution completed without a final"
+ + " result"));
+ yield new Snapshot.Succeeded(snapshotId, result, tookMillis);
+ }
+ case RETAINED_FAILED ->
+ new Snapshot.Failed(snapshotId, Objects.requireNonNull(failureForResult), tookMillis);
+ case REMOVED -> throw new IllegalStateException("PPL asynchronous query was removed");
+ };
+ }
+
+ private void safeNotifyFailure(Exception cause) {
+ try {
+ responseListener.onFailure(cause);
+ } catch (RuntimeException e) {
+ LOG.warn("PPL asynchronous listener rejected failure ({})", e.getClass().getSimpleName());
+ }
+ }
+
+ private static QueryResponse defensiveCopy(QueryResponse response) {
+ Schema schema = new Schema(List.copyOf(response.getSchema().getColumns()));
+ QueryResponse copy =
+ new QueryResponse(schema, List.copyOf(response.getResults()), response.getCursor());
+ copy.setWarnings(List.copyOf(response.getWarnings()));
+ return copy;
+ }
+
+ private static void closeTask(JobTask task) {
+ if (task != null) {
+ task.close();
+ }
+ }
+
+ private void cancelTaskAsync(JobTask jobTask, String reason) {
+ if (jobTask == null) {
+ return;
+ }
+ CancellableTask cancellable = jobTask.task();
+ if (cancellable == null || cancellable.isCancelled()) {
+ jobTask.close();
+ return;
+ }
+ try {
+ if (taskManager == null) {
+ cancellable.cancel(reason);
+ jobTask.close();
+ } else {
+ taskManager.cancelTaskAndDescendants(
+ cancellable,
+ reason,
+ false,
+ ActionListener.wrap(
+ ignored -> jobTask.close(),
+ error -> {
+ jobTask.close();
+ LOG.warn(
+ "Failed to cancel descendants of PPL asynchronous query task ({})",
+ error.getClass().getSimpleName());
+ }));
+ }
+ } catch (RuntimeException e) {
+ jobTask.close();
+ LOG.warn("Failed to cancel PPL asynchronous query task ({})", e.getClass().getSimpleName());
+ }
+ }
+
+ private static void closeExecution(AsyncQueryExecution execution) {
+ if (execution == null) {
+ return;
+ }
+ try {
+ execution.close();
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Failed to close PPL asynchronous query execution ({})", e.getClass().getSimpleName());
+ }
+ }
+
+ private static Exception asException(Throwable failure) {
+ Throwable cause =
+ failure instanceof CompletionException && failure.getCause() != null
+ ? failure.getCause()
+ : failure;
+ return cause instanceof Exception exception ? exception : new RuntimeException(cause);
+ }
+
+ private static ResourceNotFoundException notFound() {
+ return new ResourceNotFoundException("PPL asynchronous query not found");
+ }
+
+ private static Scheduler wrapScheduler(ThreadPool threadPool) {
+ return (delayMillis, task) -> {
+ ScheduledCancellable cancellable =
+ threadPool.schedule(
+ task, TimeValue.timeValueMillis(delayMillis), ThreadPool.Names.GENERIC);
+ return cancellable::cancel;
+ };
+ }
+
+ private static JobTask registerJobTask(
+ TaskManager taskManager, PPLQueryTask parentTask, TransportPPLQueryRequest request) {
+ Objects.requireNonNull(parentTask, "PPL asynchronous query request task is not initialized");
+ DiscoveryNode localNode =
+ Objects.requireNonNull(taskManager.localNode(), "Local node is not initialized");
+
+ // Registered directly rather than through TransportAction.execute(); reproduce the two child-
+ // task bookkeeping steps TransportAction would perform. childNode registration lets parent
+ // cancellation ban this node; parentTaskId lets the ban find and cancel the retained task.
+ Releasable childNodeRegistration = taskManager.registerChildNode(parentTask.getId(), localNode);
+ TaskId originalParent = request.getParentTask();
+ boolean registered = false;
+ try {
+ request.setParentTask(localNode.getId(), parentTask.getId());
+ Task task = taskManager.register("transport", PPLQueryAction.NAME, request);
+ if (!(task instanceof PPLQueryTask pplQueryTask)) {
+ taskManager.unregister(task);
+ throw new IllegalStateException("Failed to create PPL asynchronous query task");
+ }
+ registered = true;
+ return new JobTask(
+ pplQueryTask,
+ () -> {
+ try {
+ taskManager.unregister(pplQueryTask);
+ } finally {
+ childNodeRegistration.close();
+ }
+ });
+ } finally {
+ request.setParentTask(originalParent);
+ if (!registered) {
+ childNodeRegistration.close();
+ }
+ }
+ }
+}
diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobId.java b/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobId.java
new file mode 100644
index 00000000000..05e4c97c7cd
--- /dev/null
+++ b/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobId.java
@@ -0,0 +1,146 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin.transport.asyncquery;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.UUID;
+
+/**
+ * Opaque, URL-safe identifier for an asynchronous PPL job.
+ *
+ * The encoded value contains a format version, the owner node ID used for request routing, and a
+ * random per-job context ID. The complete encoded value is the key in the owner node's job
+ * registry. It contains no query text, user identity, or result data, and clients must treat it as
+ * opaque.
+ *
+ *
Before Base64 URL encoding, the binary layout is:
+ *
+ *
+ * int formatVersion
+ * int ownerNodeIdLength + UTF-8 ownerNodeId
+ * int contextIdLength + UTF-8 contextId
+ *
+ *
+ * The ID provides routing, not authorization. GET and DELETE still authorize the current caller
+ * against the owner stored in the job.
+ *
+ * @param ownerNodeId node that owns the in-memory job
+ * @param contextId random identifier for one job on the owner node
+ */
+public record QueryJobId(String ownerNodeId, String contextId) {
+ private static final int FORMAT_VERSION = 1;
+ private static final int MAX_ENCODED_LENGTH = 2_048;
+ private static final int MAX_OWNER_NODE_ID_BYTES = 1_024;
+ private static final int MAX_CONTEXT_ID_BYTES = 128;
+
+ /**
+ * Validates the component strings.
+ *
+ * @throws IllegalArgumentException when either component is null or blank
+ */
+ public QueryJobId {
+ if (ownerNodeId == null || ownerNodeId.isBlank()) {
+ throw new IllegalArgumentException("PPL asynchronous query owner node must not be empty");
+ }
+ if (contextId == null || contextId.isBlank()) {
+ throw new IllegalArgumentException("PPL asynchronous query context must not be empty");
+ }
+ }
+
+ /**
+ * Creates a new job ID for the given owner node.
+ *
+ * @param ownerNodeId node that will own the job
+ * @return unencoded job ID with a random context ID
+ */
+ public static QueryJobId create(String ownerNodeId) {
+ return new QueryJobId(ownerNodeId, UUID.randomUUID().toString());
+ }
+
+ /**
+ * Serializes this ID using the versioned binary format and URL-safe Base64 without padding.
+ *
+ * @return opaque value returned by the asynchronous PPL API
+ */
+ public String encode() {
+ try (ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream output = new DataOutputStream(bytes)) {
+ output.writeInt(FORMAT_VERSION);
+ writeString(output, ownerNodeId, MAX_OWNER_NODE_ID_BYTES);
+ writeString(output, contextId, MAX_CONTEXT_ID_BYTES);
+ output.flush();
+ return Base64.getUrlEncoder().withoutPadding().encodeToString(bytes.toByteArray());
+ } catch (IOException e) {
+ throw new IllegalStateException("Failed to encode PPL asynchronous query ID", e);
+ }
+ }
+
+ /**
+ * Decodes and validates an opaque job ID.
+ *
+ *
Parsing rejects empty or oversized input, unsupported versions, invalid component lengths,
+ * truncated data, and trailing bytes. All malformed input is reported uniformly to avoid exposing
+ * details of the internal encoding.
+ *
+ * @param encoded opaque value supplied by the client
+ * @return decoded owner node and context IDs
+ * @throws IllegalArgumentException when the value is not a valid job ID
+ */
+ public static QueryJobId parse(String encoded) {
+ try {
+ if (encoded == null || encoded.isBlank() || encoded.length() > MAX_ENCODED_LENGTH) {
+ throw new IllegalArgumentException("Invalid PPL asynchronous query ID length");
+ }
+ byte[] bytes = Base64.getUrlDecoder().decode(encoded);
+ try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes))) {
+ int version = input.readInt();
+ if (version != FORMAT_VERSION) {
+ throw new IllegalArgumentException(
+ "Unsupported PPL asynchronous query ID version [" + version + "]");
+ }
+ QueryJobId id =
+ new QueryJobId(
+ readString(input, MAX_OWNER_NODE_ID_BYTES),
+ readString(input, MAX_CONTEXT_ID_BYTES));
+ if (input.available() != 0) {
+ throw new IllegalArgumentException(
+ "Unexpected trailing bytes in PPL asynchronous query ID");
+ }
+ return id;
+ }
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Invalid PPL asynchronous query ID", e);
+ }
+ }
+
+ private static void writeString(DataOutputStream output, String value, int maxLength)
+ throws IOException {
+ byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+ if (bytes.length > maxLength) {
+ throw new IllegalArgumentException("PPL asynchronous query ID component is too long");
+ }
+ output.writeInt(bytes.length);
+ output.write(bytes);
+ }
+
+ private static String readString(DataInputStream input, int maxLength) throws IOException {
+ int length = input.readInt();
+ if (length < 0 || length > maxLength) {
+ throw new IllegalArgumentException("Invalid PPL asynchronous query ID component length");
+ }
+ byte[] bytes = input.readNBytes(length);
+ if (bytes.length != length) {
+ throw new IllegalArgumentException("Truncated PPL asynchronous query ID");
+ }
+ return new String(bytes, StandardCharsets.UTF_8);
+ }
+}
diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobOwner.java b/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobOwner.java
new file mode 100644
index 00000000000..243d5d6f602
--- /dev/null
+++ b/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobOwner.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin.transport.asyncquery;
+
+import java.util.List;
+import java.util.Objects;
+import org.opensearch.OpenSearchSecurityException;
+import org.opensearch.common.util.concurrent.ThreadContext;
+import org.opensearch.commons.ConfigConstants;
+import org.opensearch.commons.authuser.User;
+import org.opensearch.core.rest.RestStatus;
+
+/**
+ * Immutable owner identity used to authorize retained asynchronous query state.
+ *
+ * @param name authenticated principal, or {@code null} when no identity was supplied
+ * @param requestedTenant requested security tenant
+ * @param backendRoles backend roles captured when the job starts
+ */
+public record QueryJobOwner(String name, String requestedTenant, List backendRoles) {
+
+ /**
+ * Identity used when OpenSearch Security does not provide caller information.
+ *
+ * This identity does not bypass job ownership checks. A job owned by {@code UNSECURED} can be
+ * accessed only by a caller represented by the same identity.
+ */
+ public static final QueryJobOwner UNSECURED = new QueryJobOwner(null, null, List.of());
+
+ /**
+ * Creates an immutable asynchronous query identity.
+ *
+ * @param name authenticated principal, or {@code null} when no identity was supplied
+ * @param requestedTenant requested security tenant
+ * @param backendRoles backend roles captured when the job starts
+ * @throws IllegalArgumentException if {@code name} is blank
+ */
+ public QueryJobOwner {
+ backendRoles = backendRoles == null ? List.of() : List.copyOf(backendRoles);
+ if (name != null && name.isBlank()) {
+ throw new IllegalArgumentException("PPL asynchronous query user must not be blank");
+ }
+ }
+
+ /**
+ * Captures the current caller from the OpenSearch thread context.
+ *
+ * @param threadContext current request thread context
+ * @return immutable caller identity
+ * @throws OpenSearchSecurityException if the security identity cannot be parsed
+ */
+ public static QueryJobOwner current(ThreadContext threadContext) {
+ try {
+ Object serialized =
+ threadContext.getTransient(ConfigConstants.OPENSEARCH_SECURITY_USER_INFO_THREAD_CONTEXT);
+ if (serialized == null) {
+ return UNSECURED;
+ }
+ User user =
+ serialized instanceof User currentUser
+ ? currentUser
+ : serialized instanceof String value ? User.parse(value) : null;
+ if (user == null) {
+ throw forbidden();
+ }
+ return new QueryJobOwner(user.getName(), user.getRequestedTenant(), user.getBackendRoles());
+ } catch (RuntimeException e) {
+ throw forbidden();
+ }
+ }
+
+ /**
+ * Verifies that a caller may access asynchronous query state owned by this identity.
+ *
+ * @param caller identity of the caller requesting access
+ * @throws OpenSearchSecurityException if the caller does not match the owner identity
+ */
+ public void authorize(QueryJobOwner caller) {
+ if (!Objects.equals(name, caller.name)
+ || !Objects.equals(requestedTenant, caller.requestedTenant)
+ || !caller.backendRoles.containsAll(backendRoles)) {
+ throw forbidden();
+ }
+ }
+
+ private static OpenSearchSecurityException forbidden() {
+ return new OpenSearchSecurityException(
+ "Not authorized to access PPL asynchronous query", RestStatus.FORBIDDEN);
+ }
+}
diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobRegistry.java b/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobRegistry.java
new file mode 100644
index 00000000000..ac8cfe010d6
--- /dev/null
+++ b/plugin/src/main/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobRegistry.java
@@ -0,0 +1,103 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin.transport.asyncquery;
+
+import java.io.Closeable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Retained asynchronous PPL jobs on the owner node.
+ *
+ *
The registry is a thin wrapper around a concurrent map. It exposes the four operations {@link
+ * QueryJob} needs to make itself discoverable, and no more: capacity accounting, listener
+ * notification, and task management remain inside {@link QueryJob}. All methods are safe for
+ * concurrent use.
+ *
+ *
Only running jobs are registered. A job becomes visible when {@link QueryJob#create} publishes
+ * it and is removed on completion, DELETE, expiry, or shutdown. On {@link #close}, the registry
+ * stops accepting new jobs and asks each remaining job to release its resources.
+ */
+public final class QueryJobRegistry implements Closeable {
+ private final ConcurrentMap jobs = new ConcurrentHashMap<>();
+ private volatile boolean closed;
+
+ /**
+ * Publishes a job under its opaque ID.
+ *
+ * If a job with the same ID is already registered, the existing job is returned and the caller
+ * is expected to abort the incoming duplicate. In practice, {@link QueryJobId#create} draws its
+ * context ID from {@link java.util.UUID#randomUUID} and collisions are not observed.
+ *
+ * @param job job to publish; its ID is used as the registry key
+ * @return existing job if the ID was already registered, {@code null} on successful insertion
+ * @throws IllegalStateException when the registry has been closed
+ */
+ public QueryJob add(QueryJob job) {
+ Objects.requireNonNull(job);
+ if (closed) {
+ throw new IllegalStateException("PPL asynchronous query registry is closed");
+ }
+ return jobs.putIfAbsent(job.getJobId(), job);
+ }
+
+ /**
+ * Looks up a job by opaque ID.
+ *
+ * @param id opaque ID assigned by this node
+ * @return job when the ID resolves locally, or empty when the ID is unknown
+ */
+ public Optional get(QueryJobId id) {
+ Objects.requireNonNull(id);
+ return Optional.ofNullable(jobs.get(id));
+ }
+
+ /**
+ * Removes a job only when the given ID still resolves to the given job instance.
+ *
+ * @param id opaque ID assigned by this node
+ * @param job job that expects to still hold the registration
+ * @return {@code true} when the mapping was removed
+ */
+ public boolean remove(QueryJobId id, QueryJob job) {
+ Objects.requireNonNull(id);
+ Objects.requireNonNull(job);
+ return jobs.remove(id, job);
+ }
+
+ /**
+ * Snapshot of every job currently registered.
+ *
+ * Iteration order is unspecified. The returned collection is a defensive copy and does not
+ * reflect later registry mutations.
+ *
+ * @return jobs registered at the moment of the call
+ */
+ public Collection jobs() {
+ return List.copyOf(jobs.values());
+ }
+
+ /**
+ * Stops accepting new jobs and asks every registered job to release its resources.
+ *
+ * Each surviving job receives {@link QueryJob#discard} with a shutdown reason. The registry
+ * itself becomes idempotent: subsequent calls are no-ops and any future {@link #add} throws.
+ */
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ for (QueryJob job : jobs.values()) {
+ job.discard("PPL asynchronous query service is closing");
+ }
+ }
+}
diff --git a/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobIdTest.java b/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobIdTest.java
new file mode 100644
index 00000000000..d3cb6fd5e8b
--- /dev/null
+++ b/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobIdTest.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin.transport.asyncquery;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+import org.junit.Test;
+
+public class QueryJobIdTest {
+
+ @Test
+ public void roundTripsOwnerAndContext() {
+ QueryJobId id = QueryJobId.create("node-a");
+
+ assertEquals(id, QueryJobId.parse(id.encode()));
+ }
+
+ @Test
+ public void rejectsMalformedIds() {
+ assertThrows(IllegalArgumentException.class, () -> QueryJobId.parse(""));
+ assertThrows(IllegalArgumentException.class, () -> QueryJobId.parse("not-an-id"));
+ assertThrows(IllegalArgumentException.class, () -> new QueryJobId("", "context-id"));
+ }
+}
diff --git a/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobOwnerTest.java b/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobOwnerTest.java
new file mode 100644
index 00000000000..aa614bf7ac7
--- /dev/null
+++ b/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobOwnerTest.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin.transport.asyncquery;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+import java.util.List;
+import org.junit.Test;
+import org.opensearch.OpenSearchSecurityException;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.concurrent.ThreadContext;
+import org.opensearch.commons.ConfigConstants;
+import org.opensearch.commons.authuser.User;
+
+public class QueryJobOwnerTest {
+
+ @Test
+ public void capturesSecurityIdentity() {
+ ThreadContext context = contextWith("alice|backend-a|ppl-role|tenant-a");
+
+ assertEquals(
+ new QueryJobOwner("alice", "tenant-a", List.of("backend-a")),
+ QueryJobOwner.current(context));
+ }
+
+ @Test
+ public void acceptsUserObject() {
+ ThreadContext context =
+ contextWith(new User("alice", List.of("backend-a"), List.of("ppl-role"), null, "tenant-a"));
+
+ assertEquals(
+ new QueryJobOwner("alice", "tenant-a", List.of("backend-a")),
+ QueryJobOwner.current(context));
+ }
+
+ @Test
+ public void rejectsUnknownSecurityContext() {
+ assertThrows(
+ OpenSearchSecurityException.class, () -> QueryJobOwner.current(contextWith(new Object())));
+ }
+
+ @Test
+ public void missingIdentityRepresentsAnUnsecuredCaller() {
+ ThreadContext context = new ThreadContext(Settings.EMPTY);
+
+ assertEquals(QueryJobOwner.UNSECURED, QueryJobOwner.current(context));
+ }
+
+ @Test
+ public void requiresSamePrincipalTenantAndOriginalBackendRoles() {
+ QueryJobOwner owner = new QueryJobOwner("alice", "tenant-a", List.of("role-a"));
+
+ owner.authorize(new QueryJobOwner("alice", "tenant-a", List.of("role-a", "newly-added-role")));
+
+ assertThrows(
+ OpenSearchSecurityException.class,
+ () -> owner.authorize(new QueryJobOwner("bob", "tenant-a", List.of("role-a"))));
+ assertThrows(
+ OpenSearchSecurityException.class,
+ () -> owner.authorize(new QueryJobOwner("alice", "tenant-b", List.of("role-a"))));
+ assertThrows(
+ OpenSearchSecurityException.class,
+ () -> owner.authorize(new QueryJobOwner("alice", "tenant-a", List.of())));
+ }
+
+ @Test
+ public void unsecuredModeRequiresAnUnsecuredCaller() {
+ QueryJobOwner.UNSECURED.authorize(QueryJobOwner.UNSECURED);
+
+ assertThrows(
+ OpenSearchSecurityException.class,
+ () ->
+ QueryJobOwner.UNSECURED.authorize(new QueryJobOwner("alice", null, List.of("role-a"))));
+ }
+
+ private static ThreadContext contextWith(Object userInfo) {
+ ThreadContext context = new ThreadContext(Settings.EMPTY);
+ context.putTransient(ConfigConstants.OPENSEARCH_SECURITY_USER_INFO_THREAD_CONTEXT, userInfo);
+ return context;
+ }
+}
diff --git a/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobRegistryTest.java b/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobRegistryTest.java
new file mode 100644
index 00000000000..e0aeec06bb4
--- /dev/null
+++ b/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobRegistryTest.java
@@ -0,0 +1,99 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin.transport.asyncquery;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class QueryJobRegistryTest {
+
+ @Test
+ public void addPublishesJobAndReturnsExistingOnCollision() {
+ QueryJobRegistry registry = new QueryJobRegistry();
+ QueryJobId sharedId = new QueryJobId("node-a", "context-1");
+ QueryJob first = jobWithId(sharedId);
+ QueryJob second = jobWithId(sharedId);
+
+ assertNull(registry.add(first));
+ assertSame(first, registry.add(second));
+ assertSame(first, registry.get(sharedId).orElseThrow());
+ }
+
+ @Test
+ public void removeOnlyDropsGivenJobInstance() {
+ QueryJobRegistry registry = new QueryJobRegistry();
+ QueryJobId sharedId = new QueryJobId("node-a", "context-1");
+ QueryJob first = jobWithId(sharedId);
+ QueryJob replacement = jobWithId(sharedId);
+
+ registry.add(first);
+ assertFalse(registry.remove(sharedId, replacement));
+ assertSame(first, registry.get(sharedId).orElseThrow());
+ assertTrue(registry.remove(sharedId, first));
+ assertEquals(Optional.empty(), registry.get(sharedId));
+ }
+
+ @Test
+ public void closeDiscardsEveryJobAndRejectsFurtherAdds() {
+ QueryJobRegistry registry = new QueryJobRegistry();
+ QueryJob first = jobWithId(newId());
+ QueryJob second = jobWithId(newId());
+ registry.add(first);
+ registry.add(second);
+
+ registry.close();
+
+ Mockito.verify(first).discard("PPL asynchronous query service is closing");
+ Mockito.verify(second).discard("PPL asynchronous query service is closing");
+ assertThrows(IllegalStateException.class, () -> registry.add(jobWithId(newId())));
+ }
+
+ @Test
+ public void closeIsIdempotent() {
+ QueryJobRegistry registry = new QueryJobRegistry();
+ QueryJob job = jobWithId(newId());
+ registry.add(job);
+
+ registry.close();
+ registry.close();
+
+ Mockito.verify(job, Mockito.times(1)).discard(Mockito.anyString());
+ }
+
+ @Test
+ public void jobsReturnsDefensiveSnapshot() {
+ QueryJobRegistry registry = new QueryJobRegistry();
+ QueryJob job = jobWithId(newId());
+ registry.add(job);
+
+ var snapshot = registry.jobs();
+ registry.remove(job.getJobId(), job);
+
+ assertTrue(snapshot.contains(job));
+ assertEquals(0, registry.jobs().size());
+ }
+
+ private static final AtomicInteger COUNTER = new AtomicInteger();
+
+ private static QueryJobId newId() {
+ return new QueryJobId("node-a", "context-" + COUNTER.incrementAndGet());
+ }
+
+ private static QueryJob jobWithId(QueryJobId id) {
+ QueryJob job = Mockito.mock(QueryJob.class);
+ Mockito.when(job.getJobId()).thenReturn(id);
+ return job;
+ }
+}
diff --git a/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobTest.java b/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobTest.java
new file mode 100644
index 00000000000..28d317b4da8
--- /dev/null
+++ b/plugin/src/test/java/org/opensearch/sql/plugin/transport/asyncquery/QueryJobTest.java
@@ -0,0 +1,655 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin.transport.asyncquery;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.function.LongSupplier;
+import org.junit.Test;
+import org.opensearch.OpenSearchSecurityException;
+import org.opensearch.ResourceNotFoundException;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.action.ActionListener;
+import org.opensearch.sql.data.model.ExprValue;
+import org.opensearch.sql.data.model.ExprValueUtils;
+import org.opensearch.sql.data.type.ExprCoreType;
+import org.opensearch.sql.executor.AsyncQueryExecution;
+import org.opensearch.sql.executor.ExecutionEngine.QueryResponse;
+import org.opensearch.sql.executor.ExecutionEngine.Schema;
+import org.opensearch.sql.executor.ExecutionEngine.Schema.Column;
+import org.opensearch.sql.legacy.metrics.BasicCounter;
+import org.opensearch.sql.legacy.metrics.MetricName;
+import org.opensearch.sql.legacy.metrics.Metrics;
+import org.opensearch.sql.legacy.metrics.NumericMetric;
+import org.opensearch.tasks.CancellableTask;
+import org.opensearch.tasks.TaskManager;
+
+public class QueryJobTest {
+ private static final String LOCAL_NODE = "node-a";
+ private static final QueryJobOwner OWNER = QueryJobOwner.UNSECURED;
+ private static final TimeValue KEEP_ALIVE = TimeValue.timeValueMinutes(5);
+ private static final TimeValue DEFAULT_WAIT = TimeValue.timeValueSeconds(5);
+
+ @Test
+ public void fastSuccessReturnsDirectResultWithoutRetention() {
+ scenario()
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .withCurrentResult(response(2))
+ .start()
+ .completeAfterMillis(25)
+ .assertDirectSuccess(response(2), 25)
+ .assertNotRegistered()
+ .assertExecutionClosedOnce();
+ }
+
+ @Test
+ public void completedExecutionCanBeAttachedBeforeCompletionIsObserved() {
+ scenario()
+ .succeed(response(2))
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .start()
+ .assertDirectSuccess(response(2), 0)
+ .assertExecutionClosedOnce();
+ }
+
+ @Test
+ public void timeoutReturnsIdAndLaterGetReturnsCompleteResult() {
+ QueryJobScenario scenario =
+ scenario()
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .start()
+ .fireRetention()
+ .assertRetainedRunning(Optional.empty());
+
+ scenario
+ .succeedAfterMillis(25, response(2))
+ .get()
+ .assertGetResponse(
+ new QueryJob.Snapshot.Succeeded(Optional.of(scenario.id()), response(2), 25));
+ }
+
+ @Test
+ public void fastFailureReturnsDirectFailureWithoutId() {
+ scenario()
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .start()
+ .fail(new IllegalStateException("boom"))
+ .assertFailure(IllegalStateException.class, "boom")
+ .assertNotRegistered()
+ .assertExecutionClosedOnce();
+ }
+
+ @Test
+ public void executionStartFailureCompletesJobAndReleasesTask() {
+ QueryJobScenario scenario =
+ scenario()
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .withExecutionStarter(
+ ignored -> {
+ throw new IllegalStateException("execution did not start");
+ });
+ scenario.start();
+ scenario
+ .assertFailure(IllegalStateException.class, "execution did not start")
+ .assertNotRegistered()
+ .assertTaskClosed();
+ }
+
+ @Test
+ public void retainedFailureRetainsFailureSnapshotAndRecordsMetric() {
+ NumericMetric customerFailures =
+ new NumericMetric<>(MetricName.PPL_FAILED_REQ_COUNT_CUS.getName(), new BasicCounter());
+ Metrics.getInstance().registerMetric(customerFailures);
+ try {
+ QueryJobScenario scenario =
+ scenario()
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .start()
+ .fireRetention()
+ .fail(new IllegalArgumentException("invalid query"))
+ .get();
+
+ scenario.assertGetResponse(
+ new QueryJob.Snapshot.Failed(
+ Optional.of(scenario.id()),
+ new QueryJob.Failure("IllegalArgumentException", "invalid query"),
+ 0));
+ assertEquals(Long.valueOf(1), customerFailures.getValue());
+ } finally {
+ Metrics.getInstance().unregisterMetric(customerFailures.getName());
+ }
+ }
+
+ @Test
+ public void expiredGetRemovesJobAndCancelsTask() {
+ CancellableTask task = runningTask();
+ QueryJobScenario scenario =
+ scenario().withTask(task).start().fireRetention().advanceMillis(KEEP_ALIVE.millis());
+
+ ResourceNotFoundException error = assertThrows(ResourceNotFoundException.class, scenario::get);
+
+ assertEquals("PPL asynchronous query not found", error.getMessage());
+ scenario.assertNotRegistered();
+ verify(task).cancel("PPL asynchronous query expired");
+ }
+
+ @Test
+ public void expiryTimerRemovesJobAfterKeepAliveElapses() {
+ CancellableTask task = runningTask();
+ QueryJobScenario scenario =
+ scenario().withTask(task).start().fireRetention().assertRegistered();
+
+ scenario.advanceMillis(KEEP_ALIVE.millis()).fireExpiry();
+
+ scenario.assertNotRegistered();
+ verify(task).cancel("PPL asynchronous query expired");
+ }
+
+ @Test
+ public void deleteCancelsRunningJobAndReturnsCancelledStatus() {
+ CancellableTask task = runningTask();
+ QueryJobScenario scenario = scenario().withTask(task).start().fireRetention().delete();
+
+ scenario.assertDeleteStatus(QueryJob.Status.CANCELLED);
+ verify(task).cancel("PPL asynchronous query cancelled by user");
+ scenario.assertNotRegistered().assertExecutionClosedOnce();
+ assertThrows(ResourceNotFoundException.class, scenario::get);
+ }
+
+ @Test
+ public void deleteReturnsRetainedTerminalStatus() {
+ CancellableTask task = runningTask();
+ QueryJobScenario scenario =
+ scenario()
+ .withTask(task)
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .withCurrentResult(response(1))
+ .start()
+ .fireRetention()
+ .complete()
+ .delete();
+
+ scenario.assertDeleteStatus(QueryJob.Status.SUCCEEDED);
+ verify(task, never()).cancel(org.mockito.ArgumentMatchers.anyString());
+ scenario.assertNotRegistered().assertExecutionClosedOnce();
+ }
+
+ @Test
+ public void deleteReturnsNotFoundAfterExpirationBoundary() {
+ QueryJobScenario scenario = scenario().start().fireRetention();
+
+ scenario.advanceMillis(KEEP_ALIVE.millis());
+
+ assertThrows(ResourceNotFoundException.class, scenario::delete);
+ scenario.assertNotRegistered();
+ }
+
+ @Test
+ public void getRenewsLeaseAndResetsExpiryTimer() {
+ QueryJobScenario scenario =
+ scenario()
+ .withKeepAlive(TimeValue.timeValueSeconds(1))
+ .start()
+ .fireRetention()
+ .advanceMillis(500)
+ .getWithKeepAlive(TimeValue.timeValueSeconds(2))
+ .advanceMillis(1_999)
+ .assertRegistered();
+
+ // Original lease would have expired at 1000ms; renewed lease survives to 2499ms.
+ scenario.advanceMillis(1).fireExpiry().assertNotRegistered();
+ }
+
+ @Test
+ public void cancellationUsesTaskManagerWhenAttached() {
+ TaskManager taskManager = mock(TaskManager.class);
+ CancellableTask task = runningTask();
+ scenario().withTaskManager(taskManager).withTask(task).start().fireRetention().delete();
+
+ verify(taskManager)
+ .cancelTaskAndDescendants(
+ org.mockito.ArgumentMatchers.eq(task),
+ org.mockito.ArgumentMatchers.eq("PPL asynchronous query cancelled by user"),
+ org.mockito.ArgumentMatchers.eq(false),
+ org.mockito.ArgumentMatchers.any());
+ }
+
+ @Test
+ public void shutdownDiscardsRunningJobAndClosesResources() {
+ CancellableTask task = runningTask();
+ QueryJobScenario scenario =
+ scenario().withTask(task).withWaitForCompletion(TimeValue.timeValueSeconds(5)).start();
+
+ scenario.closeRegistry();
+
+ scenario.assertNotRegistered().assertExecutionClosedOnce();
+ verify(task).cancel("PPL asynchronous query service is closing");
+ }
+
+ @Test
+ public void unauthorizedCallerCannotAccessOwnedJob() {
+ QueryJobOwner alice = new QueryJobOwner("alice", "tenant-a", List.of("role-a"));
+ QueryJobOwner bob = new QueryJobOwner("bob", "tenant-a", List.of("role-a"));
+ QueryJobScenario scenario = scenario().withOwner(alice).start().fireRetention();
+
+ QueryJob job = scenario.job();
+ assertThrows(OpenSearchSecurityException.class, () -> job.getOwner().authorize(bob));
+ job.getOwner().authorize(alice);
+ }
+
+ @Test
+ public void lateExecutionAttachmentAfterDeleteIsClosed() {
+ TrackingExecution execution = new TrackingExecution(response(1));
+ QueryJobScenario scenario = scenario().withExecution(execution);
+ scenario
+ .withExecutionStarter(
+ ignored -> {
+ // Retention fires and delete removes the job before the execution attaches.
+ scenario.fireRetention();
+ QueryJobId retainedId =
+ ((QueryJob.Snapshot.Running) scenario.initialResponse.get()).jobId();
+ scenario.registry.get(retainedId).orElseThrow().cancel("test delete");
+ return execution;
+ })
+ .start();
+
+ assertEquals(1, execution.closes.get());
+ assertTrue(scenario.registry.jobs().isEmpty());
+ }
+
+ @Test
+ public void finalSnapshotDefensivelyCopiesRows() {
+ List rows = new ArrayList<>();
+ rows.add(ExprValueUtils.stringValue("first"));
+ QueryResponse response =
+ new QueryResponse(
+ new Schema(List.of(new Column("state", null, ExprCoreType.STRING))), rows, null);
+
+ QueryJobScenario scenario =
+ scenario()
+ .withCurrentResult(response)
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .start()
+ .complete();
+ rows.add(ExprValueUtils.stringValue("second"));
+
+ scenario.assertSucceededRowCount(1);
+ }
+
+ @Test
+ public void successfulCompletionRequiresFinalResultToBeVisible() {
+ scenario()
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .start()
+ .complete()
+ .assertFailure(IllegalStateException.class)
+ .assertExecutionClosedOnce();
+ }
+
+ @Test
+ public void retentionListenerFailureAbortsUndeliverableJob() {
+ CancellableTask task = runningTask();
+ QueryJobScenario scenario =
+ scenario()
+ .withTask(task)
+ .withResponseListener(new FailingListener())
+ .withWaitForCompletion(TimeValue.timeValueSeconds(5))
+ .start();
+
+ scenario.fireRetention();
+
+ scenario.assertNotRegistered().assertExecutionClosedOnce();
+ verify(task).cancel("PPL asynchronous query startup failed");
+ }
+
+ private QueryJobScenario scenario() {
+ return new QueryJobScenario();
+ }
+
+ private static CancellableTask runningTask() {
+ CancellableTask task = mock(CancellableTask.class);
+ when(task.isCancelled()).thenReturn(false);
+ return task;
+ }
+
+ private static QueryResponse response(int rowCount) {
+ Schema schema = new Schema(List.of(new Column("state", null, ExprCoreType.STRING)));
+ return new QueryResponse(
+ schema,
+ java.util.stream.IntStream.range(0, rowCount)
+ .mapToObj(i -> ExprValueUtils.stringValue("state-" + i))
+ .toList(),
+ null);
+ }
+
+ /** Deterministic fluent scenario driving one {@link QueryJob} through its lifecycle. */
+ private final class QueryJobScenario {
+ private final AtomicLong clock = new AtomicLong(1_000);
+ private final QueryJobRegistry registry = new QueryJobRegistry();
+ private final FakeScheduler scheduler = new FakeScheduler();
+ private QueryJobOwner owner = OWNER;
+ private TimeValue keepAlive = KEEP_ALIVE;
+ private TimeValue waitForCompletion = DEFAULT_WAIT;
+ private CancellableTask task = mock(CancellableTask.class);
+ private final AtomicInteger taskReleaseCount = new AtomicInteger();
+ private TaskManager taskManager;
+ private TrackingExecution trackingExecution = new TrackingExecution(null);
+ private AsyncQueryExecution execution = trackingExecution;
+ private Function executionStarter = ignored -> execution;
+ private ActionListener responseListener;
+ private final AtomicReference initialResponse = new AtomicReference<>();
+ private final AtomicReference failure = new AtomicReference<>();
+ private final AtomicInteger responseCount = new AtomicInteger();
+ private QueryJob.Snapshot getResponse;
+ private QueryJob.Status deleteStatus;
+ private QueryJob job;
+
+ private QueryJobScenario withOwner(QueryJobOwner owner) {
+ this.owner = owner;
+ return this;
+ }
+
+ private QueryJobScenario withKeepAlive(TimeValue keepAlive) {
+ this.keepAlive = keepAlive;
+ return this;
+ }
+
+ private QueryJobScenario withWaitForCompletion(TimeValue wait) {
+ this.waitForCompletion = wait;
+ return this;
+ }
+
+ private QueryJobScenario withTask(CancellableTask task) {
+ this.task = task;
+ return this;
+ }
+
+ private QueryJobScenario withTaskManager(TaskManager taskManager) {
+ this.taskManager = taskManager;
+ return this;
+ }
+
+ private QueryJobScenario withCurrentResult(QueryResponse response) {
+ trackingExecution.setCurrent(response);
+ return this;
+ }
+
+ private QueryJobScenario withExecution(AsyncQueryExecution execution) {
+ this.execution = execution;
+ trackingExecution = execution instanceof TrackingExecution tracking ? tracking : null;
+ executionStarter = ignored -> this.execution;
+ return this;
+ }
+
+ private QueryJobScenario withExecutionStarter(
+ Function starter) {
+ this.executionStarter = starter;
+ return this;
+ }
+
+ private QueryJobScenario withResponseListener(ActionListener listener) {
+ this.responseListener = listener;
+ return this;
+ }
+
+ private QueryJobScenario start() {
+ QueryJob.JobTask jobTask = new QueryJob.JobTask(task, taskReleaseCount::incrementAndGet);
+ ActionListener listener =
+ responseListener != null ? responseListener : defaultListener();
+ job =
+ QueryJob.create(
+ LOCAL_NODE,
+ registry,
+ (LongSupplier) clock::get,
+ scheduler,
+ taskManager,
+ jobTask,
+ owner,
+ keepAlive,
+ waitForCompletion,
+ executionStarter,
+ listener);
+ return this;
+ }
+
+ private ActionListener defaultListener() {
+ return ActionListener.wrap(
+ snapshot -> {
+ initialResponse.set(snapshot);
+ responseCount.incrementAndGet();
+ },
+ failure::set);
+ }
+
+ private QueryJobScenario succeed(QueryResponse response) {
+ trackingExecution.succeed(response);
+ return this;
+ }
+
+ private QueryJobScenario complete() {
+ trackingExecution.complete();
+ return this;
+ }
+
+ private QueryJobScenario completeAfterMillis(long elapsedMillis) {
+ clock.addAndGet(elapsedMillis);
+ trackingExecution.complete();
+ return this;
+ }
+
+ private QueryJobScenario succeedAfterMillis(long elapsedMillis, QueryResponse response) {
+ clock.addAndGet(elapsedMillis);
+ trackingExecution.succeed(response);
+ return this;
+ }
+
+ private QueryJobScenario fail(Exception exception) {
+ trackingExecution.fail(exception);
+ return this;
+ }
+
+ private QueryJobScenario advanceMillis(long delta) {
+ clock.addAndGet(delta);
+ return this;
+ }
+
+ private QueryJobScenario fireRetention() {
+ scheduler.fireNext();
+ return this;
+ }
+
+ private QueryJobScenario fireExpiry() {
+ scheduler.fireNext();
+ return this;
+ }
+
+ private QueryJobScenario get() {
+ getResponse = job.get(null);
+ return this;
+ }
+
+ private QueryJobScenario getWithKeepAlive(TimeValue newKeepAlive) {
+ getResponse = job.get(newKeepAlive);
+ return this;
+ }
+
+ private QueryJobScenario delete() {
+ deleteStatus = job.cancel("PPL asynchronous query cancelled by user");
+ return this;
+ }
+
+ private QueryJobScenario closeRegistry() {
+ registry.close();
+ return this;
+ }
+
+ private QueryJobScenario assertDirectSuccess(QueryResponse expected, long tookMillis) {
+ assertEquals(
+ new QueryJob.Snapshot.Succeeded(Optional.empty(), expected, tookMillis),
+ initialResponse.get());
+ return this;
+ }
+
+ private QueryJobScenario assertRetainedRunning(Optional expected) {
+ assertEquals(new QueryJob.Snapshot.Running(id(), expected), initialResponse.get());
+ return this;
+ }
+
+ private QueryJobScenario assertGetResponse(QueryJob.Snapshot expected) {
+ assertEquals(expected, getResponse);
+ return this;
+ }
+
+ private QueryJobScenario assertDeleteStatus(QueryJob.Status expected) {
+ assertEquals(expected, deleteStatus);
+ return this;
+ }
+
+ private QueryJobScenario assertSucceededRowCount(int expected) {
+ assertTrue(initialResponse.get() instanceof QueryJob.Snapshot.Succeeded);
+ QueryJob.Snapshot.Succeeded succeeded = (QueryJob.Snapshot.Succeeded) initialResponse.get();
+ assertEquals(expected, succeeded.response().getResults().size());
+ return this;
+ }
+
+ private QueryJobScenario assertFailure(Class type) {
+ assertTrue(type.isInstance(failure.get()));
+ return this;
+ }
+
+ private QueryJobScenario assertFailure(Class type, String message) {
+ assertTrue(type.isInstance(failure.get()));
+ assertEquals(message, failure.get().getMessage());
+ return this;
+ }
+
+ private QueryJobScenario assertRegistered() {
+ assertTrue(registry.get(job.getJobId()).isPresent());
+ return this;
+ }
+
+ private QueryJobScenario assertNotRegistered() {
+ assertTrue(registry.get(job.getJobId()).isEmpty());
+ return this;
+ }
+
+ private QueryJobScenario assertExecutionClosedOnce() {
+ if (trackingExecution != null) {
+ assertEquals(1, trackingExecution.closes.get());
+ }
+ return this;
+ }
+
+ private QueryJobScenario assertTaskClosed() {
+ assertEquals(1, taskReleaseCount.get());
+ return this;
+ }
+
+ private QueryJobId id() {
+ return job.getJobId();
+ }
+
+ private QueryJob job() {
+ return job;
+ }
+ }
+
+ private static final class FakeScheduler implements QueryJob.Scheduler {
+ private final Deque pending = new ArrayDeque<>();
+
+ @Override
+ public QueryJob.Cancellable schedule(long delayMillis, Runnable task) {
+ Pending entry = new Pending(delayMillis, task);
+ pending.addLast(entry);
+ return () -> pending.remove(entry);
+ }
+
+ void fireNext() {
+ Pending entry = pending.pollFirst();
+ assertTrue("No scheduled task", entry != null);
+ entry.task.run();
+ }
+
+ private record Pending(long delayMillis, Runnable task) {}
+ }
+
+ private static final class TrackingExecution implements AsyncQueryExecution {
+ private final AtomicReference current;
+ private final CompletableFuture completion = new CompletableFuture<>();
+ private final AtomicBoolean closed = new AtomicBoolean();
+ final AtomicInteger reads = new AtomicInteger();
+ final AtomicInteger closes = new AtomicInteger();
+
+ private TrackingExecution(QueryResponse current) {
+ this.current = new AtomicReference<>(current);
+ }
+
+ private void setCurrent(QueryResponse response) {
+ current.set(response);
+ }
+
+ private void succeed(QueryResponse response) {
+ current.set(response);
+ completion.complete(null);
+ }
+
+ private void complete() {
+ completion.complete(null);
+ }
+
+ private void fail(Exception failure) {
+ completion.completeExceptionally(failure);
+ }
+
+ @Override
+ public Optional currentResult() {
+ reads.incrementAndGet();
+ return Optional.ofNullable(current.get());
+ }
+
+ @Override
+ public CompletionStage completion() {
+ return completion;
+ }
+
+ @Override
+ public void close() {
+ if (closed.compareAndSet(false, true)) {
+ closes.incrementAndGet();
+ }
+ }
+ }
+
+ private static final class FailingListener implements ActionListener {
+ @Override
+ public void onResponse(QueryJob.Snapshot snapshot) {
+ throw new IllegalStateException("listener rejected snapshot");
+ }
+
+ @Override
+ public void onFailure(Exception e) {
+ // no-op; QueryJob may deliver a subsequent failure once discard runs.
+ }
+ }
+}
diff --git a/ppl/src/main/java/org/opensearch/sql/ppl/DefaultAsyncQueryExecution.java b/ppl/src/main/java/org/opensearch/sql/ppl/DefaultAsyncQueryExecution.java
new file mode 100644
index 00000000000..d0c1e5b3921
--- /dev/null
+++ b/ppl/src/main/java/org/opensearch/sql/ppl/DefaultAsyncQueryExecution.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ppl;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import org.opensearch.sql.common.response.ResponseListener;
+import org.opensearch.sql.executor.AsyncQueryExecution;
+import org.opensearch.sql.executor.ExecutionEngine.QueryResponse;
+
+/**
+ * Final-result implementation of the asynchronous execution contract.
+ *
+ * This adapter keeps the existing callback-based query execution unchanged. Until partial-result
+ * producers are added, {@link #currentResult()} is empty while execution is running and exposes the
+ * final response immediately before successful completion is published.
+ */
+final class DefaultAsyncQueryExecution
+ implements AsyncQueryExecution, ResponseListener {
+ private final CompletableFuture completion = new CompletableFuture<>();
+ private volatile QueryResponse finalResult;
+
+ @Override
+ public void onResponse(QueryResponse response) {
+ finalResult = Objects.requireNonNull(response);
+ completion.complete(null);
+ }
+
+ @Override
+ public void onFailure(Exception failure) {
+ completion.completeExceptionally(Objects.requireNonNull(failure));
+ }
+
+ @Override
+ public Optional currentResult() {
+ return Optional.ofNullable(finalResult);
+ }
+
+ @Override
+ public CompletionStage completion() {
+ return completion;
+ }
+
+ @Override
+ public void close() {}
+}
diff --git a/ppl/src/main/java/org/opensearch/sql/ppl/PPLService.java b/ppl/src/main/java/org/opensearch/sql/ppl/PPLService.java
index e2572b5f0da..b8ac519fbbe 100644
--- a/ppl/src/main/java/org/opensearch/sql/ppl/PPLService.java
+++ b/ppl/src/main/java/org/opensearch/sql/ppl/PPLService.java
@@ -18,6 +18,7 @@
import org.opensearch.sql.common.setting.Settings;
import org.opensearch.sql.common.utils.QueryContext;
import org.opensearch.sql.executor.AnalyzeResponse;
+import org.opensearch.sql.executor.AsyncQueryExecution;
import org.opensearch.sql.executor.ExecutionEngine.ExplainResponse;
import org.opensearch.sql.executor.QueryManager;
import org.opensearch.sql.executor.QueryType;
@@ -88,6 +89,40 @@ public void execute(
}
}
+ /**
+ * Starts query execution and immediately returns its lifecycle-facing handle.
+ *
+ * The existing callback execution remains internal to the PPL execution module. This
+ * final-only implementation exposes no current result until the callback publishes the
+ * authoritative response.
+ *
+ * @param request PPL query request
+ * @param anonymizedQuerySink receives anonymized query text for metrics
+ * @return lifecycle-facing asynchronous execution handle
+ */
+ public AsyncQueryExecution executeAsync(
+ PPLQueryRequest request, Consumer anonymizedQuerySink) {
+ DefaultAsyncQueryExecution execution = new DefaultAsyncQueryExecution();
+ execute(
+ request,
+ execution,
+ new ResponseListener<>() {
+ @Override
+ public void onResponse(ExplainResponse response) {
+ execution.onFailure(
+ new IllegalStateException(
+ "Asynchronous query execution received an explain response"));
+ }
+
+ @Override
+ public void onFailure(Exception e) {
+ execution.onFailure(e);
+ }
+ },
+ anonymizedQuerySink);
+ return execution;
+ }
+
/**
* Explain the query in {@link PPLQueryRequest} using {@link ResponseListener} to get and format
* explain response.
diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/DefaultAsyncQueryExecutionTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/DefaultAsyncQueryExecutionTest.java
new file mode 100644
index 00000000000..f4a93f7fc05
--- /dev/null
+++ b/ppl/src/test/java/org/opensearch/sql/ppl/DefaultAsyncQueryExecutionTest.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ppl;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.Test;
+import org.opensearch.sql.data.model.ExprValueUtils;
+import org.opensearch.sql.data.type.ExprCoreType;
+import org.opensearch.sql.executor.ExecutionEngine.QueryResponse;
+import org.opensearch.sql.executor.ExecutionEngine.Schema;
+import org.opensearch.sql.executor.ExecutionEngine.Schema.Column;
+
+public class DefaultAsyncQueryExecutionTest {
+
+ @Test
+ public void finalResultIsVisibleBeforeSuccessfulCompletionNotification() {
+ DefaultAsyncQueryExecution execution = new DefaultAsyncQueryExecution();
+ AtomicBoolean visibleFromCompletion = new AtomicBoolean();
+ execution
+ .completion()
+ .whenComplete(
+ (ignored, failure) ->
+ visibleFromCompletion.set(
+ failure == null
+ && execution.currentResult().orElseThrow().getResults().size() == 1));
+
+ execution.onResponse(response("final"));
+
+ assertTrue(execution.completion().toCompletableFuture().isDone());
+ assertFalse(execution.completion().toCompletableFuture().isCompletedExceptionally());
+ assertTrue(visibleFromCompletion.get());
+ }
+
+ @Test
+ public void failureCompletesExceptionallyWithoutPublishingRows() {
+ DefaultAsyncQueryExecution execution = new DefaultAsyncQueryExecution();
+
+ execution.onFailure(new IllegalStateException("boom"));
+
+ CompletionException failure =
+ assertThrows(
+ CompletionException.class, () -> execution.completion().toCompletableFuture().join());
+ assertTrue(failure.getCause() instanceof IllegalStateException);
+ assertTrue(execution.currentResult().isEmpty());
+ }
+
+ private static QueryResponse response(String value) {
+ return new QueryResponse(
+ new Schema(List.of(new Column("state", null, ExprCoreType.STRING))),
+ List.of(ExprValueUtils.stringValue(value)),
+ null);
+ }
+}
diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/PPLServiceTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/PPLServiceTest.java
index fea5e3d4030..8a83dc6923b 100644
--- a/ppl/src/test/java/org/opensearch/sql/ppl/PPLServiceTest.java
+++ b/ppl/src/test/java/org/opensearch/sql/ppl/PPLServiceTest.java
@@ -20,6 +20,7 @@
import org.mockito.junit.MockitoJUnitRunner;
import org.opensearch.sql.common.response.ResponseListener;
import org.opensearch.sql.common.setting.Settings;
+import org.opensearch.sql.executor.AsyncQueryExecution;
import org.opensearch.sql.executor.DefaultQueryManager;
import org.opensearch.sql.executor.ExecutionEngine;
import org.opensearch.sql.executor.ExecutionEngine.ExplainResponse;
@@ -115,6 +116,27 @@ public void testExecuteShouldPass() {
getExplainListener(false));
}
+ @Test
+ public void testExecuteAsyncReturnsCompletedFinalResult() {
+ QueryResponse response = new QueryResponse(schema, Collections.emptyList(), Cursor.None);
+ doAnswer(
+ invocation -> {
+ ResponseListener listener = invocation.getArgument(4);
+ listener.onResponse(response);
+ return null;
+ })
+ .when(queryService)
+ .execute(any(), any(), any(), anyBoolean(), any());
+
+ AsyncQueryExecution execution =
+ pplService.executeAsync(
+ new PPLQueryRequest("search source=t a=1", null, QUERY),
+ PPLService.NO_ANONYMIZED_QUERY_SINK);
+
+ execution.completion().toCompletableFuture().join();
+ Assert.assertSame(response, execution.currentResult().orElseThrow());
+ }
+
@Test
public void testExecuteCsvFormatShouldPass() {
doAnswer(