diff --git a/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java b/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java
index c7f3bc373ac..9ee2556668b 100644
--- a/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java
+++ b/core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java
@@ -46,6 +46,10 @@ public class CalcitePlanContext {
   /** This thread local variable is only used to skip script encoding in script pushdown. */
   public static final ThreadLocal<Boolean> skipEncoding = ThreadLocal.withInitial(() -> false);
 
+  /** When true, disables the OpenSearch shard request cache for the current query. */
+  public static final ThreadLocal<Boolean> disableRequestCache =
+      ThreadLocal.withInitial(() -> false);
+
   /** When true, the execution engine strips all-null columns from the result (used by timewrap). */
   public static final ThreadLocal<Boolean> stripNullColumns = ThreadLocal.withInitial(() -> false);
 
@@ -124,22 +128,6 @@ public class CalcitePlanContext {
   /** Whether we're currently inside a lambda context. */
   @Getter @Setter private boolean inLambdaContext = false;
 
-  /**
-   * When enabled, tracks which RelNode ids were produced by each AST command. Each entry maps an
-   * AST node class name to the list of RelNode ids it produced (excluding children).
-   */
-  @Getter @Setter private boolean trackingEnabled = false;
-
-  @Getter private final List<NodeIdMapping> nodeIdMappings = new ArrayList<>();
-
-  /** Records a mapping from an AST command to the RelNode ids it produced. */
-  public void recordMapping(String astNodeType, List<Integer> relNodeIds) {
-    nodeIdMappings.add(new NodeIdMapping(astNodeType, relNodeIds));
-  }
-
-  /** A mapping from one AST command to the RelNode ids it produced. */
-  public record NodeIdMapping(String astNodeType, List<Integer> relNodeIds) {}
-
   private CalcitePlanContext(FrameworkConfig config, SysLimit sysLimit, QueryType queryType) {
     this.config = config;
     this.sysLimit = sysLimit;
diff --git a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java
index e1f2e666c86..0f877b8b627 100644
--- a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java
+++ b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java
@@ -234,45 +234,11 @@ public CalciteRelNodeVisitor(DataSourceService dataSourceService) {
   }
 
   public RelNode analyze(UnresolvedPlan unresolved, CalcitePlanContext context) {
-    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();
-      List<Integer> producedIds = new ArrayList<>();
-      for (int id = idBefore + 1; id <= idAfter; id++) {
-        producedIds.add(id);
-      }
-      context.recordMapping(unresolved.getClass().getSimpleName(), producedIds);
-      return result;
-    }
     return unresolved.accept(this, context);
   }
 
   @Override
   public RelNode visitChildren(Node node, CalcitePlanContext context) {
-    if (context.isTrackingEnabled() && node instanceof UnresolvedPlan) {
-      // Track each child's total contribution (the subtree it produces)
-      RelNode result = null;
-      for (Node child : node.getChild()) {
-        int idBefore = context.relBuilder.size() > 0 ? context.relBuilder.peek().getId() : -1;
-        RelNode childResult = child.accept(this, context);
-        result = childResult;
-        // After child.accept returns, the child's visit* method has fully completed,
-        // so all RelNodes produced by that child (including ITS children) are on the stack.
-        int idAfter = context.relBuilder.peek().getId();
-        if (child instanceof UnresolvedPlan) {
-          List<Integer> producedIds = new ArrayList<>();
-          for (int id = idBefore + 1; id <= idAfter; id++) {
-            producedIds.add(id);
-          }
-          context.recordMapping(child.getClass().getSimpleName(), producedIds);
-        }
-      }
-      if (node instanceof UnresolvedPlan plan) {
-        mapPathMaterializer.materializePaths(plan, context);
-      }
-      return result;
-    }
     RelNode result = super.visitChildren(node, context);
     if (node instanceof UnresolvedPlan plan) {
       mapPathMaterializer.materializePaths(plan, context);
diff --git a/core/src/main/java/org/opensearch/sql/executor/AnalyzeResponse.java b/core/src/main/java/org/opensearch/sql/executor/AnalyzeResponse.java
index d8e0b12a8e7..6411ba888d5 100644
--- a/core/src/main/java/org/opensearch/sql/executor/AnalyzeResponse.java
+++ b/core/src/main/java/org/opensearch/sql/executor/AnalyzeResponse.java
@@ -13,19 +13,16 @@
 @Data
 @Builder
 public class AnalyzeResponse {
-
-  private final String query;
-  private final List<QuerySegment> querySegments;
-  // private final String ast;
+  // private final String query;
   private final List<String> logicalPlan;
   private final List<String> physicalPlan;
   private final QueryProfile profile;
-  private final List<OperatorNode> operator_tree;
-  private final List<String> recommendations;
+  private final List<Recommendation> recommendations;
   private final List<SchemaColumn> schema;
   private final Object[][] datarows;
   private final long total;
   private final long size;
+  private final boolean possibleCacheHit;
 
   @Data
   @Builder
@@ -34,23 +31,19 @@ public static class SchemaColumn {
     private final String type;
   }
 
-  @Data
-  @Builder
-  public static class QuerySegment {
-    private final String nodeType;
-    private final String source;
+  public enum RecommendationSeverityLevel {
+    INFO,
+    WARNING,
+    CRITICAL
   }
 
   @Data
   @Builder
-  public static class OperatorNode {
-    private final String source;
-    private final List<String> node_type;
-    private final List<String> description;
-    private final String estimated_cost;
-    private final Long estimated_rows;
-    private final String actual_time_ms;
-    private final Long actual_rows;
-    private final Boolean is_pushed_down;
+  public static class Recommendation {
+    private final RecommendationSeverityLevel severity;
+    private final String rule;
+    private final String message;
+    private final String affected_node;
+    private final String suggestion;
   }
 }
diff --git a/core/src/main/java/org/opensearch/sql/executor/ExecutionEngine.java b/core/src/main/java/org/opensearch/sql/executor/ExecutionEngine.java
index 9b51876c004..856eec37cc7 100644
--- a/core/src/main/java/org/opensearch/sql/executor/ExecutionEngine.java
+++ b/core/src/main/java/org/opensearch/sql/executor/ExecutionEngine.java
@@ -86,6 +86,14 @@ default void explain(
     explain(plan, mode, context, listener);
   }
 
+  /**
+   * Get the cumulative request cache hit count for the given indices. Returns -1 if not supported
+   * by this engine.
+   */
+  default long getRequestCacheHitCount(String... indexNames) {
+    return -1;
+  }
+
   /** Data class that encapsulates ExprValue. */
   @Data
   class QueryResponse {
diff --git a/core/src/main/java/org/opensearch/sql/executor/QueryService.java b/core/src/main/java/org/opensearch/sql/executor/QueryService.java
index 858ba0598e6..cabc653015a 100644
--- a/core/src/main/java/org/opensearch/sql/executor/QueryService.java
+++ b/core/src/main/java/org/opensearch/sql/executor/QueryService.java
@@ -6,11 +6,8 @@
 package org.opensearch.sql.executor;
 
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
-import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -43,8 +40,10 @@
 import org.apache.calcite.tools.Programs;
 import org.opensearch.sql.analysis.AnalysisContext;
 import org.opensearch.sql.analysis.Analyzer;
+import org.opensearch.sql.ast.Node;
 import org.opensearch.sql.ast.statement.ExplainMode;
 import org.opensearch.sql.ast.tree.HighlightConfig;
+import org.opensearch.sql.ast.tree.Relation;
 import org.opensearch.sql.ast.tree.UnresolvedPlan;
 import org.opensearch.sql.calcite.CalcitePlanContext;
 import org.opensearch.sql.calcite.CalciteRelNodeVisitor;
@@ -64,6 +63,7 @@
 import org.opensearch.sql.datasource.DataSourceService;
 import org.opensearch.sql.exception.CalciteUnsupportedException;
 import org.opensearch.sql.exception.NonFallbackCalciteException;
+import org.opensearch.sql.executor.analyze.AnalyzeRecommendationBuilder;
 import org.opensearch.sql.monitor.profile.MetricName;
 import org.opensearch.sql.monitor.profile.ProfileContext;
 import org.opensearch.sql.monitor.profile.ProfileMetric;
@@ -326,9 +326,8 @@ public void explainWithCalcite(
 
   public void analyzeWithCalcite(
       String query,
-      List<AnalyzeResponse.QuerySegment> querySegments,
       UnresolvedPlan plan,
-      QueryType queryType,
+      QueryType queryType, // boolean disableCache,
       ResponseListener<AnalyzeResponse> listener) {
     if (!shouldUseCalcite(queryType)) {
       listener.onFailure(
@@ -337,10 +336,19 @@ public void analyzeWithCalcite(
                   + " (plugins.calcite.enabled=true) and a PPL query type"));
       return;
     }
+    boolean disableCache = true;
     // Phase 1: Execute via the exact same path as executeWithCalcite + executionEngine.execute
     // to get identical profile timings. Use a latch to synchronize the async callback.
     // Force profiling on so executeWithCalcite activates QueryProfiling.
     QueryContext.setProfile(true);
+
+    String[] indexNames = extractIndexNames(plan);
+    long cacheHitsBefore = disableCache ? -1 : executionEngine.getRequestCacheHitCount(indexNames);
+
+    if (disableCache) {
+      CalcitePlanContext.disableRequestCache.set(true);
+    }
+
     AtomicReference<ExecutionEngine.QueryResponse> queryResponseRef = new AtomicReference<>();
     AtomicReference<QueryProfile> profileRef = new AtomicReference<>();
     AtomicReference<Exception> errorRef = new AtomicReference<>();
@@ -379,8 +387,11 @@ public void onFailure(Exception e) {
       latch.await();
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
+      CalcitePlanContext.disableRequestCache.remove();
       listener.onFailure(new RuntimeException("Interrupted while waiting for query execution", e));
       return;
+    } finally {
+      CalcitePlanContext.disableRequestCache.remove();
     }
 
     if (errorRef.get() != null) {
@@ -388,44 +399,17 @@ public void onFailure(Exception e) {
       return;
     }
 
+    long cacheHitsAfter = disableCache ? -1 : executionEngine.getRequestCacheHitCount(indexNames);
+    boolean possibleCacheHit =
+        !disableCache
+            && cacheHitsBefore >= 0
+            && cacheHitsAfter >= 0
+            && cacheHitsAfter > cacheHitsBefore;
+
     ExecutionEngine.QueryResponse queryResponse = queryResponseRef.get();
     QueryProfile profile = profileRef.get();
 
-    // If the profile plan tree has branching (any node with >1 child), our linear
-    // operator tree logic won't work. Return a response that 'fallsback' on `profile`
-    // by only including fields mirroring the `profile` endpoint.
-    if (profile != null && profile.getPlan() != null && !isLinearPlanTree(profile)) {
-      List<AnalyzeResponse.SchemaColumn> schema = new ArrayList<>();
-      if (queryResponse.getSchema() != null) {
-        for (ExecutionEngine.Schema.Column col : queryResponse.getSchema().getColumns()) {
-          schema.add(
-              AnalyzeResponse.SchemaColumn.builder()
-                  .name(col.getName())
-                  .type(col.getExprType().typeName())
-                  .build());
-        }
-      }
-      Object[][] datarows = new Object[queryResponse.getResults().size()][];
-      int rowIdx = 0;
-      for (var exprValue : queryResponse.getResults()) {
-        datarows[rowIdx++] =
-            exprValue.tupleValue().entrySet().stream()
-                .map(e -> e.getValue().value())
-                .toArray(Object[]::new);
-      }
-      listener.onResponse(
-          AnalyzeResponse.builder()
-              // .query(query)
-              .profile(profile)
-              .schema(schema)
-              .datarows(datarows)
-              .total(datarows.length)
-              .size(datarows.length)
-              .build());
-      return;
-    }
-
-    // Phase 2: Re-run with tracking to capture logical/physical plans and node mappings.
+    // Phase 2: Re-run to capture logical/physical plans.
     // This run benefits from warm caches but we don't report its timings.
     CalcitePlanContext.run(
         () -> {
@@ -436,17 +420,15 @@ public void onFailure(Exception e) {
                   CalcitePlanContext context =
                       CalcitePlanContext.create(
                           buildFrameworkConfig(), SysLimit.fromSettings(settings), queryType);
-                  context.setTrackingEnabled(true);
                   RelNode relNode = analyze(plan, context);
-                  RelNode calcitePlan = convertToCalcitePlan(relNode, context);
+                  RelNode calcitePlan =
+                      withCheckedArithmetic(convertToCalcitePlan(relNode, context), context);
 
                   AtomicReference<String> physicalPlanRef = new AtomicReference<>();
-                  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));
                           })) {
@@ -470,16 +452,6 @@ public void onFailure(Exception e) {
                           .filter(s -> !s.isEmpty())
                           .toList();
 
-                  // Build operator tree using phase 2's tracking data + phase 1's profile.
-                  List<AnalyzeResponse.OperatorNode> operatorTree =
-                      buildOperatorTree(
-                          querySegments,
-                          logicalPlanNodes,
-                          context.getNodeIdMappings(),
-                          calcitePlan,
-                          physicalRelRef.get(),
-                          profile);
-
                   // Convert QueryResponse results to analyze format.
                   List<AnalyzeResponse.SchemaColumn> schema = new ArrayList<>();
                   if (queryResponse.getSchema() != null) {
@@ -502,15 +474,17 @@ public void onFailure(Exception e) {
                             .toArray(Object[]::new);
                   }
 
+                  List<AnalyzeResponse.Recommendation> recommendations =
+                      new AnalyzeRecommendationBuilder(profile).build();
+
                   AnalyzeResponse response =
                       AnalyzeResponse.builder()
-                          .query(query)
-                          .querySegments(querySegments)
+                          // .query(query)
                           .logicalPlan(logicalPlanNodes)
                           .physicalPlan(physicalPlanNodes)
-                          .operator_tree(operatorTree)
-                          .recommendations(List.of())
+                          .recommendations(recommendations)
                           .profile(profile)
+                          .possibleCacheHit(possibleCacheHit)
                           .schema(schema)
                           .datarows(datarows)
                           .total(datarows.length)
@@ -530,265 +504,6 @@ public void onFailure(Exception e) {
         settings);
   }
 
-  private List<AnalyzeResponse.OperatorNode> buildOperatorTree(
-      List<AnalyzeResponse.QuerySegment> querySegments,
-      List<String> logicalPlanNodes,
-      List<CalcitePlanContext.NodeIdMapping> nodeIdMappings,
-      RelNode logicalPlan,
-      RelNode physicalPlan,
-      QueryProfile profile) {
-    // Build a map from RelNode id to its logical plan description string.
-    Map<Integer, String> idToDescription = new HashMap<>();
-    for (String node : logicalPlanNodes) {
-      int idIdx = node.lastIndexOf("id = ");
-      if (idIdx >= 0) {
-        String idStr = node.substring(idIdx + 5).trim();
-        try {
-          int id = Integer.parseInt(idStr);
-          idToDescription.put(id, node);
-        } catch (NumberFormatException ignored) {
-        }
-      }
-    }
-
-    // Compute exclusive ids per mapping by subtracting the previous mapping's ids.
-    // Mappings are recorded bottom-up: [Relation:[0], Filter:[0,1], Project:[0,1,2]]
-    // Exclusive: Relation=[0], Filter=[1], Project=[2]
-    List<Set<Integer>> exclusiveIds = new ArrayList<>();
-    Set<Integer> previousIds = new HashSet<>();
-    for (CalcitePlanContext.NodeIdMapping mapping : nodeIdMappings) {
-      Set<Integer> current = new HashSet<>(mapping.relNodeIds());
-      Set<Integer> exclusive = new HashSet<>(current);
-      exclusive.removeAll(previousIds);
-      exclusiveIds.add(exclusive);
-      previousIds = current;
-    }
-
-    // Determine how many segments from the bottom were pushed into the physical scan.
-    // The physical plan's leaf node (the scan) absorbs logical nodes from the bottom up.
-    // Physical depth tells us how many separate physical operators exist; everything else
-    // was pushed down. We count segments bottom-up until we've covered all pushed logical nodes.
-    int physicalDepth = getLinearDepth(physicalPlan);
-    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++;
-    }
-
-    // log.info(
-    //     "buildOperatorTree: pushedSegments={}, pushedLogicalNodes={}",
-    //     pushedSegments,
-    //     pushedLogicalNodes);
-
-    // Compute estimated row counts from the logical plan using RelMetadataQuery.
-    // Walk the logical plan bottom-up to get rowcount per node by id.
-    org.apache.calcite.rel.metadata.RelMetadataQuery mq =
-        logicalPlan.getCluster().getMetadataQuery();
-    Map<Integer, Double> idToRowCount = new HashMap<>();
-    collectRowCounts(logicalPlan, mq, idToRowCount);
-
-    // Compute exclusive time and rows per physical node from the profile plan tree.
-    // The plan tree is top-down; we flatten it bottom-up to match operator tree order.
-    List<double[]> physicalTimings = new ArrayList<>();
-    if (profile != null && profile.getPlan() != null) {
-      List<QueryProfile.PlanNode> planNodes = new ArrayList<>();
-      QueryProfile.PlanNode current = (QueryProfile.PlanNode) profile.getPlan();
-      while (current != null) {
-        planNodes.add(current);
-        current =
-            (current.getChildren() != null && !current.getChildren().isEmpty())
-                ? current.getChildren().get(0)
-                : null;
-      }
-      // planNodes is top-down; reverse to bottom-up
-      java.util.Collections.reverse(planNodes);
-      for (int p = 0; p < planNodes.size(); p++) {
-        double inclusive = planNodes.get(p).getTimeMillis();
-        double childInclusive = (p > 0) ? planNodes.get(p - 1).getTimeMillis() : 0;
-        double exclusive = Math.max(0, inclusive - childInclusive);
-        long rows = planNodes.get(p).getRows();
-        physicalTimings.add(new double[] {exclusive, rows});
-      }
-    }
-
-    List<AnalyzeResponse.OperatorNode> operators = new ArrayList<>();
-    int physicalIdx = 0;
-
-    // Build the pushed-down merged entry (first pushedSegments segments)
-    if (pushedSegments > 1) {
-      List<AnalyzeResponse.QuerySegment> mergedSegments = querySegments.subList(0, pushedSegments);
-      List<String> descriptions = new ArrayList<>();
-      for (int idx = 0; idx < pushedSegments; idx++) {
-        Set<Integer> ids = idx < exclusiveIds.size() ? exclusiveIds.get(idx) : Set.of();
-        ids.stream()
-            .sorted()
-            .map(idToDescription::get)
-            .filter(Objects::nonNull)
-            .forEach(descriptions::add);
-      }
-      String combinedSource =
-          mergedSegments.stream()
-              .map(AnalyzeResponse.QuerySegment::getSource)
-              .reduce((a, b) -> a + " | " + b)
-              .orElse("");
-      List<String> nodeTypes =
-          mergedSegments.stream().map(AnalyzeResponse.QuerySegment::getNodeType).toList();
-      // Collect all plan node ids in the pushed group for estimated_rows
-      Set<Integer> allPushedPlanIds = new HashSet<>();
-      for (int i = 0; i < pushedSegments; i++) {
-        Set<Integer> ids = i < exclusiveIds.size() ? exclusiveIds.get(i) : Set.of();
-        ids.stream().filter(idToDescription::containsKey).forEach(allPushedPlanIds::add);
-      }
-      double[] timing =
-          physicalIdx < physicalTimings.size() ? physicalTimings.get(physicalIdx) : null;
-      physicalIdx++;
-      operators.add(
-          AnalyzeResponse.OperatorNode.builder()
-              .source(combinedSource)
-              .node_type(nodeTypes)
-              .description(descriptions.isEmpty() ? null : descriptions)
-              .is_pushed_down(true)
-              .estimated_rows(getEstimatedRows(allPushedPlanIds, idToRowCount))
-              .actual_time_ms(timing != null ? String.format("%.2f ms", timing[0]) : null)
-              .actual_rows(timing != null ? (long) timing[1] : null)
-              .build());
-    } else if (pushedSegments == 1) {
-      AnalyzeResponse.QuerySegment seg = querySegments.get(0);
-      Set<Integer> ids = !exclusiveIds.isEmpty() ? exclusiveIds.get(0) : Set.of();
-      Set<Integer> planIds =
-          ids.stream()
-              .filter(idToDescription::containsKey)
-              .collect(java.util.stream.Collectors.toSet());
-      List<String> descriptions =
-          ids.stream().sorted().map(idToDescription::get).filter(Objects::nonNull).toList();
-      double[] timing =
-          physicalIdx < physicalTimings.size() ? physicalTimings.get(physicalIdx) : null;
-      physicalIdx++;
-      operators.add(
-          AnalyzeResponse.OperatorNode.builder()
-              .source(seg.getSource())
-              .node_type(List.of(seg.getNodeType()))
-              .description(descriptions.isEmpty() ? null : descriptions)
-              .estimated_rows(getEstimatedRows(planIds, idToRowCount))
-              .actual_time_ms(timing != null ? String.format("%.2f ms", timing[0]) : null)
-              .actual_rows(timing != null ? (long) timing[1] : null)
-              .build());
-    }
-
-    // Remaining segments map to non-scan physical nodes (physicalDepth - 1 of them).
-    // Each physical node corresponds to one logical plan node. Group segments so that each
-    // group covers exactly one logical plan node; segments with 0 plan nodes merge into the
-    // next group that has one.
-    int idx = pushedSegments;
-    while (idx < querySegments.size()) {
-      List<AnalyzeResponse.QuerySegment> group = new ArrayList<>();
-      List<String> descriptions = new ArrayList<>();
-      Set<Integer> groupPlanIds = new HashSet<>();
-      long logicalNodesInGroup = 0;
-      while (idx < querySegments.size() && logicalNodesInGroup < 1) {
-        group.add(querySegments.get(idx));
-        Set<Integer> ids = idx < exclusiveIds.size() ? exclusiveIds.get(idx) : Set.of();
-        ids.stream()
-            .sorted()
-            .map(idToDescription::get)
-            .filter(Objects::nonNull)
-            .forEach(descriptions::add);
-        ids.stream().filter(idToDescription::containsKey).forEach(groupPlanIds::add);
-        logicalNodesInGroup += ids.stream().filter(idToDescription::containsKey).count();
-        idx++;
-      }
-      String combinedSource =
-          group.stream()
-              .map(AnalyzeResponse.QuerySegment::getSource)
-              .reduce((a, b) -> a + " | " + b)
-              .orElse("");
-      List<String> nodeTypes =
-          group.stream().map(AnalyzeResponse.QuerySegment::getNodeType).toList();
-      double[] timing =
-          physicalIdx < physicalTimings.size() ? physicalTimings.get(physicalIdx) : null;
-      physicalIdx++;
-      operators.add(
-          AnalyzeResponse.OperatorNode.builder()
-              .source(combinedSource)
-              .node_type(nodeTypes)
-              .description(descriptions.isEmpty() ? null : descriptions)
-              .estimated_rows(getEstimatedRows(groupPlanIds, idToRowCount))
-              .actual_time_ms(timing != null ? String.format("%.2f ms", timing[0]) : null)
-              .actual_rows(timing != null ? (long) timing[1] : null)
-              .build());
-    }
-
-    return operators;
-  }
-
-  private static boolean isLinearPlanTree(QueryProfile profile) {
-    QueryProfile.PlanNode current = (QueryProfile.PlanNode) profile.getPlan();
-    while (current != null) {
-      if (current.getChildren() != null && current.getChildren().size() > 1) {
-        return false;
-      }
-      current =
-          (current.getChildren() != null && !current.getChildren().isEmpty())
-              ? current.getChildren().get(0)
-              : null;
-    }
-    return true;
-  }
-
-  private static int getLinearDepth(RelNode node) {
-    int depth = 0;
-    RelNode current = node;
-    while (current != null) {
-      depth++;
-      List<RelNode> inputs = current.getInputs();
-      current = inputs.isEmpty() ? null : inputs.get(0);
-    }
-    return depth;
-  }
-
-  private void collectRowCounts(
-      RelNode node,
-      org.apache.calcite.rel.metadata.RelMetadataQuery mq,
-      Map<Integer, Double> idToRowCount) {
-    try {
-      Double rowCount = mq.getRowCount(node);
-      if (rowCount != null) {
-        idToRowCount.put(node.getId(), rowCount);
-      }
-    } catch (Exception ignored) {
-    }
-    for (RelNode input : node.getInputs()) {
-      collectRowCounts(input, mq, idToRowCount);
-    }
-  }
-
-  private Long getEstimatedRows(Set<Integer> ids, Map<Integer, Double> idToRowCount) {
-    return ids.stream()
-        .filter(idToRowCount::containsKey)
-        .max(Integer::compareTo)
-        .map(id -> Math.round(idToRowCount.get(id)))
-        .orElse(null);
-  }
-
   public void executeWithLegacy(
       UnresolvedPlan plan,
       QueryType queryType,
@@ -1021,6 +736,23 @@ private FrameworkConfig buildFrameworkConfig() {
     return configBuilder.build();
   }
 
+  private static String[] extractIndexNames(UnresolvedPlan plan) {
+    Set<String> names = new HashSet<>();
+    collectRelationNames(plan, names);
+    return names.toArray(String[]::new);
+  }
+
+  private static void collectRelationNames(Node node, Set<String> names) {
+    if (node instanceof Relation relation) {
+      names.add(relation.getTableQualifiedName().toString());
+    }
+    if (node.getChild() != null) {
+      for (Node child : node.getChild()) {
+        collectRelationNames(child, names);
+      }
+    }
+  }
+
   /**
    * Convert OpenSearch Plan to Calcite Plan. Although both plans consist of Calcite RelNodes, there
    * are some differences in the topological structures or semantics between them.
diff --git a/core/src/main/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilder.java b/core/src/main/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilder.java
new file mode 100644
index 00000000000..1cc68028861
--- /dev/null
+++ b/core/src/main/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilder.java
@@ -0,0 +1,285 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.executor.analyze;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Optional;
+import org.opensearch.sql.executor.AnalyzeResponse.Recommendation;
+import org.opensearch.sql.executor.AnalyzeResponse.RecommendationSeverityLevel;
+import org.opensearch.sql.monitor.profile.QueryProfile;
+import org.opensearch.sql.monitor.profile.QueryProfile.PlanNode;
+
+/**
+ * Builds the list of {@link Recommendation}s returned by the PPL {@code analyze} endpoint from the
+ * {@link QueryProfile}'s plan-node tree and phase timings.
+ *
+ * <p>Each rule lives in its own method. Per-node rules ({@code ineffectiveFilter}, {@code
+ * joinRowExplosion}, {@code expensiveSort}) scan every plan node and may emit more than one
+ * recommendation, so they return a {@link List}. Whole-query rules ({@code bottleneckStage}, {@code
+ * optimizePhaseDominates}) emit at most one and return an {@link Optional}. {@link #build()} runs
+ * every rule and concatenates the results.
+ *
+ * <p>Row semantics: a plan node's {@code rows} is its output row count ({@code rows_out}); {@code
+ * rows_in} is the sum of its children's {@code rows}.
+ *
+ * <p>Timing semantics: a node's {@code time_ms} is cumulative wall-time (it includes its
+ * descendants' time), not the node's own duration. A node's self-time is therefore {@code time_ms -
+ * max(child.time_ms)} (see {@link #duration}). Time-fraction rules use this self-time so they
+ * attribute the cost actually spent in the stage rather than the whole subtree beneath it.
+ */
+public class AnalyzeRecommendationBuilder {
+
+  // Configurable thresholds (defaults from the rule spec).
+
+  /** Ineffective filter/project: fires when rows_out / rows_in exceeds this pass-through ratio. */
+  private static final double INEFFECTIVE_FILTER_MAX_PASS_RATIO = 0.95;
+
+  /** Join row explosion: fires when rows_out / rows_in exceeds this ratio (WARNING). */
+  private static final double JOIN_EXPLOSION_RATIO = 5.0;
+
+  /**
+   * Join row explosion escalates to CRITICAL at or above this ratio. Not specified by the rule
+   * table; chosen as a higher tier above {@link #JOIN_EXPLOSION_RATIO}.
+   */
+  private static final double JOIN_EXPLOSION_CRITICAL_RATIO = 20.0;
+
+  /** Expensive sort: fires when the sort's time fraction of execute time exceeds this. */
+  private static final double EXPENSIVE_SORT_TIME_FRACTION = 0.20;
+
+  /** Expensive sort: additionally requires at least this many input rows. */
+  private static final long EXPENSIVE_SORT_MIN_ROWS = 50_000;
+
+  /** Bottleneck stage: fires when the slowest node's time fraction of execute exceeds this. */
+  private static final double BOTTLENECK_TIME_FRACTION = 0.60;
+
+  /** Optimize phase dominates: fires when optimize time exceeds this (ms) and beats execute. */
+  private static final double OPTIMIZE_DOMINATES_MIN_MS = 75.0;
+
+  private final QueryProfile profile;
+
+  public AnalyzeRecommendationBuilder(QueryProfile profile) {
+    this.profile = profile;
+  }
+
+  /** Runs every recommendation rule and concatenates the ones that fired. */
+  public List<Recommendation> build() {
+    List<Recommendation> recommendations = new ArrayList<>();
+    if (profile == null) {
+      return recommendations;
+    }
+    recommendations.addAll(ineffectiveFilter());
+    recommendations.addAll(joinRowExplosion());
+    recommendations.addAll(expensiveSort());
+    bottleneckStage().ifPresent(recommendations::add);
+    optimizePhaseDominates().ifPresent(recommendations::add);
+    return recommendations;
+  }
+
+  /** Flags filter/project stages that barely reduced their input. */
+  private List<Recommendation> ineffectiveFilter() {
+    List<Recommendation> recommendations = new ArrayList<>();
+    for (PlanNode node : planNodes()) {
+      String name = node.getNode().toLowerCase(Locale.ROOT);
+      if (!name.contains("filter") && !name.contains("project")) {
+        continue;
+      }
+      long rowsIn = rowsIn(node);
+      if (rowsIn <= 0) {
+        continue;
+      }
+      double ratio = (double) node.getRows() / rowsIn;
+      if (ratio > INEFFECTIVE_FILTER_MAX_PASS_RATIO) {
+        long droppedPct = Math.round((1.0 - ratio) * 100);
+        recommendations.add(
+            Recommendation.builder()
+                .severity(RecommendationSeverityLevel.WARNING)
+                .rule("Ineffective Filter")
+                .message("Filter only dropped " + droppedPct + "% of rows")
+                .affected_node(node.getNode())
+                .suggestion("Consider removing the filter or making it more selective.")
+                .build());
+      }
+    }
+    return recommendations;
+  }
+
+  /** Flags joins whose output greatly exceeds their combined input. */
+  private List<Recommendation> joinRowExplosion() {
+    List<Recommendation> recommendations = new ArrayList<>();
+    for (PlanNode node : planNodes()) {
+      if (!node.getNode().toLowerCase(Locale.ROOT).contains("join")) {
+        continue;
+      }
+      long rowsIn = rowsIn(node);
+      if (rowsIn <= 0) {
+        continue;
+      }
+      double ratio = (double) node.getRows() / rowsIn;
+      if (ratio > JOIN_EXPLOSION_RATIO) {
+        RecommendationSeverityLevel severity =
+            ratio >= JOIN_EXPLOSION_CRITICAL_RATIO
+                ? RecommendationSeverityLevel.CRITICAL
+                : RecommendationSeverityLevel.WARNING;
+        recommendations.add(
+            Recommendation.builder()
+                .severity(severity)
+                .rule("Join Row Explosion")
+                .message(
+                    "Join expanded "
+                        + rowsIn
+                        + " rows into "
+                        + node.getRows()
+                        + " rows ("
+                        + String.format(Locale.ROOT, "%.1f", ratio)
+                        + "×)")
+                .affected_node(node.getNode())
+                .suggestion("Add filters to the subqueries before the join to reduce rows.")
+                .build());
+      }
+    }
+    return recommendations;
+  }
+
+  /** Flags sorts over large inputs that consumed a large share of execution time. */
+  private List<Recommendation> expensiveSort() {
+    List<Recommendation> recommendations = new ArrayList<>();
+    double executeMs = phaseTime("execute");
+    if (executeMs <= 0) {
+      return recommendations;
+    }
+    for (PlanNode node : planNodes()) {
+      if (!node.getNode().toLowerCase(Locale.ROOT).contains("sort")) {
+        continue;
+      }
+      long rowsIn = rowsIn(node);
+      double durationMs = duration(node);
+      double timeFraction = durationMs / executeMs;
+      if (timeFraction > EXPENSIVE_SORT_TIME_FRACTION && rowsIn > EXPENSIVE_SORT_MIN_ROWS) {
+        long pct = Math.round(timeFraction * 100);
+        recommendations.add(
+            Recommendation.builder()
+                .severity(RecommendationSeverityLevel.WARNING)
+                .rule("Expensive Sort")
+                .message(
+                    "Sorting "
+                        + rowsIn
+                        + " rows took "
+                        + durationMs
+                        + " ms ("
+                        + pct
+                        + "% of execution)")
+                .affected_node(node.getNode())
+                .suggestion("Filter or limit rows before sorting (e.g. add head or a where).")
+                .build());
+      }
+    }
+    return recommendations;
+  }
+
+  /** Flags the single stage that dominated execution time. */
+  private Optional<Recommendation> bottleneckStage() {
+    double executeMs = phaseTime("execute");
+    if (executeMs <= 0) {
+      return Optional.empty();
+    }
+    PlanNode slowest = null;
+    double slowestDuration = 0;
+    for (PlanNode node : planNodes()) {
+      double durationMs = duration(node);
+      if (slowest == null || durationMs > slowestDuration) {
+        slowest = node;
+        slowestDuration = durationMs;
+      }
+    }
+    if (slowest == null) {
+      return Optional.empty();
+    }
+    double timeFraction = slowestDuration / executeMs;
+    if (timeFraction <= BOTTLENECK_TIME_FRACTION) {
+      return Optional.empty();
+    }
+    long pct = Math.round(timeFraction * 100);
+    return Optional.of(
+        Recommendation.builder()
+            .severity(RecommendationSeverityLevel.INFO)
+            .rule("Bottleneck Stage")
+            .message(
+                slowest.getNode() + " took " + slowestDuration + " ms (" + pct + "% of execution)")
+            .affected_node(slowest.getNode())
+            .build());
+  }
+
+  /** Flags queries that spent more time planning than executing. */
+  private Optional<Recommendation> optimizePhaseDominates() {
+    double executeMs = phaseTime("execute");
+    double optimizeMs = phaseTime("optimize");
+    if (executeMs < optimizeMs && optimizeMs > OPTIMIZE_DOMINATES_MIN_MS) {
+      return Optional.of(
+          Recommendation.builder()
+              .severity(RecommendationSeverityLevel.INFO)
+              .rule("Optimize Phase Dominates")
+              .message(
+                  "Query planning took " + optimizeMs + " ms vs " + executeMs + " ms executing")
+              .build());
+    }
+    return Optional.empty();
+  }
+
+  /** Flattens the profile's plan-node tree into a list (root first, depth-first). */
+  private List<PlanNode> planNodes() {
+    List<PlanNode> nodes = new ArrayList<>();
+    if (profile.getPlan() instanceof PlanNode root) {
+      collect(root, nodes);
+    }
+    return nodes;
+  }
+
+  private static void collect(PlanNode node, List<PlanNode> out) {
+    out.add(node);
+    if (node.getChildren() != null) {
+      for (PlanNode child : node.getChildren()) {
+        collect(child, out);
+      }
+    }
+  }
+
+  /**
+   * Self-time for a node in milliseconds. {@code time_ms} is cumulative wall-time (a node's clock
+   * includes its descendants'), so the node's own duration is its time minus the slowest child's
+   * time. Clamped at 0 to guard against measurement jitter making a child appear slower than its
+   * parent.
+   */
+  private static double duration(PlanNode node) {
+    double maxChild = 0;
+    if (node.getChildren() != null) {
+      for (PlanNode child : node.getChildren()) {
+        maxChild = Math.max(maxChild, child.getTimeMillis());
+      }
+    }
+    return Math.max(0, node.getTimeMillis() - maxChild);
+  }
+
+  /** Input rows for a node: the sum of its children's output rows. */
+  private static long rowsIn(PlanNode node) {
+    if (node.getChildren() == null || node.getChildren().isEmpty()) {
+      return 0;
+    }
+    long sum = 0;
+    for (PlanNode child : node.getChildren()) {
+      sum += child.getRows();
+    }
+    return sum;
+  }
+
+  /** Millis for a named profile phase, or 0 if absent. */
+  private double phaseTime(String phaseName) {
+    QueryProfile.Phase phase =
+        profile.getPhases() == null ? null : profile.getPhases().get(phaseName);
+    return phase == null ? 0 : phase.getTimeMillis();
+  }
+}
diff --git a/core/src/main/java/org/opensearch/sql/executor/execution/AnalyzePlan.java b/core/src/main/java/org/opensearch/sql/executor/execution/AnalyzePlan.java
index a43bc32792e..8e0a5762a39 100644
--- a/core/src/main/java/org/opensearch/sql/executor/execution/AnalyzePlan.java
+++ b/core/src/main/java/org/opensearch/sql/executor/execution/AnalyzePlan.java
@@ -5,12 +5,10 @@
 
 package org.opensearch.sql.executor.execution;
 
-import java.util.List;
 import org.opensearch.sql.ast.statement.ExplainMode;
 import org.opensearch.sql.ast.tree.UnresolvedPlan;
 import org.opensearch.sql.common.response.ResponseListener;
 import org.opensearch.sql.executor.AnalyzeResponse;
-import org.opensearch.sql.executor.AnalyzeResponse.QuerySegment;
 import org.opensearch.sql.executor.ExecutionEngine;
 import org.opensearch.sql.executor.QueryId;
 import org.opensearch.sql.executor.QueryService;
@@ -20,7 +18,6 @@
 public class AnalyzePlan extends AbstractPlan {
 
   private final String query;
-  private final List<QuerySegment> querySegments;
   private final UnresolvedPlan plan;
   private final QueryService queryService;
   private final ResponseListener<AnalyzeResponse> listener;
@@ -29,13 +26,11 @@ public AnalyzePlan(
       QueryId queryId,
       QueryType queryType,
       String query,
-      List<QuerySegment> querySegments,
       UnresolvedPlan plan,
       QueryService queryService,
       ResponseListener<AnalyzeResponse> listener) {
     super(queryId, queryType);
     this.query = query;
-    this.querySegments = querySegments;
     this.plan = plan;
     this.queryService = queryService;
     this.listener = listener;
@@ -43,7 +38,7 @@ public AnalyzePlan(
 
   @Override
   public void execute() {
-    queryService.analyzeWithCalcite(query, querySegments, plan, getQueryType(), listener);
+    queryService.analyzeWithCalcite(query, plan, getQueryType(), listener);
   }
 
   @Override
diff --git a/core/src/main/java/org/opensearch/sql/executor/execution/QueryPlanFactory.java b/core/src/main/java/org/opensearch/sql/executor/execution/QueryPlanFactory.java
index 0e44dd02e7e..a8eff67b513 100644
--- a/core/src/main/java/org/opensearch/sql/executor/execution/QueryPlanFactory.java
+++ b/core/src/main/java/org/opensearch/sql/executor/execution/QueryPlanFactory.java
@@ -7,7 +7,6 @@
 
 import static java.util.Objects.requireNonNull;
 
-import java.util.List;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.tuple.Pair;
 import org.opensearch.sql.ast.AbstractNodeVisitor;
@@ -153,11 +152,9 @@ public AbstractPlan visitExplain(
   /** Create an AnalyzePlan that produces AST node and logical plan RelNode. */
   public AbstractPlan createAnalyzePlan(
       String query,
-      List<AnalyzeResponse.QuerySegment> querySegments,
       UnresolvedPlan plan,
       QueryType queryType,
       ResponseListener<AnalyzeResponse> listener) {
-    return new AnalyzePlan(
-        QueryId.queryId(), queryType, query, querySegments, plan, queryService, listener);
+    return new AnalyzePlan(QueryId.queryId(), queryType, query, plan, queryService, listener);
   }
 }
diff --git a/core/src/test/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilderTest.java b/core/src/test/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilderTest.java
new file mode 100644
index 00000000000..29974906e4e
--- /dev/null
+++ b/core/src/test/java/org/opensearch/sql/executor/analyze/AnalyzeRecommendationBuilderTest.java
@@ -0,0 +1,268 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.executor.analyze;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.EnumMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import org.junit.jupiter.api.Test;
+import org.opensearch.sql.executor.AnalyzeResponse.Recommendation;
+import org.opensearch.sql.executor.AnalyzeResponse.RecommendationSeverityLevel;
+import org.opensearch.sql.monitor.profile.MetricName;
+import org.opensearch.sql.monitor.profile.QueryProfile;
+import org.opensearch.sql.monitor.profile.QueryProfile.PlanNode;
+
+class AnalyzeRecommendationBuilderTest {
+
+  private static QueryProfile profile(double optimizeMs, double executeMs, PlanNode plan) {
+    Map<MetricName, Double> phases = new EnumMap<>(MetricName.class);
+    phases.put(MetricName.OPTIMIZE, optimizeMs);
+    phases.put(MetricName.EXECUTE, executeMs);
+    return new QueryProfile(optimizeMs + executeMs, phases, plan);
+  }
+
+  private static PlanNode leaf(String name, long rows) {
+    return new PlanNode(name, 1.0, rows, null);
+  }
+
+  private static Optional<Recommendation> ruleOf(List<Recommendation> recs, String rule) {
+    return recs.stream().filter(r -> r.getRule().equals(rule)).findFirst();
+  }
+
+  @Test
+  void nullProfileYieldsNoRecommendations() {
+    assertTrue(new AnalyzeRecommendationBuilder(null).build().isEmpty());
+  }
+
+  @Test
+  void ineffectiveFilterFiresWhenFilterBarelyReducesRows() {
+    // filter passes 990 of 1000 rows -> ratio 0.99 > 0.95
+    PlanNode filter = new PlanNode("CalciteFilter", 5.0, 990, List.of(leaf("scan", 1000)));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 10, filter)).build();
+
+    Recommendation r = ruleOf(recs, "Ineffective Filter").orElseThrow();
+    assertEquals(RecommendationSeverityLevel.WARNING, r.getSeverity());
+    assertEquals("Filter only dropped 1% of rows", r.getMessage());
+    assertEquals("CalciteFilter", r.getAffected_node());
+  }
+
+  @Test
+  void ineffectiveFilterSilentWhenFilterIsSelective() {
+    // filter passes 100 of 1000 rows -> ratio 0.10, not ineffective
+    PlanNode filter = new PlanNode("CalciteFilter", 5.0, 100, List.of(leaf("scan", 1000)));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 10, filter)).build();
+    assertTrue(ruleOf(recs, "Ineffective Filter").isEmpty());
+  }
+
+  @Test
+  void joinRowExplosionWarnsAboveRatioFive() {
+    // 100 in -> 800 out = 8x (>5, <20) -> WARNING
+    PlanNode join =
+        new PlanNode("EnumerableHashJoin", 5.0, 800, List.of(leaf("l", 60), leaf("r", 40)));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 10, join)).build();
+
+    Recommendation r = ruleOf(recs, "Join Row Explosion").orElseThrow();
+    assertEquals(RecommendationSeverityLevel.WARNING, r.getSeverity());
+    assertEquals("Join expanded 100 rows into 800 rows (8.0×)", r.getMessage());
+  }
+
+  @Test
+  void joinRowExplosionCriticalAtHighRatio() {
+    // 100 in -> 3000 out = 30x (>=20) -> CRITICAL
+    PlanNode join =
+        new PlanNode("EnumerableHashJoin", 5.0, 3000, List.of(leaf("l", 50), leaf("r", 50)));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 10, join)).build();
+    assertEquals(
+        RecommendationSeverityLevel.CRITICAL,
+        ruleOf(recs, "Join Row Explosion").orElseThrow().getSeverity());
+  }
+
+  @Test
+  void expensiveSortFiresOnLargeSlowSort() {
+    // time_ms is cumulative: scan 10ms, sort 40ms -> sort self-time 30ms of 100 execute (30% >
+    // 20%); 60k input rows (> 50k)
+    PlanNode scan = new PlanNode("scan", 10.0, 60_000, null);
+    PlanNode sort = new PlanNode("EnumerableSort", 40.0, 60_000, List.of(scan));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 100, sort)).build();
+
+    Recommendation r = ruleOf(recs, "Expensive Sort").orElseThrow();
+    assertEquals(RecommendationSeverityLevel.WARNING, r.getSeverity());
+    assertTrue(r.getMessage().contains("Sorting 60000 rows"));
+    assertTrue(r.getMessage().contains("30.0 ms"));
+    assertTrue(r.getMessage().contains("30% of execution"));
+  }
+
+  @Test
+  void expensiveSortSilentWhenSelfTimeIsSmall() {
+    // sort cumulative 90ms but its child took 89ms -> self-time only 1ms, not expensive
+    PlanNode scan = new PlanNode("scan", 89.0, 60_000, null);
+    PlanNode sort = new PlanNode("EnumerableSort", 90.0, 60_000, List.of(scan));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 100, sort)).build();
+    assertTrue(ruleOf(recs, "Expensive Sort").isEmpty());
+  }
+
+  @Test
+  void expensiveSortSilentWhenInputSmall() {
+    // 30% self-time but only 100 input rows (< 50k)
+    PlanNode scan = new PlanNode("scan", 10.0, 100, null);
+    PlanNode sort = new PlanNode("EnumerableSort", 40.0, 100, List.of(scan));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 100, sort)).build();
+    assertTrue(ruleOf(recs, "Expensive Sort").isEmpty());
+  }
+
+  @Test
+  void bottleneckStageFiresOnSlowestSelfTimeNotCumulative() {
+    // Root project cumulative 100ms but its child scan took 90ms -> project self-time 10ms,
+    // scan self-time 90ms. Bottleneck should be the scan (90% > 75%), NOT the root.
+    PlanNode scan = new PlanNode("CalciteEnumerableIndexScan", 90.0, 10, null);
+    PlanNode project = new PlanNode("EnumerableProject", 100.0, 10, List.of(scan));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 100, project)).build();
+
+    Recommendation r = ruleOf(recs, "Bottleneck Stage").orElseThrow();
+    assertEquals(RecommendationSeverityLevel.INFO, r.getSeverity());
+    assertTrue(r.getMessage().contains("CalciteEnumerableIndexScan"));
+    assertTrue(r.getMessage().contains("90% of execution"));
+  }
+
+  @Test
+  void bottleneckStageSilentWhenNoNodeDominatesBySelfTime() {
+    // Root cumulative 100ms but 45ms is its own and 55ms is the child's -> no single node > 75%.
+    PlanNode scan = new PlanNode("CalciteEnumerableIndexScan", 55.0, 10, null);
+    PlanNode project = new PlanNode("EnumerableProject", 100.0, 10, List.of(scan));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 100, project)).build();
+    assertTrue(ruleOf(recs, "Bottleneck Stage").isEmpty());
+  }
+
+  @Test
+  void optimizePhaseDominatesFiresWhenPlanningExceedsExecution() {
+    // optimize 100 > execute 10, and optimize > 75ms
+    PlanNode scan = new PlanNode("CalciteEnumerableIndexScan", 5.0, 10, null);
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(100, 10, scan)).build();
+
+    Recommendation r = ruleOf(recs, "Optimize Phase Dominates").orElseThrow();
+    assertEquals(RecommendationSeverityLevel.INFO, r.getSeverity());
+    assertTrue(r.getMessage().contains("Query planning took 100.0 ms vs 10.0 ms executing"));
+  }
+
+  @Test
+  void optimizePhaseDominatesSilentWhenExecutionLarger() {
+    PlanNode scan = new PlanNode("CalciteEnumerableIndexScan", 5.0, 10, null);
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(100, 200, scan)).build();
+    assertTrue(ruleOf(recs, "Optimize Phase Dominates").isEmpty());
+  }
+
+  @Test
+  void ineffectiveFilterFiresForProjectNodes() {
+    // The rule matches "project" as well as "filter"; a project passing 999 of 1000 rows fires.
+    PlanNode project = new PlanNode("EnumerableProject", 5.0, 999, List.of(leaf("scan", 1000)));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 10, project)).build();
+
+    Recommendation r = ruleOf(recs, "Ineffective Filter").orElseThrow();
+    assertEquals(RecommendationSeverityLevel.WARNING, r.getSeverity());
+    assertEquals("EnumerableProject", r.getAffected_node());
+  }
+
+  @Test
+  void ineffectiveFilterSilentForLeafFilterWithNoInput() {
+    // A "filter"-named leaf has no children, so rows_in is 0 and the ratio guard skips it.
+    PlanNode leafFilter = leaf("CalciteFilter", 1000);
+    List<Recommendation> recs =
+        new AnalyzeRecommendationBuilder(profile(1, 10, leafFilter)).build();
+    assertTrue(ruleOf(recs, "Ineffective Filter").isEmpty());
+  }
+
+  @Test
+  void ineffectiveFilterSilentExactlyAtThreshold() {
+    // ratio == 0.95 exactly; the rule uses strict '>' so it must NOT fire.
+    PlanNode filter = new PlanNode("CalciteFilter", 5.0, 950, List.of(leaf("scan", 1000)));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 10, filter)).build();
+    assertTrue(ruleOf(recs, "Ineffective Filter").isEmpty());
+  }
+
+  @Test
+  void joinRowExplosionSilentExactlyAtThreshold() {
+    // ratio == 5.0 exactly; strict '>' means no recommendation.
+    PlanNode join =
+        new PlanNode("EnumerableHashJoin", 5.0, 500, List.of(leaf("l", 50), leaf("r", 50)));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 10, join)).build();
+    assertTrue(ruleOf(recs, "Join Row Explosion").isEmpty());
+  }
+
+  @Test
+  void timeBasedRulesSilentWhenExecutePhaseIsZero() {
+    // executeMs == 0 -> Expensive Sort and Bottleneck Stage short-circuit to empty.
+    PlanNode sort = new PlanNode("EnumerableSort", 40.0, 60_000, List.of(leaf("scan", 60_000)));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 0, sort)).build();
+    assertTrue(ruleOf(recs, "Expensive Sort").isEmpty());
+    assertTrue(ruleOf(recs, "Bottleneck Stage").isEmpty());
+  }
+
+  @Test
+  void nonPlanNodePlanYieldsNoRecommendations() {
+    // profile.plan is a pre-rendered object (not a PlanNode) -> no per-node rules can run.
+    Map<MetricName, Double> phases = new EnumMap<>(MetricName.class);
+    phases.put(MetricName.OPTIMIZE, 1.0);
+    phases.put(MetricName.EXECUTE, 10.0);
+    QueryProfile p = new QueryProfile(11.0, phases, "some pre-rendered plan string");
+    assertTrue(new AnalyzeRecommendationBuilder(p).build().isEmpty());
+  }
+
+  @Test
+  void nullPlanYieldsNoPerNodeRecommendations() {
+    // No plan tree, but phase-based Optimize Phase Dominates can still fire.
+    QueryProfile p = profile(100, 10, null);
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(p).build();
+    assertTrue(ruleOf(recs, "Bottleneck Stage").isEmpty());
+    assertTrue(ruleOf(recs, "Expensive Sort").isEmpty());
+    assertTrue(ruleOf(recs, "Optimize Phase Dominates").isPresent());
+  }
+
+  @Test
+  void durationClampedToZeroWhenChildSlowerThanParent() {
+    // Measurement jitter: child (60ms) reads slower than parent (50ms) -> parent self-time 0,
+    // so the parent is not treated as the bottleneck despite a large cumulative time.
+    PlanNode child = new PlanNode("scan", 60.0, 10, null);
+    PlanNode parent = new PlanNode("EnumerableProject", 50.0, 10, List.of(child));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 100, parent)).build();
+
+    // The scan (60ms self-time, 60% of execute) is below 75%, so nothing fires -- and crucially
+    // the clamp prevents a negative parent duration from being ranked as the max.
+    Recommendation r = ruleOf(recs, "Bottleneck Stage").orElse(null);
+    if (r != null) {
+      assertTrue(r.getMessage().contains("scan"));
+    }
+  }
+
+  @Test
+  void multipleRulesFireForOneQuery() {
+    // A single tree that trips both Ineffective Filter and Bottleneck Stage, plus Optimize
+    // Phase Dominates from the phases. Verifies build() concatenates across rules.
+    PlanNode scan = new PlanNode("CalciteEnumerableIndexScan", 90.0, 990, null);
+    PlanNode filter = new PlanNode("CalciteFilter", 100.0, 990, List.of(scan));
+    // optimize 200 > execute 100 and > 75ms
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(200, 100, filter)).build();
+
+    assertTrue(ruleOf(recs, "Ineffective Filter").isPresent());
+    assertTrue(ruleOf(recs, "Bottleneck Stage").isPresent());
+    assertTrue(ruleOf(recs, "Optimize Phase Dominates").isPresent());
+  }
+
+  @Test
+  void perNodeRuleFiresOncePerMatchingNode() {
+    // Two ineffective filters in one tree -> two recommendations.
+    PlanNode scan = new PlanNode("scan", 1.0, 1000, null);
+    PlanNode filter1 = new PlanNode("CalciteFilter", 2.0, 999, List.of(scan));
+    PlanNode filter2 = new PlanNode("CalciteFilter", 3.0, 998, List.of(filter1));
+    List<Recommendation> recs = new AnalyzeRecommendationBuilder(profile(1, 10, filter2)).build();
+
+    long count = recs.stream().filter(r -> r.getRule().equals("Ineffective Filter")).count();
+    assertEquals(2, count);
+  }
+}
diff --git a/docs/user/ppl/interfaces/endpoint.md b/docs/user/ppl/interfaces/endpoint.md
index 6704cc0cd48..4dfd2798924 100644
--- a/docs/user/ppl/interfaces/endpoint.md
+++ b/docs/user/ppl/interfaces/endpoint.md
@@ -154,7 +154,7 @@ calcite:
 ```
 ## Analyze (Experimental)
 
-You can enable analysis on the PPL endpoint to capture query execution details including per-stage timings, logical and physical plans, operator tree with pushdown visibility, and optimization recommendations. Analysis is returned only for regular query execution (not explain) and only when using the default `format=jdbc`.
+You can enable analysis on the PPL endpoint to capture query execution details including per-phase timings, logical and physical plans, and optimization recommendations. Analysis is returned only for regular query execution (not explain) and only when using the default `format=jdbc`.
 
 ### Example
 
@@ -162,7 +162,7 @@ You can enable analysis on the PPL endpoint to capture query execution details i
 curl -sS -H 'Content-Type: application/json' \
   -X POST localhost:9200/_plugins/_ppl \
   -d '{
-        "query": "source=accounts | where age < 30 | eval full_name = firstname + \" \" + lastname | fields full_name, email, age", 
+        "query": "source=test_data | where bytes_sent < 30 | eval full_trip =  client_city + \" \" + client_country | fields full_trip",
         "analyze": true
       }'
 ```
@@ -171,58 +171,66 @@ Expected output (trimmed):
 
 ```json
 {
-  "query": "source=accounts | where age < 30 | eval full_name = firstname + \" \" + lastname | fields full_name, email, age",
-  "querySegments": [
-    {"nodeType": "SearchFrom", "source": "source=accounts"},
-    {"nodeType": "WhereCommand", "source": "where age < 30"},
-  ],
   "logicalPlan": [
-    "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT]): rowcount = 5000.0, cumulative cost = {114000.0 rows, 145000.0 cpu, 0.0 io}, id = 4229",
-    "LogicalProject(full_name=[||(||($0, ' '), $4)], email=[$3], age=[$2]): rowcount = 5000.0, cumulative cost = {109000.0 rows, 25000.0 cpu, 0.0 io}, id = 4228",
-    "LogicalFilter(condition=[<($2, 30)]): rowcount = 5000.0, cumulative cost = {104000.0 rows, 10000.0 cpu, 0.0 io}, id = 4226",
-    "CalciteLogicalIndexScan(table=[[OpenSearch, accounts]]): rowcount = 10000.0, cumulative cost = {99000.0 rows, 0.0 cpu, 0.0 io}, id = 4225"
+    "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT]): rowcount = 5000.0, cumulative cost = {645000.0 rows, 95000.0 cpu, 0.0 io}, id = 15952",
+    "LogicalProject(full_trip=[||(||($22, ' '), $59)]): rowcount = 5000.0, cumulative cost = {640000.0 rows, 15000.0 cpu, 0.0 io}, id = 15951",
+    "LogicalFilter(condition=[<($50, 30)]): rowcount = 5000.0, cumulative cost = {635000.0 rows, 10000.0 cpu, 0.0 io}, id = 15949",
+    "CalciteLogicalIndexScan(table=[[OpenSearch, test_data]]): rowcount = 10000.0, cumulative cost = {630000.0 rows, 0.0 cpu, 0.0 io}, id = 15948"
   ],
   "physicalPlan": [
-    "EnumerableCalc(expr#0..3=[{inputs}], expr#4=[' '], expr#5=[||($t0, $t4)], expr#6=[||($t5, $t3)], full_name=[$t6], email=[$t2], age=[$t1]): rowcount = 5000.0, cumulative cost = {22996.4 rows, 50000.0 cpu, 0.0 io}, id = 4319",
-    "CalciteEnumerableIndexScan(table=[[OpenSearch, accounts]], PushDownContext=[[PROJECT->[firstname, age, email, lastname], FILTER-><($1, 30), LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":10000,\"timeout\":\"1m\",\"query\":{
+    "EnumerableCalc(expr#0..1=[{inputs}], expr#2=[' '], expr#3=[||($t0, $t2)], expr#4=[||($t3, $t1)], full_trip=[$t4]): rowcount = 5000.0, cumulative cost = {13998.2 rows, 30000.0 cpu, 0.0 io}, id = 16035",
+    "CalciteEnumerableIndexScan(table=[[OpenSearch, test_data]], PushDownContext=[[PROJECT->[client_city, bytes_sent, client_country], FILTER-><($1, 30), LIMIT->10000, PROJECT->[client_city, client_country]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":10000,\"timeout\":\"1m\",\"query\":{\"range\":{\"bytes_sent\":{\"from\":null,\"to\":30,\"include_lower\":true,\"include_upper\":false,\"boost\":1.0}}},\"_source\":{\"includes\":[\"client_city\",\"client_country\"]}}, requestedTotalSize=10000, pageSize=null, startFrom=0)]): rowcount = 5000.0, cumulative cost = {8998.2 rows, 0.0 cpu, 0.0 io}, id = 16027"
   ],
   "profile": {
     "summary": {
-      "total_time_ms": 37.13
+      "total_time_ms": 40.85
     },
     "phases": {
-      "analyze": { "time_ms": 7.06 },
-      "optimize": { "time_ms": 25.29 },
-      "execute": { "time_ms": 4.73 },
-      "format": { "time_ms": 0.03 }
+      "analyze": {
+        "time_ms": 4.52
+      },
+      "optimize": {
+        "time_ms": 20.04
+      },
+      "execute": {
+        "time_ms": 16.17
+      },
+      "format": {
+        "time_ms": 0.0
+      }
     },
     "plan": {
       "node": "EnumerableCalc",
-      "time_ms": 3.44,
-      "rows": 3,
+      "time_ms": 14.4,
+      "rows": 0,
       "children": [
-        { "node": "CalciteEnumerableIndexScan", "time_ms": 3.31, "rows": 3 }
+        {
+          "node": "CalciteEnumerableIndexScan",
+          "time_ms": 14.19,
+          "rows": 0
+        }
       ]
-    }
+    },
+    "thread_pool": "sql-worker"
   },
-  "operator_tree": [
+  "recommendations": [
     {
-      "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 = 4225",
-        "LogicalFilter(condition=[<($2, 30)]): rowcount = 5000.0, cumulative cost = {104000.0 rows, 10000.0 cpu, 0.0 io}, id = 4226"
-      ],
-      "estimated_rows": 5000,
-      "actual_time_ms": "3.31 ms",
-      "actual_rows": 3,
-      "is_pushed_down": true
-    },
+      "severity": "INFO",
+      "rule": "Bottleneck Stage",
+      "message": "CalciteEnumerableIndexScan took 14.19 ms (88% of execution)",
+      "affected_node": "CalciteEnumerableIndexScan"
+    }
+  ],
+  "schema": [
+    {
+      "name": "full_trip",
+      "type": "STRING"
+    }
   ],
-  "recommendations": []
+  "datarows": [],
+  "total": 0,
+  "size": 0,
+  "possibleCacheHit": false
 }
 ```
 
@@ -230,45 +238,25 @@ Expected output (trimmed):
 
 | Field | Type | Description |
 |-------|------|-------------|
-| `query` | String | The original PPL query. |
-| `querySegments` | Array | Breakdown of the query into AST segments with `nodeType` and `source`. |
 | `logicalPlan` | Array | Calcite logical plan nodes (top-down). |
 | `physicalPlan` | Array | Calcite physical plan nodes after optimization. |
-| `operator_tree` | Array | Per-stage execution details linking query segments to plan operators. |
 | `recommendations` | Array | Optimization suggestions generated from the execution profile. |
 | `profile` | Object | Per-phase timing breakdown (same format as the profile endpoint). |
 | `schema` | Array | Column names and types of the query result. |
 | `datarows` | Array | Query result rows. |
 | `total` | Integer | Total number of result rows. |
 | `size` | Integer | Number of result rows returned. |
-
-### Operator tree fields
-
-| Field | Type | Description |
-|-------|------|-------------|
-| `source` | String | The PPL query fragment(s) that produced this operator. |
-| `node_type` | Array | AST node type(s) (e.g. `Relation`, `Filter`, `Project`). |
-| `description` | Array | Logical plan node descriptions. |
-| `estimated_rows` | Long | Estimated row count from Calcite metadata. |
-| `actual_time_ms` | String | Exclusive wall-clock time for this operator. |
-| `actual_rows` | Long | Actual rows produced by this operator. |
-| `is_pushed_down` | Boolean | Whether the operator was pushed down to the storage engine. |
-
-
+| `possibleCacheHit` | Boolean | Returns whether or not the `analyze` query was potentially sped up due the cache. |
 
 ### Notes
 - Analyze output is only returned when the query finishes successfully.
 - Analyze requires the Calcite engine to be enabled (`plugins.calcite.enabled=true`).
-- Operator tree nodes with `is_pushed_down: true` were executed within the OpenSearch storage engine (single network round-trip). Remaining operators ran in-memory on the coordinating node.
-- This endpoint is meant to replace/override the existing `profile` endpoint. As a result, any POST requests with either `"analyze": true` or `"profile": true` (or both) will be routed to this endpoint.
-  - The `profile` section uses the same format as the previous `profile` endpoint. This means current consumers of `profile` should not face any breaking changes.
+- The `profile` section uses the same format as the `profile` endpoint.
 - The logic for `analyze` doesn't hold for queries that produce non-linear physical plan trees (for example, JOINs). In this scenario, `analyze` will return an output identical to the previous `profile` endpoint.
 
 
 ## Profile (Experimental) (Deprecated)
 
-**<u>This endpoint is outdated, see the `analyze` section above.</u>**
-
 You can enable profiling on the PPL endpoint to capture per-stage timings in milliseconds. Profiling is returned only for regular query execution (not explain) and only when using the default `format=jdbc`.
 
 ### Example
diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteAnalyzeIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteAnalyzeIT.java
index df2e1b88e9a..8df85a01825 100644
--- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteAnalyzeIT.java
+++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteAnalyzeIT.java
@@ -82,62 +82,6 @@ public void analyzeResultsMatchWithAggregation() throws IOException {
         normal.getJSONArray("datarows").length(), analyzed.getJSONArray("datarows").length());
   }
 
-  // === B. Operator tree — all pushed down ===
-
-  @Test
-  public void operatorTreeAllPushedDown() throws IOException {
-    JSONObject result =
-        executeAnalyze(
-            "source=" + TEST_INDEX_ACCOUNT + " | where age > 30 | fields firstname, age");
-    JSONArray tree = result.getJSONArray("operator_tree");
-
-    // Single physical node → all segments merged into one entry
-    assertEquals(1, tree.length());
-    JSONObject node = tree.getJSONObject(0);
-    assertTrue(node.getBoolean("is_pushed_down"));
-
-    JSONArray nodeTypes = node.getJSONArray("node_type");
-    assertTrue(nodeTypes.toString().contains("SearchFrom"));
-    assertTrue(nodeTypes.toString().contains("WhereCommand"));
-    assertTrue(nodeTypes.toString().contains("FieldsCommand"));
-  }
-
-  @Test
-  public void operatorTreeAllPushedDownWithStats() throws IOException {
-    JSONObject result =
-        executeAnalyze(
-            "source=" + TEST_INDEX_ACCOUNT + " | where age > 30 | stats count() by gender");
-    JSONArray tree = result.getJSONArray("operator_tree");
-
-    assertEquals(1, tree.length());
-    JSONObject node = tree.getJSONObject(0);
-    assertTrue(node.getBoolean("is_pushed_down"));
-
-    JSONArray nodeTypes = node.getJSONArray("node_type");
-    assertTrue(nodeTypes.toString().contains("SearchFrom"));
-    assertTrue(nodeTypes.toString().contains("WhereCommand"));
-    assertTrue(nodeTypes.toString().contains("StatsCommand"));
-  }
-
-  // === C. Operator tree — partial pushdown ===
-
-  @Test
-  public void operatorTreePartialPushdown() throws IOException {
-    JSONObject result =
-        executeAnalyze(
-            "source="
-                + TEST_INDEX_ACCOUNT
-                + " | where age > 30 | eval name = firstname | fields name, age");
-    JSONArray tree = result.getJSONArray("operator_tree");
-
-    // At least 2 entries: pushed-down group + non-pushed group
-    assertTrue(tree.length() >= 2);
-    // First entry should be pushed down
-    assertTrue(tree.getJSONObject(0).optBoolean("is_pushed_down", false));
-    // Last entry should NOT be pushed down
-    assertFalse(tree.getJSONObject(tree.length() - 1).optBoolean("is_pushed_down", false));
-  }
-
   // === D. Profile structure ===
 
   @Test
@@ -177,61 +121,6 @@ public void analyzeProfilePlanHasNodeInfo() throws IOException {
     assertTrue(plan.getLong("rows") >= 0);
   }
 
-  // === E. Timing correctness ===
-
-  @Test
-  public void operatorTreeHasTimings() throws IOException {
-    JSONObject result =
-        executeAnalyze("source=" + TEST_INDEX_ACCOUNT + " | where age > 30 | fields firstname");
-    JSONArray tree = result.getJSONArray("operator_tree");
-
-    for (int i = 0; i < tree.length(); i++) {
-      JSONObject node = tree.getJSONObject(i);
-      assertTrue("node " + i + " has actual_time_ms", node.has("actual_time_ms"));
-      assertTrue("node " + i + " has actual_rows", node.has("actual_rows"));
-      assertTrue(node.getLong("actual_rows") >= 0);
-    }
-  }
-
-  @Test
-  public void operatorTreeTimingsSumApproximatesPlanRoot() throws IOException {
-    JSONObject result =
-        executeAnalyze(
-            "source="
-                + TEST_INDEX_ACCOUNT
-                + " | where age > 30 | eval x = age * 2 | fields x, firstname");
-    JSONArray tree = result.getJSONArray("operator_tree");
-    JSONObject profile = result.getJSONObject("profile");
-
-    double totalOperatorTime = 0;
-    for (int i = 0; i < tree.length(); i++) {
-      String timeStr = tree.getJSONObject(i).getString("actual_time_ms");
-      totalOperatorTime += Double.parseDouble(timeStr.replace(" ms", ""));
-    }
-    double planRootTime = profile.getJSONObject("plan").getDouble("time_ms");
-
-    // Exclusive times should sum to roughly the root inclusive time.
-    // Allow generous tolerance for off-spine subtree time not captured.
-    assertTrue(
-        "operator times (" + totalOperatorTime + ") roughly match plan root (" + planRootTime + ")",
-        totalOperatorTime <= planRootTime * 2.0 && totalOperatorTime >= planRootTime * 0.1);
-  }
-
-  // === F. Estimated rows ===
-
-  @Test
-  public void operatorTreeHasEstimatedRows() throws IOException {
-    JSONObject result =
-        executeAnalyze("source=" + TEST_INDEX_ACCOUNT + " | where age > 30 | fields firstname");
-    JSONArray tree = result.getJSONArray("operator_tree");
-
-    for (int i = 0; i < tree.length(); i++) {
-      JSONObject node = tree.getJSONObject(i);
-      assertTrue("node " + i + " has estimated_rows", node.has("estimated_rows"));
-      assertTrue(node.getLong("estimated_rows") > 0);
-    }
-  }
-
   // === G. Logical and physical plan presence ===
 
   @Test
@@ -262,10 +151,8 @@ public void analyzeEmptyResults() throws IOException {
 
     assertEquals(0, result.getInt("total"));
     assertEquals(0, result.getJSONArray("datarows").length());
-    // Profile and operator tree should still be present
+    // Profile should still be present
     assertTrue(result.has("profile"));
-    assertTrue(result.has("operator_tree"));
-    assertTrue(result.getJSONArray("operator_tree").length() > 0);
   }
 
   @Test
@@ -279,6 +166,24 @@ public void analyzeSyntaxErrorReturnsError() {
     assertThrows(ResponseException.class, () -> executeAnalyze("this is not valid ppl"));
   }
 
+  @Test
+  public void analyzeLongArithmeticOverflowReturnsError() throws IOException {
+    // Regression guard: the analyze path must apply checked arithmetic just like execute, so a
+    // BIGINT overflow surfaces as a client error instead of silently wrapping. See issue #5164.
+    // 9223372036854775807 is Long.MAX_VALUE; + 1 overflows and has no wider integer type.
+    ResponseException e =
+        assertThrows(
+            ResponseException.class,
+            () ->
+                executeAnalyze(
+                    "source="
+                        + TEST_INDEX_ACCOUNT
+                        + " | head 1 | eval overflow = 9223372036854775807 + 1 | fields overflow"));
+    assertEquals(400, e.getResponse().getStatusLine().getStatusCode());
+    String body = getResponseBody(e.getResponse(), true);
+    assertTrue("expected an overflow error, got: " + body, body.contains("overflow"));
+  }
+
   // === I. Schema correctness ===
 
   @Test
@@ -293,6 +198,65 @@ public void analyzeSchemaMatchesQueryFields() throws IOException {
     assertEquals("age", schema.getJSONObject(1).getString("name"));
   }
 
+  // === L. Recommendations ===
+
+  @Test
+  public void analyzeIncludesRecommendationsArray() throws IOException {
+    JSONObject result =
+        executeAnalyze("source=" + TEST_INDEX_ACCOUNT + " | where age > 30 | fields firstname");
+    // recommendations is always present (possibly empty) and is an array.
+    assertTrue(result.has("recommendations"));
+    result.getJSONArray("recommendations");
+  }
+
+  @Test
+  public void analyzeRecommendationsHaveWellFormedShape() throws IOException {
+    // A non-selective filter (age >= 0 keeps every row) should trip the "Ineffective Filter" rule;
+    // even if it does not on this data, any emitted recommendation must have the required fields.
+    JSONObject result =
+        executeAnalyze("source=" + TEST_INDEX_ACCOUNT + " | where age >= 0 | fields firstname");
+    JSONArray recs = result.getJSONArray("recommendations");
+    for (int i = 0; i < recs.length(); i++) {
+      JSONObject rec = recs.getJSONObject(i);
+      assertTrue("recommendation has severity", rec.has("severity"));
+      assertTrue("recommendation has rule", rec.has("rule"));
+      assertTrue("recommendation has message", rec.has("message"));
+      String severity = rec.getString("severity");
+      assertTrue(
+          "severity is a known level",
+          severity.equals("INFO") || severity.equals("WARNING") || severity.equals("CRITICAL"));
+    }
+  }
+
+  @Test
+  public void analyzeRecommendationRulesAreKnownAndConsistent() throws IOException {
+    // Whatever rules fire against a real plan/profile, each must be one of the known rule names
+    // and carry the fields that rule populates. This validates the rules run end-to-end against a
+    // real QueryProfile.PlanNode tree (node names, timings) rather than only synthetic unit input.
+    JSONObject result =
+        executeAnalyze(
+            "source=" + TEST_INDEX_ACCOUNT + " | where age >= 0 | stats count() by state");
+    JSONArray recs = result.getJSONArray("recommendations");
+    for (int i = 0; i < recs.length(); i++) {
+      JSONObject rec = recs.getJSONObject(i);
+      String rule = rec.getString("rule");
+      assertTrue(
+          "unexpected rule: " + rule,
+          rule.equals("Ineffective Filter")
+              || rule.equals("Join Row Explosion")
+              || rule.equals("Expensive Sort")
+              || rule.equals("Bottleneck Stage")
+              || rule.equals("Optimize Phase Dominates"));
+      // affected_node accompanies the per-node rules.
+      if (rule.equals("Ineffective Filter")
+          || rule.equals("Join Row Explosion")
+          || rule.equals("Expensive Sort")
+          || rule.equals("Bottleneck Stage")) {
+        assertTrue("per-node rule should name a node", rec.has("affected_node"));
+      }
+    }
+  }
+
   // === J. Profile timing similarity to standalone profile endpoint ===
 
   @Test
@@ -312,39 +276,4 @@ public void analyzeTimingsInSameOrderOfMagnitudeAsProfile() throws IOException {
         "analyze total (" + analyzeTotal + ") within 5x of profile total (" + profileTotal + ")",
         analyzeTotal < profileTotal * 5 && analyzeTotal > profileTotal / 5);
   }
-
-  // === K. Pushdown disabled ===
-
-  @Test
-  public void analyzeWithPushdownDisabledShowsNoPushdown() throws IOException {
-    // Disable pushdown
-    updateClusterSettings(
-        new ClusterSetting(
-            "transient",
-            org.opensearch.sql.common.setting.Settings.Key.CALCITE_PUSHDOWN_ENABLED.getKeyValue(),
-            "false"));
-    try {
-      JSONObject result =
-          executeAnalyze(
-              "source=" + TEST_INDEX_ACCOUNT + " | where age > 30 | fields firstname, age");
-      JSONArray tree = result.getJSONArray("operator_tree");
-
-      // With pushdown disabled, nothing should be marked as pushed down
-      // (or it should have multiple nodes since operations stay separate)
-      if (tree.length() == 1) {
-        // If still 1 node, it shouldn't be marked pushed_down
-        assertFalse(tree.getJSONObject(0).optBoolean("is_pushed_down", false));
-      } else {
-        // Multiple nodes means operations weren't merged
-        assertTrue(tree.length() > 1);
-      }
-    } finally {
-      // Re-enable pushdown
-      updateClusterSettings(
-          new ClusterSetting(
-              "transient",
-              org.opensearch.sql.common.setting.Settings.Key.CALCITE_PUSHDOWN_ENABLED.getKeyValue(),
-              "true"));
-    }
-  }
 }
diff --git a/integ-test/src/yamlRestTest/resources/rest-api-spec/test/api/ppl.analyze.yml b/integ-test/src/yamlRestTest/resources/rest-api-spec/test/api/ppl.analyze.yml
index a1087dbe6da..abb6685eea4 100644
--- a/integ-test/src/yamlRestTest/resources/rest-api-spec/test/api/ppl.analyze.yml
+++ b/integ-test/src/yamlRestTest/resources/rest-api-spec/test/api/ppl.analyze.yml
@@ -51,11 +51,8 @@ teardown:
         body:
           query: 'source=ppl_analyze | fields message'
           analyze: true
-  - match: {query: 'source=ppl_analyze | fields message'}
-  - is_true: querySegments
   - is_true: logicalPlan
   - is_true: physicalPlan
-  - is_true: operator_tree
   - is_true: recommendations
   - is_true: profile
   - gt: {profile.summary.total_time_ms: 0.0}
@@ -65,23 +62,6 @@ teardown:
   - match: {total: 2}
   - match: {size: 2}
 
----
-"Analyze returns operator tree with pushdown info":
-  - skip:
-      features:
-        - headers
-        - allowed_warnings
-  - do:
-      headers:
-        Content-Type: 'application/json'
-      ppl:
-        body:
-          query: 'source=ppl_analyze | where age > 30 | fields message'
-          analyze: true
-  - is_true: operator_tree
-  - match: {total: 1}
-  - match: {size: 1}
-
 ---
 "Analyze returns recommendations field":
   - skip:
@@ -110,8 +90,8 @@ teardown:
         body:
           query: 'source=ppl_analyze | fields message'
           analyze: true
-  - match: {query: null}
-  - match: {operator_tree: null}
+  - match: {recommendations: null}
+  - match: {profile: null}
 
 ---
 "Analyze with non-jdbc format still returns response":
@@ -126,4 +106,4 @@ teardown:
         body:
           query: 'source=ppl_analyze | fields message'
           analyze: true
-  - is_true: query
+  - is_true: datarows
diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java
index d9681898f73..60f8c22c331 100644
--- a/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java
+++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/client/OpenSearchNodeClient.java
@@ -126,12 +126,6 @@ public Map<String, IndexMapping> getIndexMappings(String... indexExpression) {
     }
   }
 
-  /**
-   * Fetch index.max_result_window settings according to index expression given.
-   *
-   * @param indexExpression index expression
-   * @return map from index name to its max result window
-   */
   @Override
   public Map<String, Integer> getIndexMaxResultWindows(String... indexExpression) {
     try {
diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java
index 483f2684d61..0a964e674f4 100644
--- a/opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java
+++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/executor/OpenSearchExecutionEngine.java
@@ -359,6 +359,30 @@ public void execute(
         });
   }
 
+  @Override
+  public long getRequestCacheHitCount(String... indexNames) {
+    Optional<NodeClient> nodeClientOpt = client.getNodeClient();
+    if (nodeClientOpt.isEmpty()) {
+      return -1;
+    }
+    try {
+      return nodeClientOpt
+          .get()
+          .admin()
+          .indices()
+          .prepareStats(indexNames)
+          .clear()
+          .setRequestCache(true)
+          .get()
+          .getTotal()
+          .getRequestCache()
+          .getHitCount();
+    } catch (Exception e) {
+      logger.warn("Failed to retrieve request cache stats", e);
+      return -1;
+    }
+  }
+
   /**
    * Substring of the error OpenSearch's {@code SearchService} raises when a node has no free PIT
    * contexts. The engine opens a PIT (one reader context per shard) to page over a query it cannot
diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchQueryRequest.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchQueryRequest.java
index a3a6954cadd..48aaedc16eb 100644
--- a/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchQueryRequest.java
+++ b/opensearch/src/main/java/org/opensearch/sql/opensearch/request/OpenSearchQueryRequest.java
@@ -41,6 +41,7 @@
 import org.opensearch.search.sort.FieldSortBuilder;
 import org.opensearch.search.sort.ShardDocSortBuilder;
 import org.opensearch.search.sort.SortBuilders;
+import org.opensearch.sql.calcite.CalcitePlanContext;
 import org.opensearch.sql.opensearch.data.value.OpenSearchExprValueFactory;
 import org.opensearch.sql.opensearch.response.OpenSearchResponse;
 import org.opensearch.sql.opensearch.storage.OpenSearchIndex;
@@ -86,6 +87,8 @@ public class OpenSearchQueryRequest implements OpenSearchRequest {
 
   @ToString.Exclude private Map<String, Object> afterKey;
 
+  @EqualsAndHashCode.Exclude @ToString.Exclude private final boolean disableRequestCache;
+
   @TestOnly
   static OpenSearchQueryRequest of(
       String indexName, int size, OpenSearchExprValueFactory factory, List<String> includes) {
@@ -140,6 +143,7 @@ public static OpenSearchQueryRequest pitOf(
     this.includes = includes;
     this.cursorKeepAlive = cursorKeepAlive;
     this.pitId = pitId;
+    this.disableRequestCache = CalcitePlanContext.disableRequestCache.get();
   }
 
   /** true if the request is a count aggregation request. */
@@ -185,6 +189,7 @@ public OpenSearchQueryRequest(StreamInput in, OpenSearchStorageEngine engine) th
     exprValueFactory =
         new OpenSearchExprValueFactory(
             index.getFieldOpenSearchTypes(), index.isFieldTypeTolerance());
+    this.disableRequestCache = false;
   }
 
   @Override
@@ -218,6 +223,11 @@ private OpenSearchResponse search(Function<SearchRequest, SearchResponse> search
 
       SearchRequest searchRequest =
           new SearchRequest().indices(indexName.getIndexNames()).source(this.sourceBuilder);
+      if (disableRequestCache) {
+        searchRequest.requestCache(false);
+      }
+      // LOG.info("[CACHE_DEBUG] disableRequestCache={}, searchRequest.requestCache()={}",
+      // disableRequestCache, searchRequest.requestCache());
       this.searchResponse = searchAction.apply(searchRequest);
 
       openSearchResponse =
@@ -275,6 +285,9 @@ public OpenSearchResponse searchWithPIT(Function<SearchRequest, SearchResponse>
       }
       SearchRequest searchRequest =
           new SearchRequest().indices(indexName.getIndexNames()).source(this.sourceBuilder);
+      if (disableRequestCache) {
+        searchRequest.requestCache(false);
+      }
       this.searchResponse = searchAction.apply(searchRequest);
 
       openSearchResponse =
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 7f117e7bb0b..7a73fc8acb3 100644
--- a/ppl/src/main/java/org/opensearch/sql/ppl/PPLService.java
+++ b/ppl/src/main/java/org/opensearch/sql/ppl/PPLService.java
@@ -8,10 +8,7 @@
 import static org.opensearch.sql.executor.ExecutionEngine.QueryResponse;
 import static org.opensearch.sql.executor.execution.QueryPlanFactory.NO_CONSUMER_RESPONSE_LISTENER;
 
-import java.util.ArrayList;
-import java.util.List;
 import lombok.extern.log4j.Log4j2;
-import org.antlr.v4.runtime.ParserRuleContext;
 import org.antlr.v4.runtime.tree.ParseTree;
 import org.opensearch.sql.ast.statement.Query;
 import org.opensearch.sql.ast.statement.Statement;
@@ -20,14 +17,12 @@
 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.AnalyzeResponse.QuerySegment;
 import org.opensearch.sql.executor.ExecutionEngine.ExplainResponse;
 import org.opensearch.sql.executor.QueryManager;
 import org.opensearch.sql.executor.QueryType;
 import org.opensearch.sql.executor.execution.AbstractPlan;
 import org.opensearch.sql.executor.execution.QueryPlanFactory;
 import org.opensearch.sql.ppl.antlr.PPLSyntaxParser;
-import org.opensearch.sql.ppl.antlr.parser.OpenSearchPPLParser;
 import org.opensearch.sql.ppl.domain.PPLQueryRequest;
 import org.opensearch.sql.ppl.parser.AstBuilder;
 import org.opensearch.sql.ppl.parser.AstStatementBuilder;
@@ -124,62 +119,14 @@ public void analyze(PPLQueryRequest request, ResponseListener<AnalyzeResponse> l
           QueryContext.getRequestId(),
           anonymizer.anonymizeStatement(statement));
 
-      List<QuerySegment> querySegments = extractQuerySegments(cst, queryText);
       UnresolvedPlan unresolvedPlan = ((Query) statement).getPlan();
       queryManager.submit(
-          queryExecutionFactory.createAnalyzePlan(
-              queryText, querySegments, unresolvedPlan, PPL_QUERY, listener));
+          queryExecutionFactory.createAnalyzePlan(queryText, unresolvedPlan, PPL_QUERY, listener));
     } catch (Exception e) {
       listener.onFailure(e);
     }
   }
 
-  private List<QuerySegment> extractQuerySegments(ParseTree cst, String queryText) {
-    List<QuerySegment> segments = new ArrayList<>();
-    OpenSearchPPLParser.QueryStatementContext queryStmt = findQueryStatement(cst);
-    if (queryStmt == null) {
-      return segments;
-    }
-
-    // First segment: the search/source command (pplCommands)
-    OpenSearchPPLParser.PplCommandsContext pplCommands = queryStmt.pplCommands();
-    if (pplCommands != null) {
-      segments.add(buildSegment(pplCommands, queryText));
-    }
-
-    // Remaining segments: each piped command
-    for (OpenSearchPPLParser.CommandsContext cmd : queryStmt.commands()) {
-      segments.add(buildSegment(cmd, queryText));
-    }
-    return segments;
-  }
-
-  private OpenSearchPPLParser.QueryStatementContext findQueryStatement(ParseTree tree) {
-    if (tree instanceof OpenSearchPPLParser.QueryStatementContext ctx) {
-      return ctx;
-    }
-    for (int i = 0; i < tree.getChildCount(); i++) {
-      OpenSearchPPLParser.QueryStatementContext result = findQueryStatement(tree.getChild(i));
-      if (result != null) {
-        return result;
-      }
-    }
-    return null;
-  }
-
-  private QuerySegment buildSegment(ParserRuleContext ctx, String queryText) {
-    int start = ctx.getStart().getStartIndex();
-    int stop = ctx.getStop().getStopIndex();
-    String source = queryText.substring(start, stop + 1);
-    // For wrapper rules like CommandsContext, drill into the specific child command
-    ParserRuleContext target = ctx;
-    if (ctx.getChildCount() == 1 && ctx.getChild(0) instanceof ParserRuleContext child) {
-      target = child;
-    }
-    String nodeType = target.getClass().getSimpleName().replace("Context", "");
-    return QuerySegment.builder().nodeType(nodeType).source(source).build();
-  }
-
   private AbstractPlan plan(
       PPLQueryRequest request,
       ResponseListener<QueryResponse> queryListener,
diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLTrackingTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLTrackingTest.java
deleted file mode 100644
index 8469348f399..00000000000
--- a/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLTrackingTest.java
+++ /dev/null
@@ -1,266 +0,0 @@
-/*
- * Copyright OpenSearch Contributors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-package org.opensearch.sql.ppl.calcite;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.opensearch.sql.executor.QueryType.PPL;
-
-import java.util.List;
-import org.apache.calcite.plan.Contexts;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.test.CalciteAssert;
-import org.apache.calcite.tools.Frameworks;
-import org.apache.calcite.tools.RelBuilder;
-import org.junit.Before;
-import org.junit.Test;
-import org.opensearch.sql.ast.Node;
-import org.opensearch.sql.ast.statement.Query;
-import org.opensearch.sql.calcite.CalcitePlanContext;
-import org.opensearch.sql.calcite.CalcitePlanContext.NodeIdMapping;
-import org.opensearch.sql.calcite.CalciteRelNodeVisitor;
-import org.opensearch.sql.calcite.SysLimit;
-import org.opensearch.sql.common.setting.Settings;
-import org.opensearch.sql.datasource.DataSourceService;
-import org.opensearch.sql.ppl.antlr.PPLSyntaxParser;
-import org.opensearch.sql.ppl.parser.AstBuilder;
-import org.opensearch.sql.ppl.parser.AstStatementBuilder;
-
-public class CalcitePPLTrackingTest {
-
-  private final Frameworks.ConfigBuilder config;
-  private final CalciteRelNodeVisitor planTransformer;
-  private final Settings settings;
-  private final DataSourceService dataSourceService;
-  private final PPLSyntaxParser pplParser = new PPLSyntaxParser();
-
-  public CalcitePPLTrackingTest() {
-    this.dataSourceService = mock(DataSourceService.class);
-    this.planTransformer = new CalciteRelNodeVisitor(dataSourceService);
-    this.settings = mock(Settings.class);
-    this.config =
-        Frameworks.newConfigBuilder()
-            .defaultSchema(
-                CalciteAssert.addSchema(
-                    Frameworks.createRootSchema(true),
-                    CalciteAssert.SchemaSpec.SCOTT_WITH_TEMPORAL))
-            .programs();
-  }
-
-  @Before
-  public void init() {
-    doReturn(true).when(settings).getSettingValue(Settings.Key.CALCITE_ENGINE_ENABLED);
-    doReturn(true).when(settings).getSettingValue(Settings.Key.CALCITE_SUPPORT_ALL_JOIN_TYPES);
-    doReturn(true).when(settings).getSettingValue(Settings.Key.PPL_SYNTAX_LEGACY_PREFERRED);
-    doReturn(-1).when(settings).getSettingValue(Settings.Key.PPL_JOIN_SUBSEARCH_MAXOUT);
-    doReturn(-1).when(settings).getSettingValue(Settings.Key.PPL_SUBSEARCH_MAXOUT);
-    doReturn(false).when(dataSourceService).dataSourceExists(any());
-  }
-
-  private CalcitePlanContext createContext() {
-    config.context(Contexts.of(RelBuilder.Config.DEFAULT));
-    return CalcitePlanContext.create(config.build(), SysLimit.fromSettings(settings), PPL);
-  }
-
-  private Node plan(String query) {
-    final AstStatementBuilder builder =
-        new AstStatementBuilder(
-            new AstBuilder(query, settings),
-            AstStatementBuilder.StatementBuilderContext.builder().build());
-    return builder.visit(pplParser.parse(query));
-  }
-
-  private RelNode getRelNode(String ppl, CalcitePlanContext context) {
-    Query query = (Query) plan(ppl);
-    planTransformer.analyze(query.getPlan(), context);
-    return context.relBuilder.build();
-  }
-
-  @Test
-  public void testTrackingProducesSameLogicalPlanAsNonTracking() {
-    String ppl = "source=EMP | eval a = 1 | fields EMPNO, a";
-
-    CalcitePlanContext withoutTracking = createContext();
-    RelNode expected = getRelNode(ppl, withoutTracking);
-
-    CalcitePlanContext withTracking = createContext();
-    withTracking.setTrackingEnabled(true);
-    RelNode actual = getRelNode(ppl, withTracking);
-
-    assertEquals(expected.explain().replace("\r\n", "\n"), actual.explain().replace("\r\n", "\n"));
-  }
-
-  @Test
-  public void testTrackingDisabledProducesNoMappings() {
-    String ppl = "source=EMP | eval a = 1";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(false);
-    getRelNode(ppl, context);
-
-    assertTrue(context.getNodeIdMappings().isEmpty());
-  }
-
-  @Test
-  public void testTrackingEvalRecordsMappings() {
-    String ppl = "source=EMP | eval a = 1";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(true);
-    getRelNode(ppl, context);
-
-    List<NodeIdMapping> mappings = context.getNodeIdMappings();
-    assertFalse(mappings.isEmpty());
-
-    List<String> astTypes = mappings.stream().map(NodeIdMapping::astNodeType).toList();
-    assertTrue("Should contain Relation mapping", astTypes.contains("Relation"));
-    assertTrue("Should contain Eval mapping", astTypes.contains("Eval"));
-  }
-
-  @Test
-  public void testTrackingFilterRecordsMappings() {
-    String ppl = "source=EMP | where SAL > 1000";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(true);
-    getRelNode(ppl, context);
-
-    List<NodeIdMapping> mappings = context.getNodeIdMappings();
-    List<String> astTypes = mappings.stream().map(NodeIdMapping::astNodeType).toList();
-    assertTrue("Should contain Relation mapping", astTypes.contains("Relation"));
-    assertTrue("Should contain Filter mapping", astTypes.contains("Filter"));
-  }
-
-  @Test
-  public void testTrackingSortRecordsMappings() {
-    String ppl = "source=EMP | sort SAL";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(true);
-    getRelNode(ppl, context);
-
-    List<NodeIdMapping> mappings = context.getNodeIdMappings();
-    List<String> astTypes = mappings.stream().map(NodeIdMapping::astNodeType).toList();
-    assertTrue("Should contain Relation mapping", astTypes.contains("Relation"));
-    assertTrue("Should contain Sort mapping", astTypes.contains("Sort"));
-  }
-
-  @Test
-  public void testTrackingMultipleCommandsRecordsMappings() {
-    String ppl = "source=EMP | where SAL > 1000 | eval bonus = SAL * 0.1 | sort SAL";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(true);
-    getRelNode(ppl, context);
-
-    List<NodeIdMapping> mappings = context.getNodeIdMappings();
-    List<String> astTypes = mappings.stream().map(NodeIdMapping::astNodeType).toList();
-    assertTrue("Should contain Relation mapping", astTypes.contains("Relation"));
-    assertTrue("Should contain Filter mapping", astTypes.contains("Filter"));
-    assertTrue("Should contain Eval mapping", astTypes.contains("Eval"));
-    assertTrue("Should contain Sort mapping", astTypes.contains("Sort"));
-  }
-
-  @Test
-  public void testTrackingMappingsHaveNonEmptyRelNodeIds() {
-    String ppl = "source=EMP | eval a = 1";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(true);
-    getRelNode(ppl, context);
-
-    for (NodeIdMapping mapping : context.getNodeIdMappings()) {
-      assertFalse(
-          "Mapping for " + mapping.astNodeType() + " should have non-empty RelNode IDs",
-          mapping.relNodeIds().isEmpty());
-    }
-  }
-
-  @Test
-  public void testTrackingProjectRecordsMappings() {
-    String ppl = "source=EMP | fields EMPNO, ENAME";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(true);
-    getRelNode(ppl, context);
-
-    List<NodeIdMapping> mappings = context.getNodeIdMappings();
-    List<String> astTypes = mappings.stream().map(NodeIdMapping::astNodeType).toList();
-    assertTrue("Should contain Relation mapping", astTypes.contains("Relation"));
-    assertTrue("Should contain Project mapping", astTypes.contains("Project"));
-  }
-
-  @Test
-  public void testTrackingAggregationRecordsMappings() {
-    String ppl = "source=EMP | stats count() by DEPTNO";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(true);
-    getRelNode(ppl, context);
-
-    List<NodeIdMapping> mappings = context.getNodeIdMappings();
-    List<String> astTypes = mappings.stream().map(NodeIdMapping::astNodeType).toList();
-    assertTrue("Should contain Relation mapping", astTypes.contains("Relation"));
-    assertTrue("Should contain Aggregation mapping", astTypes.contains("Aggregation"));
-  }
-
-  @Test
-  public void testTrackingMultipleCommandsProducesSameLogicalPlan() {
-    String ppl = "source=EMP | where SAL > 1000 | eval bonus = SAL * 0.1 | sort SAL | head 10";
-
-    CalcitePlanContext withoutTracking = createContext();
-    RelNode expected = getRelNode(ppl, withoutTracking);
-
-    CalcitePlanContext withTracking = createContext();
-    withTracking.setTrackingEnabled(true);
-    RelNode actual = getRelNode(ppl, withTracking);
-
-    assertEquals(expected.explain().replace("\r\n", "\n"), actual.explain().replace("\r\n", "\n"));
-  }
-
-  @Test
-  public void testVisitChildrenCapturesSubtreeContribution() {
-    // visitChildren records a child's SUBTREE contribution (all RelNodes produced
-    // by that child and its descendants). For a multi-command pipeline, the mapping
-    // for Filter should include RelNodes from its own subtree (Relation + Filter itself).
-    String ppl = "source=EMP | where SAL > 1000 | eval bonus = SAL * 0.1";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(true);
-    getRelNode(ppl, context);
-
-    List<NodeIdMapping> mappings = context.getNodeIdMappings();
-
-    // Relation is a leaf — should produce exactly 1 RelNode (the scan)
-    NodeIdMapping relationMapping =
-        mappings.stream().filter(m -> m.astNodeType().equals("Relation")).findFirst().orElseThrow();
-    assertFalse(
-        "Relation (leaf) should produce at least one RelNode",
-        relationMapping.relNodeIds().isEmpty());
-
-    // Filter's subtree includes Relation beneath it, so visitChildren should
-    // capture more RelNode IDs for Filter than for Relation alone.
-    NodeIdMapping filterMapping =
-        mappings.stream().filter(m -> m.astNodeType().equals("Filter")).findFirst().orElseThrow();
-    assertTrue(
-        "Filter subtree should produce more RelNodes than Relation alone",
-        filterMapping.relNodeIds().size() > relationMapping.relNodeIds().size());
-  }
-
-  @Test
-  public void testVisitChildrenRecordsAllChildrenSeparately() {
-    // visitChildren iterates over node.getChild() and records each one.
-    // For a pipeline with multiple commands, each command gets its own mapping entry.
-    String ppl = "source=EMP | where SAL > 1000 | sort SAL | head 5";
-    CalcitePlanContext context = createContext();
-    context.setTrackingEnabled(true);
-    getRelNode(ppl, context);
-
-    List<NodeIdMapping> mappings = context.getNodeIdMappings();
-    List<String> astTypes = mappings.stream().map(NodeIdMapping::astNodeType).toList();
-
-    // Each command in the pipeline should have a separate mapping entry
-    assertTrue("Should record Relation", astTypes.contains("Relation"));
-    assertTrue("Should record Filter", astTypes.contains("Filter"));
-    assertTrue("Should record Sort", astTypes.contains("Sort"));
-    assertTrue("Should record Head", astTypes.contains("Head"));
-  }
-}