Overriding profile endpoint with analyze endpoint with operator tree and profiling

[Krish-Gandhi]

Description

• Introduces the analyze endpoint for PPL queries, which can be activated by passing "analyze": true as a request body parameter.
• Overrides the existing profile endpoint to run analyze functionality.
• Propagate PPLanalyze flag through transport and request parsing.
• Combines logical plan and physical plan nodes by walking each tree and grouping nodes into an operator_tree.
• Maps PPL query segments and estimated/actual rows to operator_tree nodes.
• Combined current profile plan response with operator_tree nodes to calculate time taken by each node.

---

Important

This PR overrides the existing profile endpoint by routing all requests with either "analyze": true or "profile": true (or both) to pplService.analyze() in TransportPPLQueryAction.java. For current end-users of the profile endpoint, this will make little difference, as the response from profile is a subset of the response from analyze. In simpler terms, current end-users can make no changes and have similar results.

All of the existing code for profile still exists and is in place. This makes it extremely simple to separate analyze and profile again. This can be done by removing the || transformedRequest.profile() part of the boolean expression in line 200 of TransportPPLQueryAction.java.

plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java:193-209:

    if (transformedRequest.isExplainRequest()) {
      pplService.explain(
          transformedRequest, createExplainResponseListener(transformedRequest, clearingListener));
    /**
     * Removing  `|| transformedRequest.profile()` from line 200 will
     * separate the `profile` and `analyze` endpoints. See PR #5568.
     */
    } else if (transformedRequest.analyze() || transformedRequest.profile()) {
      pplService.analyze(
        transformedRequest, createAnalyzeResponseListener(transformedRequest, clearingListener));
    } else {
      pplService.execute(
          transformedRequest,
          createListener(transformedRequest, clearingListener),
          createExplainResponseListener(transformedRequest, clearingListener));
    }
  }

---

Example Query and Response

The following curl command will run the query source=accounts | where age < 30 | eval full_name = firstname + \" \" + lastname | fields full_name, email, age on the analyze endpoint:

curl -X POST "localhost:9200/_plugins/_ppl" \
  -H "Content-Type: application/json" \
  -d '{"query": "source=accounts | where age < 30 | eval full_name = firstname + \" \" + lastname | fields full_name, email, age", "analyze": true}'

The response of this will be as follows. (NOTE: The "logicalPlan" and "physicalPlan" fields are included for debugging purposes and should not be included in the final version of this endpoint.)

{
  "query": "source=accounts | where age < 30 | eval full_name = firstname + \" \" + lastname | fields full_name, email, age",
  "logicalPlan": [
    "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT]): rowcount = 5000.0, cumulative cost = {114000.0 rows, 145000.0 cpu, 0.0 io}, id = 14855",
    "LogicalProject(full_name=[||(||($0, ' '), $4)], email=[$3], age=[$2]): rowcount = 5000.0, cumulative cost = {109000.0 rows, 25000.0 cpu, 0.0 io}, id = 14854",
    "LogicalFilter(condition=[<($2, 30)]): rowcount = 5000.0, cumulative cost = {104000.0 rows, 10000.0 cpu, 0.0 io}, id = 14852",
    "CalciteLogicalIndexScan(table=[[OpenSearch, accounts]]): rowcount = 10000.0, cumulative cost = {99000.0 rows, 0.0 cpu, 0.0 io}, id = 14851"
  ],
  "physicalPlan": [
    "EnumerableCalc: rowcount = 1.0, cumulative cost = {2.0 rows, 2.0 cpu, 0.0 io}, id = 14948",
    "CalciteEnumerableIndexScan: rowcount = 1.0, cumulative cost = {1.0 rows, 1.0 cpu, 0.0 io}, id = 14947"
  ],
  "profile": {
    "summary": {
      "total_time_ms": 761.06
    },
    "phases": {
      "analyze": {
        "time_ms": 1.6
      },
      "optimize": {
        "time_ms": 6.76
      },
      "execute": {
        "time_ms": 752.48
      },
      "format": {
        "time_ms": 0.18
      }
    },
    "plan": {
      "node": "EnumerableCalc",
      "time_ms": 751.68,
      "rows": 3,
      "children": [
        {
          "node": "CalciteEnumerableIndexScan",
          "time_ms": 751.51,
          "rows": 3
        }
      ]
    }
  },
  "operator_tree": [
    {
      "source": "source=accounts | where age < 30",
      "node_type": [
        "SearchFrom",
        "WhereCommand"
      ],
      "description": [
        "CalciteLogicalIndexScan(table=[[OpenSearch, accounts]]): rowcount = 10000.0, cumulative cost = {99000.0 rows, 0.0 cpu, 0.0 io}, id = 14851",
        "LogicalFilter(condition=[<($2, 30)]): rowcount = 5000.0, cumulative cost = {104000.0 rows, 10000.0 cpu, 0.0 io}, id = 14852"
      ],
      "estimated_rows": 5000,
      "actual_time_ms": "751.51 ms",
      "actual_rows": 3,
      "is_pushed_down": true
    },
    {
      "source": "eval full_name = firstname + \" \" + lastname | fields full_name, email, age",
      "node_type": [
        "EvalCommand",
        "FieldsCommand"
      ],
      "description": [
        "LogicalProject(full_name=[||(||($0, ' '), $4)], email=[$3], age=[$2]): rowcount = 5000.0, cumulative cost = {109000.0 rows, 25000.0 cpu, 0.0 io}, id = 14854"
      ],
      "estimated_rows": 5000,
      "actual_time_ms": "0.17 ms",
      "actual_rows": 3
    }
  ],
  "recommendations": [],
  "schema": [
    {
      "name": "full_name",
      "type": "VARCHAR"
    },
    {
      "name": "email",
      "type": "VARCHAR"
    },
    {
      "name": "age",
      "type": "BIGINT"
    }
  ],
  "datarows": [
    [
      "Jane Smith",
      "jane@example.com",
      28
    ],
    [
      "Kyle Miller",
      "goat@example.com",
      22
    ],
    [
      "Joette Kap",
      "coast2coast@example.com",
      22
    ]
  ],
  "total": 3,
  "size": 3
}

Performance of analyze

After writing a benchmarking script to run a query on a sample dataset, the results were as follows:

=== PPL Query Benchmark ===
Query: source=opensearch_dashboards_sample_data_flights | where DistanceMiles > 1000 | stats avg(FlightTimeMin) as avg_time, count() as cnt by DestCountry
  | sort - cnt | head 10
Iterations: 100

--- Results (after 3 warmup runs each) ---

normal      avg=1.3ms  p50=1.2ms  p95=2.0ms  min=0.9ms  max=2.3ms
profile     avg=1.6ms  p50=1.6ms  p95=2.1ms  min=1.0ms  max=2.4ms
analyze     avg=1.6ms  p50=1.6ms  p95=2.1ms  min=0.9ms  max=3.1ms

Related Issues

#5500
Resolves #4343

Check List

• New functionality includes testing.
• New functionality has been documented.
• New functionality has javadoc added.
• New functionality has a user manual doc added.
• New PPL command checklist all confirmed.
• API changes companion pull request created.
• Commits are signed per the DCO using --signoff or -s.
• Public documentation issue/PR created.

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

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 127dfe7)

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 latch.await() call blocks indefinitely if the executeWithCalcite callback never completes. If executeWithCalcite throws an exception synchronously before invoking the listener, or if the listener is never called due to an internal bug, the thread will hang forever. This can cause the analyze endpoint to become unresponsive under certain failure conditions. latch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); listener.onFailure(new RuntimeException("Interrupted while waiting for query execution", e)); return; } Possible Issue When tracking is enabled, the code assumes context.relBuilder.peek() is always valid after a child's accept returns. If a child's visit method leaves the builder stack empty or in an unexpected state, peek() will throw NoSuchElementException. This can occur if a child node's visitor does not push a RelNode onto the stack as expected. int idAfter = context.relBuilder.peek().getId(); Resource Leak The Hook.Closeable returned by Hook.PLAN_BEFORE_IMPLEMENTATION.addThread is closed in a try-with-resources block, but if OpenSearchRelRunners.run throws an exception before the hook fires, the physical plan may not be captured. The subsequent code assumes physicalRelRef.get() is non-null. If it remains null, physicalPlanRef.get().split will throw NullPointerException at line 341. try (Hook.Closeable closeable = Hook.PLAN_BEFORE_IMPLEMENTATION.addThread( obj -> { RelRoot relRoot = (RelRoot) obj; physicalRelRef.set(relRoot.rel); physicalPlanRef.set( RelOptUtil.toString(relRoot.rel, SqlExplainLevel.ALL_ATTRIBUTES)); })) { try (java.sql.PreparedStatement ignored = OpenSearchRelRunners.run(context, calcitePlan)) { } catch (java.sql.SQLException e) { throw new RuntimeException(e); } } Possible Issue The code computes pushedNodeCount = logicalDepth - physicalDepth and assumes this is non-negative. If the physical plan has more nodes than the logical plan (e.g., due to optimizer transformations that add nodes), pushedNodeCount becomes negative. The subsequent loop condition pushedLogicalNodes < pushedNodeCount will never be satisfied if pushedNodeCount is negative, causing pushedSegments to remain 0 even when pushdown occurred. int logicalDepth = getLinearDepth(logicalPlan); int pushedNodeCount = logicalDepth - physicalDepth; // log.info( // "buildOperatorTree: logicalDepth={}, physicalDepth={}, pushedNodeCount={}," // + " segments={}, exclusiveIds={}", // logicalDepth, // physicalDepth, // pushedNodeCount, // querySegments.size(), // exclusiveIds); // Walk segments bottom-up (they're already in bottom-up order) and greedily assign // them to the pushed group until we've accounted for all pushed logical nodes. // The LogicalSystemLimit added by convertToCalcitePlan counts toward the logical depth // but has no segment, so we only count nodes that appear in exclusiveIds. long pushedLogicalNodes = 0; int pushedSegments = 0; for (int idx = 0; idx < querySegments.size() && pushedLogicalNodes < pushedNodeCount; idx++) { Set<Integer> ids = idx < exclusiveIds.size() ? exclusiveIds.get(idx) : Set.of(); long planNodeCount = ids.stream().filter(idToDescription::containsKey).count(); pushedLogicalNodes += planNodeCount; pushedSegments++; }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 127dfe7

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Prevent peek on empty builder Check if context.relBuilder is empty before calling peek() after accept(). The accept() call might not push any nodes onto the builder, causing peek() to throw NoSuchElementException. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [223-225] int idBefore = context.relBuilder.size() > 0 ? context.relBuilder.peek().getId() : -1; RelNode result = unresolved.accept(this, context); -int idAfter = context.relBuilder.peek().getId(); +int idAfter = context.relBuilder.size() > 0 ? context.relBuilder.peek().getId() : -1; Suggestion importance[1-10]: 8 __ Why: The code calls context.relBuilder.peek() after accept() without checking if the builder is empty, which could throw NoSuchElementException. This is a potential runtime error that should be prevented.	Medium
	Add timeout to latch await Add a timeout to latch.await() to prevent indefinite blocking if the async execution never completes. Without a timeout, the thread could hang forever if the callback is never invoked due to an unexpected error path. core/src/main/java/org/opensearch/sql/executor/QueryService.java [283-290] -CountDownLatch latch = new CountDownLatch(1); - -executeWithCalcite( - plan, - queryType, - null, - new ResponseListener<>() { - @Override - public void onResponse(ExecutionEngine.QueryResponse response) { - ... - latch.countDown(); - } - - @Override - public void onFailure(Exception e) { - errorRef.set(e); - latch.countDown(); - } - }); - try { - latch.await(); + if (!latch.await(60, TimeUnit.SECONDS)) { + listener.onFailure(new RuntimeException("Query execution timed out after 60 seconds")); + return; + } } catch (InterruptedException e) { Thread.currentThread().interrupt(); listener.onFailure(new RuntimeException("Interrupted while waiting for query execution", e)); return; } Suggestion importance[1-10]: 7 __ Why: Adding a timeout to latch.await() prevents indefinite blocking if the async callback never completes. This is a valid defensive programming practice that improves robustness, though the system may have other safeguards in place.	Medium
General	Validate ID range before iteration Validate that idAfter >= idBefore before creating the range. If node IDs are not monotonically increasing or if the builder state is inconsistent, this could produce incorrect or negative ranges. core/src/main/java/org/opensearch/sql/executor/QueryService.java [227-229] -for (int id = idBefore + 1; id <= idAfter; id++) { - producedIds.add(id); +if (idAfter >= idBefore) { + for (int id = idBefore + 1; id <= idAfter; id++) { + producedIds.add(id); + } } Suggestion importance[1-10]: 6 __ Why: While validating idAfter >= idBefore adds safety, the logic assumes monotonically increasing IDs. The check prevents incorrect ranges but may indicate a deeper issue if the condition fails. This is a reasonable defensive check.	Low
	Restore profiling state after tracking Ensure QueryProfiling.noop() is properly restored after the tracking phase completes. If an exception occurs during the second phase, profiling state might remain in noop mode, affecting subsequent queries. core/src/main/java/org/opensearch/sql/executor/QueryService.java [305-313] QueryProfiling.noop(); -CalciteClassLoaderHelper.withCalciteClassLoader( - () -> { - CalcitePlanContext context = - CalcitePlanContext.create( - buildFrameworkConfig(), SysLimit.fromSettings(settings), queryType); - context.setTrackingEnabled(true); - RelNode relNode = analyze(plan, context); - ... - }, - QueryService.class); +try { + CalciteClassLoaderHelper.withCalciteClassLoader( + () -> { + CalcitePlanContext context = + CalcitePlanContext.create( + buildFrameworkConfig(), SysLimit.fromSettings(settings), queryType); + context.setTrackingEnabled(true); + RelNode relNode = analyze(plan, context); + ... + }, + QueryService.class); +} finally { + QueryProfiling.current(); +} Suggestion importance[1-10]: 5 __ Why: The suggestion to restore profiling state in a finally block is reasonable, but the improved_code calls QueryProfiling.current() which may not properly restore the previous state. The concern is valid but the solution needs verification.	Low

---

Previous suggestions

Suggestions up to commit 50ddb92

Category	Suggestion	Impact
Possible issue	Add null safety check The code assumes context.relBuilder.peek() is always non-null after unresolved.accept(), which may not be true if the visitor empties the builder stack. Add a null check before calling getId() to prevent potential NullPointerException. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [222-232] if (context.isTrackingEnabled()) { int idBefore = context.relBuilder.size() > 0 ? context.relBuilder.peek().getId() : -1; RelNode result = unresolved.accept(this, context); - int idAfter = context.relBuilder.peek().getId(); - java.util.List<Integer> producedIds = new java.util.ArrayList<>(); - for (int id = idBefore + 1; id <= idAfter; id++) { - producedIds.add(id); + if (context.relBuilder.size() > 0) { + int idAfter = context.relBuilder.peek().getId(); + java.util.List<Integer> producedIds = new java.util.ArrayList<>(); + for (int id = idBefore + 1; id <= idAfter; id++) { + producedIds.add(id); + } + context.recordMapping(unresolved.getClass().getSimpleName(), producedIds); } - context.recordMapping(unresolved.getClass().getSimpleName(), producedIds); return result; } Suggestion importance[1-10]: 8 __ Why: The code assumes context.relBuilder.peek() is non-null after unresolved.accept(), which could cause a NullPointerException if the visitor empties the builder stack. This is a critical safety issue that should be addressed.	Medium
	Prevent null pointer access Similar to the analyze method, context.relBuilder.peek() may be null after child.accept() completes. Verify the builder is non-empty before accessing peek().getId() to avoid NullPointerException. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [242-247] int idBefore = context.relBuilder.size() > 0 ? context.relBuilder.peek().getId() : -1; RelNode childResult = child.accept(this, context); result = childResult; -int idAfter = context.relBuilder.peek().getId(); +if (context.relBuilder.size() > 0) { + int idAfter = context.relBuilder.peek().getId(); + ... +} Suggestion importance[1-10]: 8 __ Why: Similar to the first suggestion, context.relBuilder.peek() may be null after child.accept() completes. This is a critical safety issue that could cause a NullPointerException in production.	Medium
General	Validate hook callback type The hook callback casts obj to RelRoot without validation. If the hook receives an unexpected object type, this will throw ClassCastException. Add type checking before casting to handle unexpected hook invocations gracefully. core/src/main/java/org/opensearch/sql/executor/QueryService.java [316-330] -AtomicReference<RelNode> physicalRelRef = new AtomicReference<>(); try (Hook.Closeable closeable = Hook.PLAN_BEFORE_IMPLEMENTATION.addThread( obj -> { - RelRoot relRoot = (RelRoot) obj; - physicalRelRef.set(relRoot.rel); - physicalPlanRef.set( - RelOptUtil.toString(relRoot.rel, SqlExplainLevel.ALL_ATTRIBUTES)); + if (obj instanceof RelRoot relRoot) { + physicalRelRef.set(relRoot.rel); + physicalPlanRef.set( + RelOptUtil.toString(relRoot.rel, SqlExplainLevel.ALL_ATTRIBUTES)); + } })) { - try (java.sql.PreparedStatement ignored = - OpenSearchRelRunners.run(context, calcitePlan)) { - } catch (java.sql.SQLException e) { - throw new RuntimeException(e); - } + ... } Suggestion importance[1-10]: 7 __ Why: The unchecked cast to RelRoot could throw ClassCastException if the hook receives an unexpected object type. Adding type validation improves robustness and prevents unexpected runtime failures.	Medium
	Cleanup profiling on interruption After catching InterruptedException and re-interrupting the thread, the method returns without cleaning up QueryProfiling state. Ensure QueryProfiling.noop() or equivalent cleanup is called before returning to prevent resource leaks. core/src/main/java/org/opensearch/sql/executor/QueryService.java [284-290] try { latch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); + QueryProfiling.noop(); listener.onFailure(new RuntimeException("Interrupted while waiting for query execution", e)); return; } Suggestion importance[1-10]: 6 __ Why: Adding QueryProfiling.noop() cleanup before returning on interruption prevents potential resource leaks and ensures consistent state management, though the impact is moderate since interruption is an exceptional case.	Low

Suggestions up to commit 726b7e6

Category	Suggestion	Impact
Possible issue	Ensure latch countdown on all exceptions If executeWithCalcite throws an exception before invoking the listener callback, the latch will never count down, causing the subsequent latch.await() to block indefinitely. Wrap the executeWithCalcite call in a try-catch block and ensure latch.countDown() is called in all error paths. core/src/main/java/org/opensearch/sql/executor/QueryService.java [255-282] AtomicReference<ExecutionEngine.QueryResponse> queryResponseRef = new AtomicReference<>(); AtomicReference<QueryProfile> profileRef = new AtomicReference<>(); AtomicReference<Exception> errorRef = new AtomicReference<>(); CountDownLatch latch = new CountDownLatch(1); -executeWithCalcite( - plan, - queryType, - null, - new ResponseListener<>() { - @Override - public void onResponse(ExecutionEngine.QueryResponse response) { - ... - latch.countDown(); - } +try { + executeWithCalcite( + plan, + queryType, + null, + new ResponseListener<>() { + @Override + public void onResponse(ExecutionEngine.QueryResponse response) { + ... + latch.countDown(); + } - @Override - public void onFailure(Exception e) { - errorRef.set(e); - latch.countDown(); - } - }); + @Override + public void onFailure(Exception e) { + errorRef.set(e); + latch.countDown(); + } + }); +} catch (Exception e) { + errorRef.set(e); + latch.countDown(); +} Suggestion importance[1-10]: 9 __ Why: If executeWithCalcite throws before invoking the listener, the latch never counts down, causing latch.await() to block indefinitely. Wrapping in try-catch and ensuring countdown in all error paths prevents deadlock.	High
	Prevent null pointer on empty builder The code assumes context.relBuilder.peek() will not be null after unresolved.accept(), but if the accept method doesn't push any nodes, this will throw a NoSuchElementException. Add a null/empty check before accessing peek() to prevent crashes when tracking is enabled for nodes that don't produce RelNodes. core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java [222-232] if (context.isTrackingEnabled()) { int idBefore = context.relBuilder.size() > 0 ? context.relBuilder.peek().getId() : -1; RelNode result = unresolved.accept(this, context); - int idAfter = context.relBuilder.peek().getId(); - java.util.List<Integer> producedIds = new java.util.ArrayList<>(); - for (int id = idBefore + 1; id <= idAfter; id++) { - producedIds.add(id); + if (context.relBuilder.size() > 0) { + int idAfter = context.relBuilder.peek().getId(); + java.util.List<Integer> producedIds = new java.util.ArrayList<>(); + for (int id = idBefore + 1; id <= idAfter; id++) { + producedIds.add(id); + } + context.recordMapping(unresolved.getClass().getSimpleName(), producedIds); } - context.recordMapping(unresolved.getClass().getSimpleName(), producedIds); return result; } Suggestion importance[1-10]: 8 __ Why: The code assumes context.relBuilder.peek() will not throw after unresolved.accept(), but if no nodes are pushed, peek() will throw NoSuchElementException. Adding a size check prevents a potential crash in tracking mode.	Medium
	Add timeout to prevent indefinite blocking The latch.await() call blocks indefinitely without a timeout, which can cause the thread to hang forever if the async callback never completes. Add a timeout to await() and handle the timeout case to prevent resource exhaustion and improve system resilience. core/src/main/java/org/opensearch/sql/executor/QueryService.java [284-290] try { - latch.await(); + if (!latch.await(60, java.util.concurrent.TimeUnit.SECONDS)) { + listener.onFailure(new RuntimeException("Query execution timed out after 60 seconds")); + return; + } } catch (InterruptedException e) { Thread.currentThread().interrupt(); listener.onFailure(new RuntimeException("Interrupted while waiting for query execution", e)); return; } Suggestion importance[1-10]: 7 __ Why: The latch.await() blocks indefinitely without a timeout. If the async callback never completes, the thread hangs forever. Adding a timeout improves resilience and prevents resource exhaustion.	Medium

[github-actions]

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit 127dfe7.

Path	Line	Severity	Description
plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java	196	medium	The `profile=true` flag is silently rerouted through the new analyze path (`analyze() || profile()`), which executes the query twice (once for timing, once for plan extraction). This doubles resource consumption for all existing profile requests without caller awareness. The inline comment acknowledges this is intentional but temporary, making it a deliberate behavioral change that could be exploited as a resource exhaustion vector if profile requests are not rate-limited.
core/src/main/java/org/opensearch/sql/executor/QueryService.java	237	low	The analyzeWithCalcite method forces profiling on globally via `QueryContext.setProfile(true)` before the async execution and does not reset it on the failure path or if the latch is interrupted. If the thread is reused (thread pool), subsequent requests on the same thread may inherit the profile=true state unintentionally.

The table above displays the top 10 most important findings.

Total: 2 | Critical: 0 | High: 0 | Medium: 1 | Low: 1

---

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

---

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

[github-actions]

Persistent review updated to latest commit 50ddb92

[github-actions]

Persistent review updated to latest commit 127dfe7