Repository navigation
Conversation
…pter Foundational classes for the async PPL lifecycle (checkpoint 1/N). - PPLQueryRequest / TransportPPLQueryRequest / TransportPPLQueryRequestFactory: add wait_for_completion_timeout and keep_alive body fields (stored as raw strings; parsed to TimeValue in the transport action). Presence of either signals async submit; absent both keeps existing sync behavior. - PPLQueryRunner (ppl module): QueryRunner impl over PPLService.execute, same shape as the earlier iteration. - OpenSearchSecurityAdapter (opensearch module): SecurityAdapter reading OPENSEARCH_SECURITY_USER_INFO_THREAD_CONTEXT. Adds common-utils dep. Signed-off-by: Peng Huo <penghuo@amazon.com>
Checkpoint 2/N. - RetentionPolicy: TTL-based eviction of terminal jobs from QueryJobStore. Timers fire on ThreadPool.GENERIC. - OpenSearchQueryJobService: accepts an optional RetentionPolicy; arms it on every submit right after registration. - RoutingAsyncQueryExecutorService: id-shape router in the async-query module. Existing TransportGet/Cancel actions become backend-neutral: QueryJobId ids go to QueryJobService, Spark-shaped ids continue to AsyncQueryExecutorServiceImpl. Zero new REST endpoints, zero new transport ActionTypes. - Transport actions: inject AsyncQueryExecutorService (interface) so the router can be swapped in. - AsyncExecutorServiceModule: binds the router; existing test provides QueryJobService and SecurityAdapter mocks. - OpenSearchPluginModule: binds QueryJobStore, SecurityAdapter, RetentionPolicy, and QueryJobService (with 5-minute retention TTL). - TransportPPLQueryAction: async submit path per issue opensearch-project#5765. Body fields wait_for_completion_timeout / keep_alive trigger the async path; runner races against the wait; on timeout the queryId is returned in the {id, status: RUNNING, empty rows} shape; on completion within budget the sync response is returned unchanged. Explain-in-statement still routes to the two-listener direct path. Signed-off-by: Peng Huo <penghuo@amazon.com>
Checkpoint 3/N. Unit coverage for the new classes: - PPLQueryRunnerTest: success/failure round trip, single-use invariant. - OpenSearchSecurityAdapterTest: UNSECURED default, captures identity from ThreadContext, forbids unparseable transient. - RetentionPolicyTest: eviction on terminal transition, immediate eviction when scheduler rejects, positive-TTL validation. - RoutingAsyncQueryExecutorServiceTest: id-shape routing, Spark fallback, cancel dispatch, create-side always Spark. Signed-off-by: Peng Huo <penghuo@amazon.com>
Checkpoint 4/N. OpenSearch's transport-injector resolves @Inject on transport actions by exact runtime type, so the earlier RoutingAsyncQueryExecutorService subclass could not be seen as an AsyncQueryExecutorServiceImpl. Instead of subclassing, embed the id-shape routing directly into AsyncQueryExecutorServiceImpl: - Optional QueryJobService + SecurityAdapter fields, wired via a new 5-arg constructor. Existing 3-arg constructor stays for tests / legacy composition; both delegate to the same body. - getAsyncQueryResults / cancelQuery: parse the id as QueryJobId when jobService is present; on match route to the in-JVM service; on miss fall through to the existing Spark path. - toAsyncResponse: neutral QueryJobStatus -> AsyncQueryExecutionResponse mapping (SUCCEEDED carries schema+rows, FAILED carries error, others carry empty rows). - AsyncExecutorServiceModule: @provides on the impl now takes QueryJobService + SecurityAdapter. - Transport actions revert to @Inject AsyncQueryExecutorServiceImpl — which is the routing-aware impl. - Deleted RoutingAsyncQueryExecutorService{,Test}. - Added AsyncPPLQueryLifecycleIT covering sync, async submit, poll, and cancel per issue opensearch-project#5765. Signed-off-by: Peng Huo <penghuo@amazon.com>
Two Guice injectors were creating separate QueryJobService instances: - SQLPlugin's main injector (used by AsyncQueryExecutorServiceImpl for fetch/cancel routing). - TransportPPLQueryAction's local injector (previously used to resolve QueryJobService at submit time via injector.getInstance()). Result: submit stored the job in one QueryJobStore; get/cancel looked in another; every async fetch failed with QueryJobNotFoundException. Wire the SQLPlugin injector's QueryJobService and SecurityAdapter through createComponents() and inject them into TransportPPLQueryAction via @Inject (OpenSearchQueryJobService / OpenSearchSecurityAdapter concrete types so OpenSearch DI can resolve by runtime class). Both paths now share the same in-memory store. Verified with AsyncPPLQueryLifecycleIT: 4/4 pass end-to-end (sync default, async submit, async fetch, async cancel). Signed-off-by: Peng Huo <penghuo@amazon.com>
User-facing doc for the two new body fields on POST /_plugins/_ppl. Explains the async submit contract (sync default when absent, presence of either field switches to async), the runner-vs-timeout race outcomes, the still-running response shape, and how to fetch/cancel via the existing /_plugins/_async_query endpoints. Follows the semantics locked in by issue opensearch-project#5765. Signed-off-by: Peng Huo <penghuo@amazon.com>
Adds transport-level forwarding to the owner node for QueryJob-id ids. When TransportGetAsyncQueryResultAction / TransportCancelAsyncQueryRequestAction receive a queryId whose parsed ownerNodeId != this node, they forward the same request to the owner via transportService.sendRequest using the same action name. The owner node's handler resolves in its local QueryJobStore and returns the JSON-formatted response, relayed back unchanged. Same ActionType, same REST endpoint — no new AppSec surface. Spark-shaped ids continue to the local Spark path via the existing behavior. Signed-off-by: Peng Huo <penghuo@amazon.com>
Adds a dedicated two-node testCluster (asyncMultiNodeIT) and a companion Gradle task that runs AsyncPPLMultiNodeRoutingIT against it. The test submits an async query on node A, then GETs / DELETEs on node B using per-node RestClients pinned to specific HttpHosts. Verifies that node B forwards the routed request to the owner via the existing transport action and returns the terminal snapshot unchanged. The primary integTest task keeps its single-node cluster; the new class is excluded there and included in the new task. Signed-off-by: Peng Huo <penghuo@amazon.com>
…cate
Statement-level explain (query text begins with 'explain') now flows
through the async pipeline end-to-end. The runner produces one of two
QueryResult variants — Rows for ordinary queries, Explain for explain
statements — and both propagate through submit/get/cancel and cross-node
routing unchanged.
Changes:
* QueryResult becomes a sealed interface with Rows and Explain records.
QueryResult.of(QueryResponse, tookMillis) → Rows.
QueryResult.of(ExplainResponse, tookMillis) → Explain.
* PPLQueryRunner wires both listeners; explainListener firing now
completes the future with a QueryResult.Explain instead of failing.
* AsyncQueryExecutorServiceImpl.toAsyncResponse switches on the variant.
Explain results are pre-formatted to JSON via
ExplainResponseJsonFormatter (new, shared helper reused by both
async-query-core and the plugin) and carried on
AsyncQueryExecutionResponse.explainJson.
* TransportGetAsyncQueryResultAction: if explainJson is non-null,
return it verbatim; otherwise use the existing AsyncQueryResult
formatter.
* TransportPPLQueryAction.submitAsync: sync-window Explain result is
rendered via createExplainResponseListener so the body matches the
sync explain shape byte-for-byte.
* TransportPPLQueryAction: introduce shouldExecuteAsync predicate;
drop the isExplainInStatement filter (no longer needed).
* ITs cover explain-statement end-to-end (single-node + cross-node).
Coverage:
:integ-test:integTest AsyncPPLQueryLifecycleIT 5/5 pass
:integ-test:asyncMultiNodeIT AsyncPPLMultiNodeRoutingIT 3/3 pass
Signed-off-by: Peng Huo <penghuo@amazon.com>
…edicate
Two smells called out on review:
1. DEFAULT_WAIT_FOR_COMPLETION was a private constant on TransportPPLQueryAction
plus a private resolveWaitForCompletion helper. The transport layer had no
business knowing the domain default. Move it into PPLQueryRequest:
- Field type becomes java.time.Duration (was String).
- New constants DEFAULT_WAIT_FOR_COMPLETION_TIMEOUT (5s) and DEFAULT_KEEP_ALIVE (5m).
- New effectiveWaitForCompletion() / effectiveKeepAlive() return the resolved
Duration (default applied); getters preserve the was-it-set? distinction.
PPLQueryRequestFactory parses "5s" via TimeValue.parseTimeValue once at the
request boundary and stores a typed Duration; downstream never re-parses.
2. shouldExecuteAsync was a one-line wrapper around request.isAsync(). No
logic, just indirection. Deleted; the branch reads request.isAsync() inline.
If admission gates / feature flags land later, reintroduce the predicate
with real content.
TransportPPLQueryRequest field types match (Duration) and the wire format
switches to writeOptionalLong(millis) / readOptionalLong.
Signed-off-by: Peng Huo <penghuo@amazon.com>
Address the review checklist:
1. Hand-rolled JSON in asyncRunningResponse — replaced with JSONObject
(safe escaping, matches the existing json library used elsewhere in
this file). Renamed to runningSnapshot for clarity.
2. Duplicated tryParseAsJobId + forwardToOwner across two transport
actions — extracted:
- QueryJobId.tryParse(String) → Optional<QueryJobId> (on the record
itself so callers pattern-match one abstraction).
- AsyncQueryOwnerRouting.forwardToOwner(...) → shared helper in the
async-query transport package.
TransportGetAsyncQueryResultAction and TransportCancelAsyncQueryRequestAction
collapse to two-line routing hops each.
3. Ad-hoc ExecutionException unwrap in submitAsync — replaced with the
promoted-to-public QueryJob.unwrap(Throwable), the same helper the
state machine already uses. One canonical unwrap for every consumer
of QueryJob.completion().
4. effectiveKeepAlive() defined but unreachable — removed along with
DEFAULT_KEEP_ALIVE constant until retention wiring makes the value
observable. keep_alive still triggers isAsync() via field presence.
No dead code promising behavior it does not deliver.
5. AsyncQueryExecutionResponse.explainJson set via @Setter on an
otherwise immutable @DaTa class — now final. The one-arg constructor
Lombok generated widens to include explainJson; existing callers pass
null. Response shape is now immutable end-to-end.
7. submitAsync was ~55 lines and did four things — split:
- submitAsync : construct runner, submit, dispatch to await path.
- awaitOrReturnId : race the runner, hand off terminal / timeout.
- renderTerminal : switch on QueryResult.Rows / Explain variant.
- runningSnapshot : format the RUNNING response (was smell #1).
Each new method is 5-15 lines with one job.
8. Trivial @provides AsyncQueryExecutorService(...) that returned the
impl unchanged — replaced with a Guice bind() call inside configure().
Smell #6 (two-mode AsyncQueryExecutorServiceImpl driven by nullable
fields) is architectural; defer per the audit note. The pragmatic escape
around OpenSearch DI's runtime-type matching stays as-is.
Signed-off-by: Peng Huo <penghuo@amazon.com>
…etion().get()
Blocking `job.completion().toCompletableFuture().get(waitMillis, MS)` in the
REST transport thread leaked the reactive primitive to callers and pinned a
worker for the full `wait_for_completion_timeout`. Under load every submit
holding a slot until timeout starved the pool.
Replace with a sealed `Outcome { Terminal | Pending | Failed }` and a
non-blocking `QueryJob.awaitOutcome(Duration)`:
- `awaitOutcome` returns a `CompletionStage<Outcome>` that fires exactly once,
whichever event wins — runner completion or `completeOnTimeout`. Errors are
already unwrapped; non-`Exception` throwables re-throw so OOM etc. is not
downgraded to `Failed`.
- New `QueryJob.onTerminal(Runnable)` covers subscribers (retention, metrics)
that need only "runner is done" — no result consumption, no stage leak.
- `completion()` is deleted from the public surface; the underlying
`CompletableFuture` is fully encapsulated so no caller can invent its own
blocking pattern.
- `TransportPPLQueryAction.submitAsync` becomes callback-based. The listener
is wrapped with `ContextPreservingActionListener.wrapPreservingContext` so
the ThreadContext (security identity, trace headers) that submitted the
query is restored when the terminal callback fires on a JDK Delayer thread
or the runner-completing thread.
- `RetentionPolicy` switches from `completion().whenComplete(...)` to
`onTerminal(...)`.
Tests: 6 new `awaitOutcome_*` cases plus cancellation-as-Failed and
onTerminal-null-guard. QueryJobTest / RetentionPolicyTest / doctest /
AsyncPPLQueryLifecycleIT / AsyncPPLMultiNodeRoutingIT all pass.
Signed-off-by: Peng Huo <penghuo@amazon.com>
QueryResult is now the wire-protocol response for a submission — Rows, Explain, or Running(id). The previous Outcome wrapper (Terminal | Pending | Failed) existed only to distinguish "no result yet" from "here's your result"; with Running as a first-class QueryResult variant, Outcome is redundant and deleted. QueryJob.await(Duration) returns CompletionStage<QueryResult> directly. Failures surface via the stage's error channel — the whenComplete's error arg maps 1:1 to ActionListener.onFailure, natural at the transport boundary. Errors are unwrapped inside QueryJob; Error rethrows so OOM etc. does not downgrade to Failed. TransportPPLQueryAction.submitAsync becomes a single exhaustive switch on QueryResult. runningSnapshot(QueryJob) — a hand-rolled JSON builder — is replaced by formatRunning(QueryResult.Running) that consumes the typed protocol record. The wire body is unchanged; the format is now anchored on the type rather than an ad-hoc function. Verbose comments on QueryJob's private lifecycle transitions removed — the method names and code carry the meaning. Tests: QueryJobTest migrated to await_* cases checking QueryResult variants and CompletionException.getCause() for failure paths. doctest, AsyncPPLQueryLifecycleIT (5), AsyncPPLMultiNodeRoutingIT (3) all pass. Signed-off-by: Peng Huo <penghuo@amazon.com>
Scope-down: keep the neutral state-machine cancel (QueryJob#cancel, QueryJobService#cancel, QueryJobState.CANCELLED, QueryRunner#cancel) — it is essential to the design and stays. Only the transport-layer DELETE integration with QueryJobService is removed; follow-up PR will restore it. - AsyncQueryExecutorServiceImpl#cancelQuery no longer short-circuits QueryJobId-shaped ids to queryJobService.cancel(...). All DELETE traffic flows through the existing Spark path unchanged. - TransportCancelAsyncQueryRequestAction reverted to its pre-routing constructor (transportService + ActionFilters + AsyncQueryExecutorService), dropping ClusterService and owner-node forwarding. - Two integration tests dropped: - AsyncPPLQueryLifecycleIT.async_cancelledJobReportsCancelled - AsyncPPLMultiNodeRoutingIT.delete_forwardsFromNonOwnerNodeToOwner Both exercised the DELETE-API integration; their helper delete() methods are removed with them. - TransportCancelAsyncQueryRequestActionTest constructor call adjusted. doctest / AsyncPPLQueryLifecycleIT (4) / AsyncPPLMultiNodeRoutingIT (2) all pass. Signed-off-by: Peng Huo <penghuo@amazon.com>
Feature-gate the async submit path so callers get a 400 instead of silently falling through to the sync pipeline when their request cannot be run asynchronously. Enforced at the top of TransportPPLQueryAction#doExecute via a new supportsAsync(request) predicate. Rejected when isAsync() is true and any of the following holds: - isExplainRequest() — the _plugins/_ppl/_explain REST endpoint. - analyze() — the _plugins/_ppl/_analyze REST endpoint. - profile() — profile queries collect live execution metrics that don't survive the async submit/poll round trip. - unifiedQueryHandler.isAnalyticsIndex(request, PPL) — analytics-engine indices route to a separate pipeline that never touches QueryJobService. - Response format is not JDBC — the running-snapshot body is JSON-shaped, so csv / raw / viz can't render a consistent pending response. Statement-level explain (query text starts with "explain ...") is untouched: it flows through the normal PPL execute path and terminates as QueryResult.Explain — this PR's async-explain support stays intact. IT coverage: async_rejectedForExplainEndpoint, async_rejectedForCsvFormat in AsyncPPLQueryLifecycleIT. Signed-off-by: Peng Huo <penghuo@amazon.com>
…ailure routing, terser error - Rename queryFormatter → responseListener. The variable is an ActionListener wrapping the response formatter; the previous name implied a formatter object with a .format() method. - Reorder: build ctxListener + responseListener before queryJobService.submit so both the submit-throw and the await-complete failure paths route through the same listener chain (ThreadContext preservation + response formatting). - Error message: drop the "with jdbc format" specificity per review; add the actionable "Please remove wait_for_completion_timeout and retry." Signed-off-by: Peng Huo <penghuo@amazon.com>
PR feedback: retention was hardcoded to 5 minutes regardless of the request's keep_alive value. Now the caller's keep_alive flows through submission and drives the eviction TTL; default applies only when the caller omits the field. - QueryJobService.submit gains a Duration keepAlive parameter; non-null, positive. Service arms retention with the caller-supplied TTL. - RetentionPolicy.arm(job, ttl) takes per-job TTL. The policy no longer carries a fixed default; validation moves from constructor to arm(). - PPLQueryRequest.effectiveKeepAlive() resolves to caller value or the new DEFAULT_KEEP_ALIVE (5 min). TransportPPLQueryAction passes this to submit(). Matches how effectiveWaitForCompletion() already works. - OpenSearchPluginModule drops the RETENTION_TTL constant and its associated javadoc; RetentionPolicy Provides no longer takes a TTL arg. - SQLPlugin, PPLQueryRequestFactory: drop explanatory comments per review. - endpoint.md: keep_alive row now describes actual per-request behavior; MVP wording removed. Signed-off-by: Peng Huo <penghuo@amazon.com>
Per PR feedback, only keep javadoc changes forced by actual API changes.
Restored the upstream comments on the constructor, accessors (id, owner,
status), cancel(), startRunner(), and the private lifecycle transitions
(onRunnerSuccess, onRunnerFailure, safeCancelRunner, unwrap).
Retained changes are all code-driven:
- Class-level thread-safety block: dead reference to #completion() replaced
with pointers to #await(Duration) / #onTerminal(Runnable); "four
accessors" → "five".
- New await(Duration) method with its javadoc (replaces completion()).
- New onTerminal(Runnable) method with its javadoc.
- cancel(): {@link #completion()} → "the internal future" (link target
removed).
- unwrap(): private → public (used by TransportPPLQueryAction).
Signed-off-by: Peng Huo <penghuo@amazon.com>
…ormatter Two PR-comment fixes. 1) AsyncExecutorServiceModule: transport action ctors now take the AsyncQueryExecutorService interface, not the AsyncQueryExecutorServiceImpl. The concrete-type parameter was left over from an earlier iteration that needed impl-specific access; the current bodies only call interface methods (createAsyncQuery, getAsyncQueryResults, cancelQuery). - @provides returns the interface; the concrete Provider goes away. - Explicit bind(AsyncQueryExecutorService.class).to(...) becomes unnecessary and is removed. - TransportCreate/Get/Cancel*Action ctors switch to the interface; AsyncQueryExecutorServiceImpl import dropped from each. 2) ExplainResponseJsonFormatter: extracted into the protocol module as a proper JsonResponseFormatter<ExplainResponse> subclass. Both callers use the shared class: - TransportPPLQueryAction.createExplainResponseListener: the ~15-line anonymous JsonResponseFormatter body is replaced by `new ExplainResponseJsonFormatter(PRETTY)`. - AsyncQueryExecutorServiceImpl.toAsyncResponse: switches from the ad-hoc async-query-core copy to the shared formatter. - async-query-core/.../ExplainResponseJsonFormatter.java deleted. Signed-off-by: Peng Huo <penghuo@amazon.com>
…erage via existing ITs
- PR feedback cleanup for the DI module and the explain formatter:
* TransportPPLQueryAction: unwrap CompletionException on the whenComplete error path so
OpenSearch's status mapper sees the engine's original exception type (preserves 4xx
classification and PPL error context).
* AsyncExecutorServiceModule / Transport{Create,Get,Cancel}AsyncQueryRequestAction: moved
the DI module explain-formatter work and consolidated the ad-hoc async-query-core copy to
protocol/.../ExplainResponseJsonFormatter — one shared JsonResponseFormatter<ExplainResponse>
used by both the sync and the async-GET-via-Spark render paths.
* AsyncQueryResultResponseFormatter: emit schema/datarows for the neutral SUCCEEDED status in
addition to the legacy "success"; previously the formatter stripped them on cross-node GET.
* AsyncQueryExecutorServiceImpl#getAsyncQueryResults: translate QueryJobNotFoundException ->
AsyncQueryNotFoundException so REST maps a missing / evicted queryId to 4xx (client error)
instead of defaulting to 5xx.
- AsyncPPLIT: new JUnit Suite that re-runs a random sample of Calcite PPL ITs under the async
submit path. Pool is 122 classes in org.opensearch.sql.calcite.remote; SAMPLE_SIZE=5; the
selection uses the OpenSearch gradle `tests.seed` for reproducibility. The suite's
@BeforeClass sets `test.ppl.async_wait_for_completion_timeout=60s`, so every re-executed test
flows through TransportPPLQueryAction.submitAsync and asserts the same schema / rows the
main integTest already verifies in sync mode. Verified by temporary breakage: dropping one
row in the Rows render path turned 102/104 sampled test methods red with
"expected:<N> but was:<N-1>", confirming end-to-end async coverage; revert passes 11/11.
- PPLIntegTestCase.buildRequest: when the `test.ppl.wait_for_completion_timeout` system
property is set and the endpoint isn't `/_explain` / `/_analyze`, inject
`wait_for_completion_timeout` into the request body via raw String.format (preserving the
existing JSON-escape convention for pre-formatted query strings).
- integ-test/build.gradle: asyncMultiNodeIT task now sets project.root +
defaultQuerySizeLimit/BucketSize/MaxResultWindow so AsyncPPLMultiNodeRoutingIT reuses the
real accounts dataset.
- AsyncPPLQueryLifecycleIT rewrite (11 tests, up from 6):
* sync + async submit-shape assertions use MatcherUtils (verifySchema, verifyDataRows,
verifyNumOfRows).
* strict SUCCEEDED on the terminal fetch.
* runner-wins-race: wait_for_completion_timeout=30s returns sync-shape, no id.
* explain polling now parses JSON and checks `has("calcite")|has("root")` instead of
substring contains.
* sync-only-gate rejection coverage expanded to _explain, _analyze, profile flag, csv format.
* 4xx on unknown queryId.
* keep_alive drives retention: TTL=3s submit then poll until the 4xx evicted response.
- AsyncPPLMultiNodeRoutingIT: swapped the ad-hoc inline index for `loadIndex(Index.ACCOUNT)`;
verifies schema + rows via MatcherUtils (was: only status).
- Shared AsyncPPLTestHelpers removes POST/GET/pollUntilTerminal duplication between the two
async ITs.
Signed-off-by: Peng Huo <penghuo@amazon.com>
… be async PR feedback: rejecting with 400 when a sync-only request shape carries wait_for_completion_timeout is nitpicky. Lean API now ignores the field and runs the request synchronously. Covers the _plugins/_ppl/_explain endpoint, csv/raw/viz response formats, profile=true, and analytics-engine indices. TransportPPLQueryAction clears waitForCompletionTimeout and keepAlive on the transformed request so the subsequent dispatch takes the sync branch. No behavior change for well-formed async requests. Tests: three AsyncPPLQueryLifecycleIT rejection cases flipped to assert the sync-shape response is returned instead of 4xx (explain plan tree, csv body carrying the count, profile schema + datarows). Signed-off-by: Peng Huo <penghuo@amazon.com>
…yRequest.shouldRunAsync Replace the TransportPPLQueryAction top-level mutation + supportsAsync predicate pair with a single PPLQueryRequest#shouldRunAsync() method. PPLQueryRequest.isAsync() answered "caller declared async intent" (two nullable fields). The transport then had a second predicate supportsAsync() for "the request shape is compatible with async" and mutated the request to clear async fields on mismatch. Two concepts, mutation, and dead ambiguity if a caller later inspected isAsync(). Fold both into a single, well-named predicate: shouldRunAsync() returns true iff the caller declared async intent AND the shape is compatible. Transport dispatch reads at one place, no mutation. - PPLQueryRequest#isAsync -> shouldRunAsync; the body merges the sync-only checks (explain endpoint, analyze endpoint, profile flag, non-JDBC format) into the predicate. - TransportPPLQueryAction: delete the top-level async-gate block, delete supportsAsync; the dispatch "else if (shouldRunAsync())" handles the decision; sync-only shapes fall through to pplService.execute. - AsyncPPLQueryLifecycleIT.async_keepAliveEvictsAfterTtl: keep_alive=1s, terminal poll ceiling 5s, eviction poll ceiling 3s (was 3s / 15s / 15s). Signed-off-by: Peng Huo <penghuo@amazon.com>
…line
Address 5 correctness defects raised in review (P1-2, P1-3, P2-5, P2-6,
P2-7). P1-1 is a security-plugin role mapping (separate repo); documented
here. P2-4 (wait budget covers runner preparation) is a larger refactor
tracked for follow-up.
P1-2 Enforce the advertised time-value bounds.
PPLQueryRequestFactory now rejects wait_for_completion_timeout
outside [0s, 60s] and keep_alive outside (0, 24h]. Values like
keep_alive=365d (bypassing retention) or wait=-1 return 400.
P1-3 Discard retained state when the runner wins the submit race.
TransportPPLQueryAction renders Rows/Explain inline when the budget
covers runner completion; no id is returned so no poll will ever
reach for the job. Previously the job (and its result rows) stayed
pinned in InMemoryQueryJobStore for the full keep_alive, leaking
heap on every "fast" async submit.
Fix: new QueryJobService#discard(job) removes the entry; Transport
calls it on the inline-terminal paths. Retention timer still fires
but its conditional store.remove is a no-op.
P2-5 Preserve 404 status across owner-node forwarding.
TransportGetAsyncQueryResultAction catches
QueryJobNotFoundException and rethrows as
org.opensearch.ResourceNotFoundException (OpenSearchException with
status NOT_FOUND). AsyncQueryExecutorServiceImpl no longer wraps
itself. Previously the neutral RuntimeException crossed the
transport as NotSerializableExceptionWrapper mapped to 500.
P2-6 Align datasource feature gate with PPL async access.
plugins.query.datasources.enabled governs the Spark stack, not
neutral PPL async jobs. RestAsyncQueryManagementAction skips the
gate for GET/DELETE whose queryId parses as QueryJobId.
P2-7 Translate ownership denials to 403.
Same catch block translates QueryJobForbiddenException to
OpenSearchStatusException(FORBIDDEN), symmetric with the not-found
mapping. A caller authorized for the GET action but not owning the
job now gets 403 instead of 500.
Tests
- OpenSearchQueryJobServiceTest#discard_removesJobFromStore (P1-3)
- TransportGetAsyncQueryResultActionTest:
queryJobNotFound_translatesToResourceNotFoundException (P2-5),
queryJobForbidden_translatesToOpenSearchStatusForbidden (P2-7).
- RestAsyncQueryManagementActionTest: 5 x isPplRequest predicate
cases + pplGet_passesWhenDataSourcesDisabled (P2-6).
- AsyncPPLQueryLifecycleIT: 3 x bounds rejection (P1-2).
- AsyncPPLMultiNodeRoutingIT: getUnknownPplIdFromNonOwner_returns404
(P2-5 end-to-end cross-node).
Docs: endpoint.md now documents bounds + rejection, the sync fallthrough
for sync-only shapes, the datasources gate's narrower scope, and the
required cluster permissions for callers using custom security roles.
Signed-off-by: Peng Huo <penghuo@amazon.com>
Address review feedback: the async section was implementation-oriented and the examples were pseudo-code. Rewrite as user documentation. - Description paragraph focused on what callers get back, not internals. - Parameters table kept, bounds clearly called out. - Limitations section: sync-only shapes (explain, profile, non-JSON formats), retention cost awareness, and the three required cluster permissions. - Example replaces the pseudo "POST /_plugins/_ppl + arrow" blocks with a real curl invocation and matching JSON response — now exercised by the doctest suite. Verified by temporarily corrupting the expected count (4 -> 999): doctest failed on the mismatch, confirming the example is executed end-to-end. Reverted; doctest 86/86 green. Signed-off-by: Peng Huo <penghuo@amazon.com>
Per feedback — the 5s default was too eager for typical interactive queries, forcing clients to poll even when the runner would have finished within a reasonable submit response. 30s matches the common 'wait up to ~minute' expectation for synchronous submission and stays within the documented 60s cap. Signed-off-by: Peng Huo <penghuo@amazon.com>
P1-1: cross-node GET lost queryId. GetAsyncQueryResultActionRequest had no writeTo override and its StreamInput ctor did not read queryId, so owner-node forwarding delivered a null id and returned 500 instead of the terminal snapshot. P1-2: discard did not actually release the job. RetentionPolicy's scheduled eviction lambda captured QueryJob + QueryJobStore, and the scheduler's DelayedQueue held that lambda for the full keep_alive. The store entry was gone after discard but result rows stayed pinned in old-gen until the TTL fired. RetentionPolicy now tracks the pending ScheduledCancellable per job and exposes disarm(jobId); discard cancels it before removing from the store so GC can reclaim the job immediately. AsyncPPLMultiNodeRoutingIT was green under the pre-fix code because getClusterHosts() returns both IPv4 and IPv6 bindings per node, so hosts[0]/hosts[1] were the same physical node and forwarding never triggered. Probe /_nodes/_local on each host to pick two with distinct node ids. With that fixed, reverting the serialization change made the IT fail with 500 as expected, confirming the regression is now caught. Wire asyncMultiNodeIT into check.dependsOn so CI exercises the two-node routing path by default. Signed-off-by: Peng Huo <penghuo@amazon.com>
Signed-off-by: Peng Huo <penghuo@gmail.com>
Signed-off-by: Peng Huo <penghuo@gmail.com>
Signed-off-by: Peng Huo <penghuo@gmail.com>
Format statement-level explain results in TransportGetAsyncQueryResultAction instead of AsyncQueryExecutorServiceImpl. AsyncQueryExecutionResponse now carries the core ExplainResponse, so async-query-core no longer needs the protocol module (which transitively pulls in the opensearch module). Signed-off-by: Peng Huo <penghuo@gmail.com>
Issue: a job that failed after wait_for_completion_timeout had its error collapsed to a bare type + reason string on GET, dropping code / stage / context / location / suggestion. Clients polling the GET endpoint lost the same diagnostic they got from a sync POST or an inline async failure. Capture the full sync-shape error at QueryJob.onRunnerFailure time via a pluggable renderer: - Core: QueryFailure gains a Map<String,Object> details field, defensively copied. QueryJob accepts a Function<Throwable, Map> renderer and calls it from onRunnerFailure; renderer exceptions are swallowed so the state machine always reaches FAILED. - Opensearch: OpenSearchQueryJobService threads the renderer through to QueryJob. ErrorMessage.getErrorAsJson() is made public so the renderer can reuse the sync REST error object without the outer HTTP status envelope. - Plugin: new SyncErrorReportRenderer extracts the status-mapping and structured-error rendering that was private to RestPPLQueryAction. The sync REST handler now calls the shared helper; its behavior is unchanged. OpenSearchPluginModule wires the renderer into the job service. - Async-query-core / async-query: AsyncQueryExecutionResponse and AsyncQueryResult gain a nullable errorDetails map, carried through to the formatter. AsyncQueryResultResponseFormatter emits error as an object when errorDetails is non-null; the existing string branch is preserved so Spark output stays byte-identical (every Spark construction site passes null). ITs: AsyncPPLQueryLifecycleIT.parity_* exercise all three request shapes (pure sync POST, inline async wait=30s, GET after wait=0 + poll) and assert deep structural equality against the sync baseline, mirroring the scenarios in CalciteErrorReportStageIT. AsyncPPLMultiNodeRoutingIT confirms the full error body also survives cross-node forwarding to the non-owner. Signed-off-by: Peng Huo <penghuo@gmail.com>
Release notes are published out-of-band on release; not needed in this PR. Signed-off-by: Peng Huo <penghuo@gmail.com>
Add an IT that sets plugins.ppl.query.timeout to a sub-millisecond budget, submits a long-running self-join query with wait_for_completion_timeout=0, polls the GET endpoint, and asserts the terminal body deep-equals the sync body for the same query. Guards against regressions that would leave a timed-out job stuck in RUNNING without retention attached. The sync timeout works via Thread.interrupt() on the engine thread, which is cooperative. CPU-bound hangs and dropped-signal cases are tracked for a submit-time max-lifetime backstop in issue opensearch-project#5801. Signed-off-by: Peng Huo <penghuo@gmail.com>
Test name alone is enough; the sync-vs-async parity is already explained on assertAsyncErrorMatchesSync. Signed-off-by: Peng Huo <penghuo@gmail.com>
With a 1ms budget the interrupt races against Calcite's rule-by-rule planner — the exact rule that observes Thread.interrupted() is non-deterministic, so sync and async GET bodies can carry different 'details' strings (one hit LimitIndexScanRule, the other hit EnumerableTopKMergeRule). Keep the stable guarantees — terminal FAILED state, structured error object with non-empty type and reason — and drop the field-by-field deep equal check. Deep parity is already covered by the deterministic parity_* tests (FIELD_NOT_FOUND, IndexNotFound, alias-with-suggestion). Signed-off-by: Peng Huo <penghuo@gmail.com>
Without Calcite the legacy v2 analyzer throws SemanticCheckException synchronously from pplService.execute for field-not-found; the exception leaves startRunner() through the catch block before await(0) returns, so the submit stage completes exceptionally and the async POST returns 400 inline instead of a RUNNING id. The cross-node forwarding test then fails when postPpl throws ResponseException. Calcite delivers the same error through the response listener on the engine thread, so submit returns a RUNNING id and the GET body carries the structured sync-shape error — matching the single-node lifecycle IT which already calls enableCalcite(). Signed-off-by: Peng Huo <penghuo@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Adds opt-in async PPL submission through
POST /_plugins/_pplusingwait_for_completion_timeoutandkeep_alive, with polling through the existingGET /_plugins/_async_query/{id}endpoint. Queries that finish within the wait return the existing synchronous response shape.QueryJobService.submit(runner, principal, waitForCompletion, keepAlive)now returnsCompletionStage<QueryResult>and owns the submit race and retention decision:RUNNINGoutcome exposes a polling ID and attaches retention after terminal completion. Completion before retention registration is handled by the job's terminal callback.The transport action only formats the submission outcome.
RetentionPolicyonly schedules terminal-result eviction; the publicdiscard()API,disarm()operation, and pending-timer registry are removed. This eliminates cancellation races between inline responses and timer creation. Owner-node forwarding, transport-safe 403/404 errors, datasource-gate handling, and request bounds are included.This is an alternative to #16 on a separate branch, targeting
main. Implements opensearch-project/sql#5765.Verification
All six module unit-test suites passed: 9,522 passed, 46 skipped, zero failures.
Async PPL lifecycle integration tests: 15/15 passed, including inline explain output, inline client errors, polling, retention expiry, bounds rejection, and synchronous fallthrough.
./gradlew -DignorePrometheus :integ-test:integTest --tests 'org.opensearch.sql.ppl.AsyncPPLQueryLifecycleIT' --max-workers=4Distinct-node routing integration tests: 3/3 passed, covering result forwarding, explain forwarding, and forwarded missing-job 404 responses. The task is included in
check.Spotless validation for all 37 Java files changed by this PR and
git diff --checkpassed. The two async module suites were rerun successfully after formatting. Independent review found no blocking issues.Compatibility and follow-ups
QueryJobService.submitnow supply both durations and consume a submission stage rather than a mutable job.Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.