[analytics engine] Add /_plugins/_analytics/stats endpoint

[OVI3D0]

Description

Builds ontop of #21660 and adds per stage timing but rolled up so users won't need to hit _explain for every request

Example:

  $ curl -s
  'http://localhost:9200/_plugins/_analytics/stats' | jq
  {
    "analytics": {
      "queries": {
        "total": 46,
        "succeeded": 46,
        "failed": 0,
        "in_flight": 0,
        "elapsed_ms":  { "count": 46, "sum_ms": 633, "max_ms": 419, "p50_ms": 4, "p95_ms": 14, "p99_ms": 419 },
        "planning_ms": { "count": 46, "sum_ms": 437, "max_ms": 223, "p50_ms": 2, "p95_ms": 46, "p99_ms": 223 }
      },
      "stages_by_type": {
        "SHARD_FRAGMENT": {
          "started": 46, "succeeded": 46, "failed": 0, "cancelled": 0,
          "rows_processed_total": 334,
          "elapsed_ms": { "count": 46, "sum_ms": 649, "max_ms": 415, "p50_ms": 4, "p95_ms": 15, "p99_ms": 415 }
        }
      },
      "fragments": {
        "total": 90, "succeeded": 90, "failed": 0,
        "elapsed_ms": { "count": 90, "sum_ms": 507, "max_ms": 291, "p50_ms": 2, "p95_ms": 5, "p99_ms": 291 }
      }
    }
  }

Some follow ups that can be added:

• TransportNodesAction to broadcast/merge results, since right now this API is per-node only
• Planning-time failures don't increment queries.failed — query lifecycle events fire from QueryScheduler.execute(), but parse / table-resolution / planning failures fail upstream of that

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue The executeInternal method always calls executeWithProfile regardless of the profile parameter value. When profile is false, the profile data is discarded after execution, but the profiling overhead (capturing stage timings, building the profile snapshot) is still incurred. This contradicts the intent of having separate code paths and wastes resources on non-profiling queries. ActionListener<Iterable<Object[]>> rowsListener = profile ? buildProfilingRowsListener(execRef, context, fullPlan, planningTimeMs, listener) : ActionListener.wrap(rows -> listener.onResponse(new ProfiledResult(rows, null, null)), listener::onFailure); Possible Issue The executeFragmentStreamingAsync method increments rows by batch.getRowCount() but never calls responseHandler.onBatch(batch) inside the loop. The batch is read but not sent to the handler, so the streaming response will be empty. The handler only receives onComplete() with no data. while (it.hasNext()) { EngineResultBatch batch = it.next(); rows += batch.getRowCount(); responseHandler.onBatch(batch); Possible Issue The stageTypeByStageId map is used to route terminal stage events to the correct bucket, but it is never cleaned up when a stage completes. The map will grow unbounded over the lifetime of the node, leaking memory proportional to the number of stages executed. The comment claims entries are pruned on terminal states, but currentTypeForStage only removes the entry—it does not prevent accumulation if the same stageId is reused across queries. private final ConcurrentMap<Integer, String> stageTypeByStageId = new ConcurrentHashMap<>(); private String currentTypeForStage(int stageId) { String t = stageTypeByStageId.remove(stageId); return t != null ? t : "unknown"; }

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Prevent potential race condition NPE The failure path accesses execRef.get() twice without null-guarding the second access. If execRef.get() returns non-null but getGraph() returns null between the two calls (race condition), this could throw NullPointerException. Cache the result in a local variable to ensure consistent null checks. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/DefaultPlanExecutor.java [248-252] } catch (Exception e) { - QueryProfile qp = execRef.get() != null && execRef.get().getGraph() != null - ? QueryProfileBuilder.snapshot(execRef.get().getGraph(), context, fullPlan, planningTimeMs) + QueryExecution exec = execRef.get(); + QueryProfile qp = exec != null && exec.getGraph() != null + ? QueryProfileBuilder.snapshot(exec.getGraph(), context, fullPlan, planningTimeMs) : new QueryProfile(context.queryId(), java.util.List.of(), planningTimeMs, 0L, java.util.List.of()); listener.onResponse(new ProfiledResult(null, e, qp)); }); Suggestion importance[1-10]: 7 __ Why: Valid concern about accessing execRef.get() twice without caching. While the race condition is unlikely in practice (the AtomicReference is set once and never cleared), caching the result in a local variable improves code clarity and eliminates any theoretical race condition risk.	Medium
General	Prevent unbounded map growth The stageTypeByStageId map grows unbounded if stages never reach terminal states (e.g., stuck queries, cancellation failures). This creates a memory leak. Consider using a bounded cache with eviction policy or adding explicit cleanup on query-level terminal events to prevent unbounded growth. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/stats/AnalyticsStatsCollector.java [152-156] -private final ConcurrentMap<Integer, String> stageTypeByStageId = new ConcurrentHashMap<>(); +// Use a bounded cache with size limit and LRU eviction +private final Map<Integer, String> stageTypeByStageId = Collections.synchronizedMap( + new LinkedHashMap<Integer, String>(128, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<Integer, String> eldest) { + return size() > 1024; // Limit to 1024 in-flight stages + } + } +); private String currentTypeForStage(int stageId) { String t = stageTypeByStageId.remove(stageId); return t != null ? t : "unknown"; } Suggestion importance[1-10]: 6 __ Why: The suggestion identifies a legitimate memory leak concern where stageTypeByStageId could grow unbounded if stages never reach terminal states. The proposed bounded cache solution is reasonable, though the specific implementation using LinkedHashMap with synchronization may not be ideal for concurrent access patterns already using ConcurrentHashMap.	Low
	Prevent stack overflow on deep DAGs The recursive findStageById traversal has no depth limit and could cause stack overflow on deeply nested DAGs (e.g., pathological query plans with hundreds of stages). Convert to iterative traversal using a queue to prevent stack exhaustion on malformed or adversarial queries. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/profile/QueryProfileBuilder.java [116-123] private static Stage findStageById(Stage root, int stageId) { - if (root.getStageId() == stageId) return root; - for (Stage child : root.getChildStages()) { - Stage found = findStageById(child, stageId); - if (found != null) return found; + java.util.Queue<Stage> queue = new java.util.LinkedList<>(); + queue.add(root); + while (!queue.isEmpty()) { + Stage current = queue.poll(); + if (current.getStageId() == stageId) return current; + queue.addAll(current.getChildStages()); } return null; } Suggestion importance[1-10]: 4 __ Why: While converting to iterative traversal is a valid defensive programming practice, the concern about stack overflow on deeply nested DAGs is theoretical. Real-world query plans rarely exceed a few dozen stages, making stack overflow extremely unlikely. The iterative solution adds complexity without addressing a practical problem.	Low
	Avoid unnecessary wrapper allocation When profiling is disabled, the non-profile path still wraps rows in a ProfiledResult with null profile. This creates an unnecessary object allocation for every non-profiled query. Consider passing listener directly to avoid the wrapper allocation overhead. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/DefaultPlanExecutor.java [210-212] ActionListener<Iterable<Object[]>> rowsListener = profile ? buildProfilingRowsListener(execRef, context, fullPlan, planningTimeMs, listener) : ActionListener.wrap(rows -> listener.onResponse(new ProfiledResult(rows, null, null)), listener::onFailure); +ActionListener<Iterable<VectorSchemaRoot>> batchesListener = ActionListener.runAfter( + ActionListener.wrap(batches -> { + Iterable<Object[]> rows = batchesToRows(batches); + if (profile) { + rowsListener.onResponse(rows); + } else { + listener.onResponse(new ProfiledResult(rows, null, null)); + } + }, profile ? rowsListener::onFailure : listener::onFailure), + () -> taskManager.unregister(queryTask) +); + Suggestion importance[1-10]: 3 __ Why: The suggestion correctly identifies that the non-profile path creates a ProfiledResult wrapper, but the proposed solution is more complex and doesn't eliminate the allocation—it just moves it. The current code is clearer and the allocation overhead is negligible for the non-profile path.	Low

[github-actions]

❌ Gradle check result for 74d5d76: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?